Coverage for src/__init__.py: 100%
114 statements
« prev ^ index » next coverage.py v7.17.0, created at 2026-10-26 21:28 +0000
1"""Root module."""
3__author__ = "Mandar Patil (mandarons@pm.me)"
5import logging
6import os
7import sys
8import warnings
9from logging.handlers import RotatingFileHandler
11from ruamel.yaml import YAML
13DEFAULT_ROOT_DESTINATION = "./icloud"
14DEFAULT_DRIVE_DESTINATION = "drive"
15DEFAULT_PHOTOS_DESTINATION = "photos"
16DEFAULT_RETRY_LOGIN_INTERVAL_SEC = 600 # 10 minutes
17DEFAULT_SYNC_INTERVAL_SEC = 1800 # 30 minutes
18DEFAULT_REQUEST_TIMEOUT_SEC = 30 # 30 seconds
19# Share of a destination that obsolete-cleanup may delete in one run before
20# it refuses and reports instead. A sync does not normally make a quarter of
21# a library obsolete; when it appears to, a tracking bug is likelier than a
22# real mass deletion. 0 disables the limit.
23DEFAULT_OBSOLETE_DELETE_LIMIT_PERCENT = 25
24DEFAULT_ENUMERATION_CHUNK_SIZE = 1000 # photos buffered per streaming chunk
25# Oldest a buffered photo download may get before the buffer is drained
26# early. Download URLs expire ~30-40 minutes after the listing that returned
27# them, and on a mostly backed-up library the few photos to fetch trickle in
28# across the whole walk. This keeps their wait short; anything that still
29# expires is recovered by the expired-URL refresh.
30BUFFERED_DOWNLOAD_MAX_AGE_SEC = 600 # 10 minutes
31DEFAULT_CONFIG_FILE_NAME = "config.yaml"
32ENV_ICLOUD_PASSWORD_KEY = "ENV_ICLOUD_PASSWORD"
33ENV_CONFIG_FILE_PATH_KEY = "ENV_CONFIG_FILE_PATH"
34DEFAULT_LOGGER_LEVEL = "info"
35DEFAULT_LOG_FILE_NAME = "icloud.log"
36# Rotation bounds for the file handler. A busy library logs one line per
37# file *considered* each cycle, so an unbounded handler grows without
38# limit -- multi-GB logs on large accounts. 50 MB x 3 backups keeps a
39# useful window at a bounded 200 MB worst case.
40DEFAULT_LOG_MAX_BYTES = 50 * 1024 * 1024
41DEFAULT_LOG_BACKUP_COUNT = 3
42DEFAULT_CONFIG_FILE_PATH = os.path.join(os.path.dirname(os.path.dirname(__file__)), DEFAULT_CONFIG_FILE_NAME)
43# Operator-overridable via ICLOUD_DOCKER_CONFIG_DIR. Default ``/config`` is the
44# in-container mount point users bind their config volume to. The override
45# lets the test suite run on hosts where ``/config`` isn't writable (macOS,
46# sandboxes).
47_CONFIG_DIR = os.environ.get("ICLOUD_DOCKER_CONFIG_DIR", "/config")
48DEFAULT_COOKIE_DIRECTORY = os.path.join(_CONFIG_DIR, "session_data")
50warnings.filterwarnings("ignore", category=DeprecationWarning)
53def read_config(config_path=DEFAULT_CONFIG_FILE_PATH):
54 """Read config file."""
55 if not (config_path and os.path.exists(config_path)):
56 print(f"Config file not found at {config_path}.")
57 return None
58 with open(file=config_path, encoding="utf-8") as config_file:
59 config = YAML().load(config_file)
60 config["app"]["credentials"]["username"] = (
61 config["app"]["credentials"]["username"].strip() if config["app"]["credentials"]["username"] is not None else ""
62 )
63 return config
66def get_logger_config(config):
67 """Get logger config."""
68 logger_config = {}
69 # ``read_config`` returns None when config.yaml is missing, and a partial
70 # file may have no ``app`` section. Both reached ``config["app"]`` and
71 # raised -- from module scope, so the container could not start at all.
72 # Shape-check every level, not just presence: a parseable but wrong YAML
73 # (``app: 123``, ``logger: info``) otherwise raises TypeError here at
74 # import time -- the same restart loop this guard exists to prevent.
75 app = config.get("app") if isinstance(config, dict) else None
76 config_app_logger = app.get("logger") if isinstance(app, dict) else None
77 if not isinstance(config_app_logger, dict):
78 return None
79 logger_config["level"] = (
80 config_app_logger["level"].strip().lower() if "level" in config_app_logger else DEFAULT_LOGGER_LEVEL
81 )
82 logger_config["filename"] = (
83 config_app_logger["filename"].strip().lower() if "filename" in config_app_logger else DEFAULT_LOG_FILE_NAME
84 )
85 logger_config["max_bytes"] = _int_setting(config_app_logger, "max_bytes", DEFAULT_LOG_MAX_BYTES)
86 logger_config["backup_count"] = _int_setting(config_app_logger, "backup_count", DEFAULT_LOG_BACKUP_COUNT)
87 return logger_config
90def _int_setting(section, key, default):
91 """Read an integer logger setting, falling back to ``default`` on junk.
93 This runs while logging is being configured -- at import, before any
94 handler exists -- so a bare ``int()`` on a typo like ``max_bytes: 50MB``
95 raised out of module import and, under ``restart: unless-stopped``,
96 became a restart loop that never said which key was wrong. Say so on
97 stdout (no logger yet) and carry on with the default.
98 """
99 value = section.get(key, default)
100 try:
101 return int(value)
102 except (TypeError, ValueError):
103 print(f"app.logger.{key} must be a whole number, got {value!r}; using {default}.")
104 return default
107def log_handler_exists(logger, handler_type, **kwargs):
108 """Check for existing log handler."""
109 for handler in logger.handlers:
110 if isinstance(handler, handler_type):
111 if handler_type is logging.FileHandler:
112 if handler.baseFilename.endswith(kwargs["filename"]):
113 return True
114 elif handler_type is logging.StreamHandler:
115 if handler.stream is kwargs["stream"]:
116 return True
117 return False
120class ColorfulConsoleFormatter(logging.Formatter):
121 """Console formatter for log messages."""
123 grey = "\x1b[38;21m"
124 blue = "\x1b[38;5;39m"
125 yellow = "\x1b[38;5;226m"
126 red = "\x1b[38;5;196m"
127 bold_red = "\x1b[31;1m"
128 reset = "\x1b[0m"
130 def __init__(self, fmt):
131 """Construct with defaults."""
132 super().__init__()
133 self.fmt = fmt
134 self.formats = {
135 logging.DEBUG: self.grey + self.fmt + self.reset,
136 logging.INFO: self.blue + self.fmt + self.reset,
137 logging.WARNING: self.yellow + self.fmt + self.reset,
138 logging.ERROR: self.red + self.fmt + self.reset,
139 logging.CRITICAL: self.bold_red + self.fmt + self.reset,
140 }
142 def format(self, record):
143 """Format the record."""
144 log_fmt = self.formats.get(record.levelno)
145 formatter = logging.Formatter(log_fmt)
146 return formatter.format(record)
149def configure_icloudpy_logging():
150 """Configure icloudpy logging to match app logging level."""
151 logger_config = get_logger_config(config=read_config(config_path=os.environ.get(ENV_CONFIG_FILE_PATH_KEY, DEFAULT_CONFIG_FILE_PATH)))
152 if logger_config:
153 level_name = logging.getLevelName(level=logger_config["level"].upper())
155 # Configure icloudpy loggers to use the same level and enable propagation
156 icloudpy_loggers = [
157 logging.getLogger("icloudpy"),
158 logging.getLogger("icloudpy.base"),
159 logging.getLogger("icloudpy.services"),
160 logging.getLogger("icloudpy.services.photos"),
161 ]
162 for icloudpy_logger in icloudpy_loggers:
163 icloudpy_logger.setLevel(level=level_name)
164 # Enable propagation so messages go to root logger handlers
165 icloudpy_logger.propagate = True
166 # Remove any existing handlers to avoid duplicates
167 icloudpy_logger.handlers.clear()
170def get_logger():
171 """Return logger."""
172 logger = logging.getLogger()
173 # This runs at import time, so anything raised here stops the container
174 # before it can report why -- and `restart: unless-stopped` turns that
175 # into a restart loop. A config too broken to parse is the sync loop's
176 # problem to report; here it just means default logging.
177 try:
178 config = read_config(config_path=os.environ.get(ENV_CONFIG_FILE_PATH_KEY, DEFAULT_CONFIG_FILE_PATH))
179 except Exception: # noqa: BLE001 -- import-time: never crash on a bad config
180 config = None
181 logger_config = get_logger_config(config=config)
182 if logger_config:
183 level_name = logging.getLevelName(level=logger_config["level"].upper())
184 logger.setLevel(level=level_name)
186 # Create handlers once and add them to root logger
187 file_handler = None
188 console_handler = None
190 if not log_handler_exists(
191 logger=logger,
192 handler_type=logging.FileHandler,
193 filename=logger_config["filename"],
194 ):
195 # Rotating rather than plain FileHandler: see
196 # DEFAULT_LOG_MAX_BYTES. max_bytes=0 disables rotation for
197 # operators who hand the file to an external logrotate.
198 file_handler = RotatingFileHandler(
199 logger_config["filename"],
200 maxBytes=logger_config["max_bytes"],
201 backupCount=logger_config["backup_count"],
202 )
203 file_handler.setFormatter(
204 logging.Formatter(
205 "%(asctime)s :: %(levelname)s :: %(name)s :: %(filename)s :: %(lineno)d :: %(message)s",
206 ),
207 )
208 logger.addHandler(file_handler)
210 if not log_handler_exists(logger=logger, handler_type=logging.StreamHandler, stream=sys.stdout):
211 console_handler = logging.StreamHandler(sys.stdout)
212 console_handler.setFormatter(
213 ColorfulConsoleFormatter(
214 "%(asctime)s :: %(levelname)s :: %(name)s :: %(filename)s :: %(lineno)d :: %(message)s",
215 ),
216 )
217 logger.addHandler(console_handler)
219 # Configure icloudpy loggers to use the same level and enable propagation
220 icloudpy_loggers = [
221 logging.getLogger("icloudpy"),
222 logging.getLogger("icloudpy.base"),
223 logging.getLogger("icloudpy.services"),
224 logging.getLogger("icloudpy.services.photos"),
225 ]
226 for icloudpy_logger in icloudpy_loggers:
227 icloudpy_logger.setLevel(level=level_name)
228 # Enable propagation so messages go to root logger handlers
229 icloudpy_logger.propagate = True
230 # Remove any existing handlers to avoid duplicates
231 icloudpy_logger.handlers.clear()
232 return logger
235LOGGER = get_logger()