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

18ENV_ICLOUD_PASSWORD_KEY = "ENV_ICLOUD_PASSWORD" 

19ENV_CONFIG_FILE_PATH_KEY = "ENV_CONFIG_FILE_PATH" 

20DEFAULT_LOGGER_LEVEL = "info" 

21DEFAULT_LOG_FILE_NAME = "icloud.log" 

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

23DEFAULT_COOKIE_DIRECTORY = "/config/session_data" 

24 

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

26 

27 

28def read_config(config_path=DEFAULT_CONFIG_FILE_PATH): 

29 """Read config file.""" 

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

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

32 return None 

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

34 config = YAML().load(config_file) 

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

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

37 ) 

38 return config 

39 

40 

41def get_logger_config(config): 

42 """Get logger config.""" 

43 logger_config = {} 

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

45 return None 

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

47 logger_config["level"] = ( 

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

49 ) 

50 logger_config["filename"] = ( 

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

52 ) 

53 return logger_config 

54 

55 

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

57 """Check for existing log handler.""" 

58 for handler in logger.handlers: 

59 if isinstance(handler, handler_type): 

60 if handler_type is logging.FileHandler: 

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

62 return True 

63 elif handler_type is logging.StreamHandler: 

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

65 return True 

66 return False 

67 

68 

69class ColorfulConsoleFormatter(logging.Formatter): 

70 """Console formatter for log messages.""" 

71 

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

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

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

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

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

77 reset = "\x1b[0m" 

78 

79 def __init__(self, fmt): 

80 """Construct with defaults.""" 

81 super().__init__() 

82 self.fmt = fmt 

83 self.formats = { 

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

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

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

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

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

89 } 

90 

91 def format(self, record): 

92 """Format the record.""" 

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

94 formatter = logging.Formatter(log_fmt) 

95 return formatter.format(record) 

96 

97 

98def configure_icloudpy_logging(): 

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

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

101 if logger_config: 

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

103 

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

105 icloudpy_loggers = [ 

106 logging.getLogger("icloudpy"), 

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

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

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

110 ] 

111 for icloudpy_logger in icloudpy_loggers: 

112 icloudpy_logger.setLevel(level=level_name) 

113 # Enable propagation so messages go to root logger handlers 

114 icloudpy_logger.propagate = True 

115 # Remove any existing handlers to avoid duplicates 

116 icloudpy_logger.handlers.clear() 

117 

118 

119def get_logger(): 

120 """Return logger.""" 

121 logger = logging.getLogger() 

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

123 if logger_config: 

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

125 logger.setLevel(level=level_name) 

126 

127 # Create handlers once and add them to root logger 

128 file_handler = None 

129 console_handler = None 

130 

131 if not log_handler_exists( 

132 logger=logger, 

133 handler_type=logging.FileHandler, 

134 filename=logger_config["filename"], 

135 ): 

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

137 file_handler.setFormatter( 

138 logging.Formatter( 

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

140 ), 

141 ) 

142 logger.addHandler(file_handler) 

143 

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

145 console_handler = logging.StreamHandler(sys.stdout) 

146 console_handler.setFormatter( 

147 ColorfulConsoleFormatter( 

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

149 ), 

150 ) 

151 logger.addHandler(console_handler) 

152 

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

154 icloudpy_loggers = [ 

155 logging.getLogger("icloudpy"), 

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

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

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

159 ] 

160 for icloudpy_logger in icloudpy_loggers: 

161 icloudpy_logger.setLevel(level=level_name) 

162 # Enable propagation so messages go to root logger handlers 

163 icloudpy_logger.propagate = True 

164 # Remove any existing handlers to avoid duplicates 

165 icloudpy_logger.handlers.clear() 

166 return logger 

167 

168 

169LOGGER = get_logger()