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