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 get_logger(): 

99 """Return logger.""" 

100 logger = logging.getLogger() 

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

102 if logger_config: 

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

104 logger.setLevel(level=level_name) 

105 if not log_handler_exists( 

106 logger=logger, 

107 handler_type=logging.FileHandler, 

108 filename=logger_config["filename"], 

109 ): 

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

111 file_handler.setFormatter( 

112 logging.Formatter( 

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

114 ), 

115 ) 

116 logger.addHandler(file_handler) 

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

118 console_handler = logging.StreamHandler(sys.stdout) 

119 console_handler.setFormatter( 

120 ColorfulConsoleFormatter( 

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

122 ), 

123 ) 

124 logger.addHandler(console_handler) 

125 return logger 

126 

127 

128LOGGER = get_logger()