Coverage for src/__init__.py: 100%

1"""Root module.""" 

2 

3__author__ = "Mandar Patil (mandarons@pm.me)" 

4 

5import logging 

6import os 

7import sys 

8import warnings 

9 

10from ruamel.yaml import YAML 

11 

12DEFAULT_ROOT_DESTINATION = "./icloud" 

13DEFAULT_DRIVE_DESTINATION = "drive" 

14DEFAULT_PHOTOS_DESTINATION = "photos" 

15DEFAULT_RETRY_LOGIN_INTERVAL_SEC = 600 # 10 minutes 

16DEFAULT_SYNC_INTERVAL_SEC = 1800 # 30 minutes 

17DEFAULT_REQUEST_TIMEOUT_SEC = 30 # 30 seconds 

18DEFAULT_ENUMERATION_CHUNK_SIZE = 1000 # photos buffered per streaming chunk 

19DEFAULT_CONFIG_FILE_NAME = "config.yaml" 

20ENV_ICLOUD_PASSWORD_KEY = "ENV_ICLOUD_PASSWORD" 

21ENV_CONFIG_FILE_PATH_KEY = "ENV_CONFIG_FILE_PATH" 

22DEFAULT_LOGGER_LEVEL = "info" 

23DEFAULT_LOG_FILE_NAME = "icloud.log" 

24DEFAULT_CONFIG_FILE_PATH = os.path.join(os.path.dirname(os.path.dirname(__file__)), DEFAULT_CONFIG_FILE_NAME) 

25# Operator-overridable via ICLOUD_DOCKER_CONFIG_DIR. Default ``/config`` is the 

26# in-container mount point users bind their config volume to. The override 

27# lets the test suite run on hosts where ``/config`` isn't writable (macOS, 

28# sandboxes). 

29_CONFIG_DIR = os.environ.get("ICLOUD_DOCKER_CONFIG_DIR", "/config") 

30DEFAULT_COOKIE_DIRECTORY = os.path.join(_CONFIG_DIR, "session_data") 

31 

32warnings.filterwarnings("ignore", category=DeprecationWarning) 

33 

34 

35def read_config(config_path=DEFAULT_CONFIG_FILE_PATH): 

36 """Read config file.""" 

37 if not (config_path and os.path.exists(config_path)): 

38 print(f"Config file not found at {config_path}.") 

39 return None 

40 with open(file=config_path, encoding="utf-8") as config_file: 

41 config = YAML().load(config_file) 

42 config["app"]["credentials"]["username"] = ( 

43 config["app"]["credentials"]["username"].strip() if config["app"]["credentials"]["username"] is not None else "" 

44 ) 

45 return config 

46 

47 

48def get_logger_config(config): 

49 """Get logger config.""" 

50 logger_config = {} 

51 if "logger" not in config["app"]: 

52 return None 

53 config_app_logger = config["app"]["logger"] 

54 logger_config["level"] = ( 

55 config_app_logger["level"].strip().lower() if "level" in config_app_logger else DEFAULT_LOGGER_LEVEL 

56 ) 

57 logger_config["filename"] = ( 

58 config_app_logger["filename"].strip().lower() if "filename" in config_app_logger else DEFAULT_LOG_FILE_NAME 

59 ) 

60 return logger_config 

61 

62 

63def log_handler_exists(logger, handler_type, **kwargs): 

64 """Check for existing log handler.""" 

65 for handler in logger.handlers: 

66 if isinstance(handler, handler_type): 

67 if handler_type is logging.FileHandler: 

68 if handler.baseFilename.endswith(kwargs["filename"]): 

69 return True 

70 elif handler_type is logging.StreamHandler: 

71 if handler.stream is kwargs["stream"]: 

72 return True 

73 return False 

74 

75 

76class ColorfulConsoleFormatter(logging.Formatter): 

77 """Console formatter for log messages.""" 

78 

79 grey = "\x1b[38;21m" 

80 blue = "\x1b[38;5;39m" 

81 yellow = "\x1b[38;5;226m" 

82 red = "\x1b[38;5;196m" 

83 bold_red = "\x1b[31;1m" 

84 reset = "\x1b[0m" 

