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_CONFIG_FILE_NAME = "config.yaml" 

19ENV_ICLOUD_PASSWORD_KEY = "ENV_ICLOUD_PASSWORD" 

20ENV_CONFIG_FILE_PATH_KEY = "ENV_CONFIG_FILE_PATH" 

21DEFAULT_LOGGER_LEVEL = "info" 

22DEFAULT_LOG_FILE_NAME = "icloud.log" 

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

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

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

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

27# sandboxes). 

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

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

30 

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

32 

33 

34def read_config(config_path=DEFAULT_CONFIG_FILE_PATH): 

35 """Read config file.""" 

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

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

38 return None 

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

40 config = YAML().load(config_file) 

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

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

43 ) 

44 return config 

45 

46 

47def get_logger_config(config): 

48 """Get logger config.""" 

49 logger_config = {} 

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

51 return None 

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

53 logger_config["level"] = ( 

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

55 ) 

56 logger_config["filename"] = ( 

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

58 ) 

59 return logger_config 

60 

61 

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

63 """Check for existing log handler.""" 

64 for handler in logger.handlers: 

65 if isinstance(handler, handler_type): 

66 if handler_type is logging.FileHandler: 

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

68 return True 

69 elif handler_type is logging.StreamHandler: 

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

71 return True 

72 return False 

73 

74 

75class ColorfulConsoleFormatter(logging.Formatter): 

76 """Console formatter for log messages.""" 

77 

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

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

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

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

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

83 reset = "\x1b[0m" 

84 

85 def __init__(self, fmt): 

86 """Construct with defaults.""" 

87 super().__init__() 

88 self.fmt = fmt 

89 self.formats = { 

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

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

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

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

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

95 } 

96 

97 def format(self, record): 

98 """Format the record.""" 

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

100 formatter = logging.Formatter(log_fmt) 

101 return formatter.format(record) 

102 

103 

104def configure_icloudpy_logging(): 

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

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

107 if logger_config: 

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

109 

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

111 icloudpy_loggers = [ 

112 logging.getLogger("icloudpy"), 

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

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

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

116 ] 

117 for icloudpy_logger in icloudpy_loggers: 

118 icloudpy_logger.setLevel(level=level_name) 

119 # Enable propagation so messages go to root logger handlers 

120 icloudpy_logger.propagate = True 

121 # Remove any existing handlers to avoid duplicates 

122 icloudpy_logger.handlers.clear() 

123 

124 

125def get_logger(): 

126 """Return logger.""" 

127 logger = logging.getLogger() 

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

129 if logger_config: 

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

131 logger.setLevel(level=level_name) 

132 

133 # Create handlers once and add them to root logger 

134 file_handler = None 

135 console_handler = None 

136 

137 if not log_handler_exists( 

138 logger=logger, 

139 handler_type=logging.FileHandler, 

140 filename=logger_config["filename"], 

141 ): 

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

143 file_handler.setFormatter( 

144 logging.Formatter( 

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

146 ), 

147 ) 

148 logger.addHandler(file_handler) 

149 

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

151 console_handler = logging.StreamHandler(sys.stdout) 

152 console_handler.setFormatter( 

153 ColorfulConsoleFormatter( 

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

155 ), 

156 ) 

157 logger.addHandler(console_handler) 

158 

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

160 icloudpy_loggers = [ 

161 logging.getLogger("icloudpy"), 

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

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

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

165 ] 

166 for icloudpy_logger in icloudpy_loggers: 

167 icloudpy_logger.setLevel(level=level_name) 

168 # Enable propagation so messages go to root logger handlers 

169 icloudpy_logger.propagate = True 

170 # Remove any existing handlers to avoid duplicates 

171 icloudpy_logger.handlers.clear() 

172 return logger 

173 

174 

175LOGGER = get_logger()