85 

86 def __init__(self, fmt): 

87 """Construct with defaults.""" 

88 super().__init__() 

89 self.fmt = fmt 

90 self.formats = { 

91 logging.DEBUG: self.grey + self.fmt + self.reset, 

92 logging.INFO: self.blue + self.fmt + self.reset, 

93 logging.WARNING: self.yellow + self.fmt + self.reset, 

94 logging.ERROR: self.red + self.fmt + self.reset, 

95 logging.CRITICAL: self.bold_red + self.fmt + self.reset, 

96 } 

97 

98 def format(self, record): 

99 """Format the record.""" 

100 log_fmt = self.formats.get(record.levelno) 

101 formatter = logging.Formatter(log_fmt) 

102 return formatter.format(record) 

103 

104 

105def configure_icloudpy_logging(): 

106 """Configure icloudpy logging to match app logging level.""" 

107 logger_config = get_logger_config(config=read_config(config_path=os.environ.get(ENV_CONFIG_FILE_PATH_KEY, DEFAULT_CONFIG_FILE_PATH))) 

108 if logger_config: 

109 level_name = logging.getLevelName(level=logger_config["level"].upper()) 

110 

111 # Configure icloudpy loggers to use the same level and enable propagation 

112 icloudpy_loggers = [ 

113 logging.getLogger("icloudpy"), 

114 logging.getLogger("icloudpy.base"), 

115 logging.getLogger("icloudpy.services"), 

116 logging.getLogger("icloudpy.services.photos"), 

117 ] 

118 for icloudpy_logger in icloudpy_loggers: 

119 icloudpy_logger.setLevel(level=level_name) 

120 # Enable propagation so messages go to root logger handlers 

121 icloudpy_logger.propagate = True 

122 # Remove any existing handlers to avoid duplicates 

123 icloudpy_logger.handlers.clear() 

124 

125 

126def get_logger(): 

127 """Return logger.""" 

128 logger = logging.getLogger() 

129 logger_config = get_logger_config(config=read_config(config_path=os.environ.get(ENV_CONFIG_FILE_PATH_KEY, DEFAULT_CONFIG_FILE_PATH))) 

130 if logger_config: 

131 level_name = logging.getLevelName(level=logger_config["level"].upper()) 

132 logger.setLevel(level=level_name) 

133 

134 # Create handlers once and add them to root logger 

135 file_handler = None 

136 console_handler = None 

137 

138 if not log_handler_exists( 

139 logger=logger, 

140 handler_type=logging.FileHandler, 

141 filename=logger_config["filename"], 

142 ): 

143 file_handler = logging.FileHandler(logger_config["filename"]) 

144 file_handler.setFormatter( 

145 logging.Formatter( 

146 "%(asctime)s :: %(levelname)s :: %(name)s :: %(filename)s :: %(lineno)d :: %(message)s", 

147 ), 

148 ) 

149 logger.addHandler(file_handler) 

150 

151 if not log_handler_exists(logger=logger, handler_type=logging.StreamHandler, stream=sys.stdout): 

152 console_handler = logging.StreamHandler(sys.stdout) 

153 console_handler.setFormatter( 

154 ColorfulConsoleFormatter( 

155 "%(asctime)s :: %(levelname)s :: %(name)s :: %(filename)s :: %(lineno)d :: %(message)s", 

156 ), 

157 ) 

158 logger.addHandler(console_handler) 

159 

160 # Configure icloudpy loggers to use the same level and enable propagation 

161 icloudpy_loggers = [ 

162 logging.getLogger("icloudpy"), 

163 logging.getLogger("icloudpy.base"), 

164 logging.getLogger("icloudpy.services"), 

165 logging.getLogger("icloudpy.services.photos"), 

166 ] 

167 for icloudpy_logger in icloudpy_loggers: 

168 icloudpy_logger.setLevel(level=level_name) 

169 # Enable propagation so messages go to root logger handlers 

170 icloudpy_logger.propagate = True 

171 # Remove any existing handlers to avoid duplicates 

172 icloudpy_logger.handlers.clear() 

173 return logger 

174 

175 

176LOGGER = get_logger()