Coverage for src/config_parser.py: 100%
410 statements
« prev ^ index » next coverage.py v7.17.0, created at 2026-10-26 21:28 +0000
1"""Config file parser.
3This module provides high-level configuration retrieval functions.
4Low-level utilities are in config_utils.py, logging in config_logging.py,
5and filesystem operations in filesystem_utils.py per SRP.
6"""
8__author__ = "Mandar Patil (mandarons@pm.me)"
10import multiprocessing
11from typing import Any
13from icloudpy.services.photos import PhotoAsset
15from src import (
16 DEFAULT_DRIVE_DESTINATION,
17 DEFAULT_ENUMERATION_CHUNK_SIZE,
18 DEFAULT_OBSOLETE_DELETE_LIMIT_PERCENT,
19 DEFAULT_PHOTOS_DESTINATION,
20 DEFAULT_REQUEST_TIMEOUT_SEC,
21 DEFAULT_RETRY_LOGIN_INTERVAL_SEC,
22 DEFAULT_ROOT_DESTINATION,
23 DEFAULT_SYNC_INTERVAL_SEC,
24 configure_icloudpy_logging,
25 get_logger,
26)
27from src.config_logging import (
28 log_config_debug,
29 log_config_error,
30 log_config_found_info,
31 log_config_not_found_warning,
32 log_invalid_config_value,
33)
34from src.config_utils import (
35 config_path_to_string,
36 get_config_value,
37 get_config_value_or_default,
38 get_config_value_or_none,
39 traverse_config_path,
40)
41from src.filesystem_utils import ensure_directory_exists, join_and_ensure_path
43# Configure icloudpy logging immediately after import
44configure_icloudpy_logging()
46LOGGER = get_logger()
48# Cache for config values to prevent repeated warnings
49_config_warning_cache = set()
52def _log_config_warning_once(config_path: list, message: str) -> None:
53 """Log a configuration warning only once for the given config path.
55 Args:
56 config_path: Configuration path as list
57 message: Warning message to log
58 """
59 config_path_key = config_path_to_string(config_path)
60 if config_path_key not in _config_warning_cache:
61 _config_warning_cache.add(config_path_key)
62 log_config_not_found_warning(config_path, message)
65def clear_config_warning_cache() -> None:
66 """Clear the configuration warning cache.
68 This function is primarily intended for testing purposes to ensure
69 clean test isolation.
70 """
71 _config_warning_cache.clear()
74# =============================================================================
75# String Processing Functions
76# =============================================================================
79def validate_and_strip_username(username: str, config_path: list[str]) -> str | None:
80 """Validate and strip username string.
82 Args:
83 username: Raw username string from config
84 config_path: Config path for error logging
86 Returns:
87 Stripped username if valid, None if empty
88 """
89 username = username.strip()
90 if len(username) == 0:
91 log_config_error(config_path, "username is empty")
92 return None
93 return username
96# =============================================================================
97# Credential Configuration Functions
98# =============================================================================
101def get_username(config: dict) -> str | None:
102 """Get username from config.
104 Args:
105 config: Configuration dictionary
107 Returns:
108 Username string if found and valid, None otherwise
109 """
110 config_path = ["app", "credentials", "username"]
111 if not traverse_config_path(config=config, config_path=config_path):
112 log_config_error(config_path, "username is missing. Please set the username.")
113 return None
115 username = get_config_value(config=config, config_path=config_path)
116 return validate_and_strip_username(username, config_path)
119def get_retry_login_interval(config: dict, log_messages: bool = True) -> int:
120 """Return retry login interval from config.
122 Args:
123 config: Configuration dictionary
124 log_messages: Whether to log informational messages (default: True).
125 Callers that reuse this interval for something other than a
126 login retry pass False, so the log does not claim a login is
127 being retried when it is not.
129 Returns:
130 Retry login interval in seconds
131 """
132 config_path = ["app", "credentials", "retry_login_interval"]
134 if not traverse_config_path(config=config, config_path=config_path):
135 retry_login_interval = DEFAULT_RETRY_LOGIN_INTERVAL_SEC
136 if log_messages:
137 log_config_not_found_warning(
138 config_path,
139 f"not found. Using default {retry_login_interval} seconds ...",
140 )
141 else:
142 retry_login_interval = get_config_value(config=config, config_path=config_path)
143 if log_messages:
144 log_config_found_info(f"Retrying login every {retry_login_interval} seconds.")
146 return retry_login_interval
149def get_region(config: dict) -> str:
150 """Return region from config.
152 Args:
153 config: Configuration dictionary
155 Returns:
156 Region string ('global' or 'china')
157 """
158 config_path = ["app", "region"]
159 region = get_config_value_or_default(
160 config=config, config_path=config_path, default="global",
161 )
163 if region == "global" and not traverse_config_path(
164 config=config, config_path=config_path,
165 ):
166 log_config_not_found_warning(
167 config_path, "not found. Using default value - global ...",
168 )
169 elif region not in ["global", "china"]:
170 log_config_error(
171 config_path,
172 "is invalid. Valid values are - global or china. Using default value - global ...",
173 )
174 region = "global"
176 return region
179# =============================================================================
180# Sync Interval Configuration Functions
181# =============================================================================
184def get_sync_interval(
185 config: dict, config_path: list[str], service_name: str, log_messages: bool = True,
186) -> int:
187 """Get sync interval for a service (drive or photos).
189 Extracted common logic for retrieving sync intervals.
191 Args:
192 config: Configuration dictionary
193 config_path: Path to sync_interval config
194 service_name: Name of service for logging ("drive" or "photos")
195 log_messages: Whether to log informational messages (default: True)
197 Returns:
198 Sync interval in seconds
199 """
200 sync_interval = get_config_value_or_default(
201 config=config,
202 config_path=config_path,
203 default=DEFAULT_SYNC_INTERVAL_SEC,
204 )
206 # 0 is neither an interval nor one-shot (negative): a service that is
207 # always due would sync back to back with no sleep, which is the
208 # fastest way to get throttled by Apple. Treat it as unset.
209 if sync_interval == 0:
210 if log_messages:
211 log_invalid_config_value(
212 config_path,
213 sync_interval,
214 f"a positive number of seconds, or negative for one-shot. Using default {DEFAULT_SYNC_INTERVAL_SEC}",
215 )
216 sync_interval = DEFAULT_SYNC_INTERVAL_SEC
218 if log_messages:
219 if sync_interval == DEFAULT_SYNC_INTERVAL_SEC:
220 log_config_not_found_warning(
221 config_path,
222 f"is not found. Using default sync_interval: {sync_interval} seconds ...",
223 )
224 else:
225 log_config_found_info(
226 f"Syncing {service_name} every {sync_interval} seconds.",
227 )
229 return sync_interval
232def get_drive_sync_interval(config: dict, log_messages: bool = True) -> int:
233 """Return drive sync interval from config.
235 Args:
236 config: Configuration dictionary
237 log_messages: Whether to log informational messages (default: True)
239 Returns:
240 Drive sync interval in seconds
241 """
242 config_path = ["drive", "sync_interval"]
243 return get_sync_interval(
244 config=config,
245 config_path=config_path,
246 service_name="drive",
247 log_messages=log_messages,
248 )
251def get_drive_request_timeout(config: dict) -> int:
252 """Return drive request timeout from config.
254 Args:
255 config: Configuration dictionary
257 Returns:
258 Request timeout in seconds (default: DEFAULT_REQUEST_TIMEOUT_SEC)
259 """
260 config_path = ["drive", "request_timeout"]
261 return get_config_value_or_default(
262 config=config,
263 config_path=config_path,
264 default=DEFAULT_REQUEST_TIMEOUT_SEC,
265 )
268def get_photos_sync_interval(config: dict, log_messages: bool = True) -> int:
269 """Return photos sync interval from config.
271 Args:
272 config: Configuration dictionary
273 log_messages: Whether to log informational messages (default: True)
275 Returns:
276 Photos sync interval in seconds
277 """
278 config_path = ["photos", "sync_interval"]
279 return get_sync_interval(
280 config=config,
281 config_path=config_path,
282 service_name="photos",
283 log_messages=log_messages,
284 )
287# =============================================================================
288# Thread Configuration Functions
289# =============================================================================
292def calculate_default_max_threads() -> int:
293 """Calculate default maximum threads based on CPU cores.
295 Returns:
296 Default max threads (min of CPU count and 8)
297 """
298 return min(multiprocessing.cpu_count(), 8)
301def parse_max_threads_value(max_threads_config: Any, default_max_threads: int) -> int:
302 """Parse and validate max_threads configuration value.
304 Args:
305 max_threads_config: Raw config value (string "auto" or integer)
306 default_max_threads: Default value to use
308 Returns:
309 Validated max threads value (capped at 16)
310 """
311 # Handle "auto" value
312 if isinstance(max_threads_config, str) and max_threads_config.lower() == "auto":
313 max_threads = default_max_threads
314 log_config_found_info(
315 f"Using automatic thread count: {max_threads} threads (based on CPU cores).",
316 )
317 elif isinstance(max_threads_config, int) and max_threads_config >= 1:
318 max_threads = min(
319 max_threads_config, 16,
320 ) # Cap at 16 to avoid overwhelming servers
321 log_config_found_info(f"Using configured max_threads: {max_threads}.")
322 else:
323 log_invalid_config_value(
324 ["app", "max_threads"],
325 max_threads_config,
326 "'auto' or integer >= 1",
327 )
328 max_threads = default_max_threads
330 return max_threads
333def get_web_ui_enabled(config: dict) -> bool:
334 """Return whether the embedded web UI should start on container boot.
336 Default: **False** — opt-in. Existing mandarons installs see no
337 behaviour change; only users who explicitly set
338 ``app.web_ui.enabled: true`` open the port.
339 """
340 return bool(
341 get_config_value_or_default(
342 config=config,
343 config_path=["app", "web_ui", "enabled"],
344 default=False,
345 ),
346 )
349def get_web_ui_host(config: dict) -> str:
350 """Web UI bind address.
352 Default ``127.0.0.1`` — the web UI accepts the user's Apple ID
353 password on POST /auth/password with no built-in authentication and
354 no CSRF token (the feature assumes a reverse-proxy trust boundary
355 in front of it). Defaulting to loopback means the credential-
356 accepting form is never exposed to LAN/public on a vanilla install.
357 Users who run behind a reverse proxy or want explicit LAN exposure
358 set ``app.web_ui.host: 0.0.0.0`` consciously.
359 """
360 return str(
361 get_config_value_or_default(
362 config=config,
363 config_path=["app", "web_ui", "host"],
364 default="127.0.0.1",
365 ),
366 )
369def get_web_ui_port(config: dict) -> int:
370 """Web UI TCP port. Default ``8080``. Coexists with mandarons' legacy
371 ``EXPOSE 80`` (unused) — no port collision."""
372 return int(
373 get_config_value_or_default(
374 config=config,
375 config_path=["app", "web_ui", "port"],
376 default=8080,
377 ),
378 )
381def get_web_ui_public_url(config: dict) -> str | None:
382 """Public-facing URL for the web UI, used in notifications.
384 The daemon only knows its local bind host:port; the user-facing URL
385 (e.g. ``https://icloud.example.com`` behind a reverse proxy) must be
386 declared explicitly. When unset, notifications fall back to
387 ``http://{host}:{port}`` and log a one-shot warning at startup.
388 """
389 value = get_config_value_or_default(
390 config=config,
391 config_path=["app", "web_ui", "public_url"],
392 default=None,
393 )
394 if value is None:
395 return None
396 return str(value).rstrip("/")
399def get_trust_expiry_warn_days(config: dict) -> int:
400 """Warn this many days before Apple's trust cookie expires.
402 Default 7. The check + notification fires once per cookie-lifetime
403 when ``trust_days_remaining`` first drops below this threshold so
404 the user can tap refresh-trust *before* the sync loop hits a
405 failed-auth state.
406 """
407 return int(
408 get_config_value_or_default(
409 config=config,
410 config_path=["app", "trust_expiry_warn_days"],
411 default=7,
412 ),
413 )
416def get_trust_refresh_days(config: dict) -> int:
417 """Proactively re-trust when the cookie has this many days left.
419 Default 14. Apple's trust window is finite, but ``trust_session``
420 mints a fresh token whenever it is called on a live session, so
421 re-trusting on a schedule keeps the *persisted* token young. That is
422 what lets a container restart resume without a second factor -- the
423 common failure mode otherwise is a long-running process holding a
424 valid in-memory session while the token on disk quietly expires,
425 stranding the next cold start on a 2FA prompt.
427 Must exceed ``trust_expiry_warn_days`` for the refresh to get a
428 chance before the warning fires. Set to 0 to disable.
429 """
430 refresh_days = int(
431 get_config_value_or_default(
432 config=config,
433 config_path=["app", "trust_refresh_days"],
434 default=14,
435 ),
436 )
437 _warn_if_refresh_cannot_precede_warning(config, refresh_days)
438 return refresh_days
441def _warn_if_refresh_cannot_precede_warning(config: dict, refresh_days: int) -> None:
442 """Say so once when the refresh threshold is at or below the warn threshold.
444 The refresh exists so the expiry warning never has to fire. With
445 ``trust_refresh_days <= trust_expiry_warn_days`` the warning always fires
446 first, and the user is told to re-authenticate a session the container
447 was about to renew on its own. Not an error -- both still work -- so it
448 is a one-time warning rather than a rejected config.
449 """
450 if refresh_days <= 0:
451 return
452 warn_days = get_trust_expiry_warn_days(config=config)
453 key = "app > trust_refresh_days <= trust_expiry_warn_days"
454 if refresh_days <= warn_days and key not in _config_warning_cache:
455 _config_warning_cache.add(key)
456 LOGGER.warning(
457 f"app.trust_refresh_days ({refresh_days}) is not above "
458 f"app.trust_expiry_warn_days ({warn_days}), so the expiry warning "
459 f"will fire before the proactive refresh gets a chance. Raise "
460 f"trust_refresh_days, or lower trust_expiry_warn_days.",
461 )
464def get_app_max_threads(config: dict) -> int:
465 """Return app-level max threads from config with support for 'auto' value.
467 Args:
468 config: Configuration dictionary
470 Returns:
471 Maximum number of threads for parallel operations
472 """
473 default_max_threads = calculate_default_max_threads()
474 config_path = ["app", "max_threads"]
476 if not traverse_config_path(config=config, config_path=config_path):
477 log_config_debug(
478 f"max_threads is not found in {config_path_to_string(config_path=config_path)}. "
479 f"Using default max_threads: {default_max_threads} (auto) ...",
480 )
481 return default_max_threads
483 max_threads_config = get_config_value(config=config, config_path=config_path)
484 return parse_max_threads_value(max_threads_config, default_max_threads)
487def get_mount_marker_filename(config: dict) -> str:
488 """Return the filename used as the mount-failsafe marker (default `.mounted`).
490 The marker is the empty/sentinel file the user touches in a destination
491 directory to assert "this path is correctly bind-mounted". The mount
492 checker (``sync._check_mount_marker``) refuses to sync into a destination
493 that lacks this file when ``require_mount_marker`` is enabled.
495 Args:
496 config: Configuration dictionary
498 Returns:
499 Filename string (relative to each destination directory).
500 """
501 config_path = ["app", "mount_marker_filename"]
502 return str(get_config_value_or_default(config=config, config_path=config_path, default=".mounted"))
505def get_usage_tracking_enabled(config: dict) -> bool:
506 """Get usage tracking enabled setting from configuration.
508 Args:
509 config: Configuration dictionary
511 Returns:
512 True if usage tracking is enabled (default), False if disabled
513 """
514 config_path = ["app", "usage_tracking", "enabled"]
515 if not traverse_config_path(config=config, config_path=config_path):
516 return True # Default to enabled if not configured
518 value = get_config_value(config=config, config_path=config_path)
519 if isinstance(value, bool):
520 return value
522 # Handle string values for backwards compatibility
523 if isinstance(value, str):
524 return value.lower() not in ("false", "no", "0", "disabled", "off")
526 # Default to enabled for any other type
527 return True
530# =============================================================================
531# Root Destination Functions
532# =============================================================================
535def get_root_destination_path(config: dict) -> str:
536 """Get root destination path from config without creating directory.
538 Args:
539 config: Configuration dictionary
541 Returns:
542 Root destination path string
543 """
544 config_path = ["app", "root"]
545 root_destination = get_config_value_or_default(
546 config=config,
547 config_path=config_path,
548 default=DEFAULT_ROOT_DESTINATION,
549 )
551 if not traverse_config_path(config=config, config_path=config_path):
552 log_config_not_found_warning(
553 config_path,
554 f"root destination is missing. Using default root destination: {root_destination}",
555 )
557 return root_destination
560def prepare_root_destination(config: dict) -> str:
561 """Prepare root destination by creating directory if needed.
563 Args:
564 config: Configuration dictionary
566 Returns:
567 Absolute path to root destination directory
568 """
569 log_config_debug("Checking root destination ...")
570 root_destination = get_root_destination_path(config)
571 return ensure_directory_exists(root_destination)
574# =============================================================================
575# Drive Configuration Functions
576# =============================================================================
579def get_drive_destination_path(config: dict) -> str:
580 """Get drive destination path from config without creating directory.
582 Args:
583 config: Configuration dictionary
585 Returns:
586 Drive destination path string
587 """
588 config_path = ["drive", "destination"]
589 drive_destination = get_config_value_or_default(
590 config=config,
591 config_path=config_path,
592 default=DEFAULT_DRIVE_DESTINATION,
593 )
595 if not traverse_config_path(config=config, config_path=config_path):
596 log_config_not_found_warning(
597 config_path,
598 f"destination is missing. Using default drive destination: {drive_destination}.",
599 )
601 return drive_destination
604def prepare_drive_destination(config: dict) -> str:
605 """Prepare drive destination path by creating directory if needed.
607 Args:
608 config: Configuration dictionary
610 Returns:
611 Absolute path to drive destination directory
612 """
613 log_config_debug("Checking drive destination ...")
614 root_path = prepare_root_destination(config=config)
615 drive_destination = get_drive_destination_path(config)
616 return join_and_ensure_path(root_path, drive_destination)
619def get_drive_remove_obsolete(config: dict) -> bool:
620 """Return drive remove obsolete flag from config.
622 Args:
623 config: Configuration dictionary
625 Returns:
626 True if obsolete files should be removed, False otherwise
627 """
628 config_path = ["drive", "remove_obsolete"]
629 drive_remove_obsolete = get_config_value_or_default(
630 config=config, config_path=config_path, default=False,
631 )
633 if not drive_remove_obsolete:
634 _log_config_warning_once(
635 config_path,
636 "remove_obsolete is not found. Not removing the obsolete files and folders.",
637 )
638 else:
639 log_config_debug(
640 f"{'R' if drive_remove_obsolete else 'Not R'}emoving obsolete files and folders ...",
641 )
643 return drive_remove_obsolete
646def get_drive_flatten_packages(config: dict | None) -> bool:
647 """Return whether iCloud Drive package downloads should be kept on
648 disk as single binary files (zip / gzip bytes) instead of being
649 unpacked into bundle directories.
651 Default ``False`` — preserves the historical mandarons behaviour
652 of unpacking ``.band``-style bundles so the local representation
653 mirrors macOS's directory-bundle semantics.
655 Setting this to ``True`` is appropriate for backup-style deployments
656 (NAS, cold storage) where the operator prefers:
658 - **Single-file storage** — one mtime / size per package for
659 simpler dedup, restoration, and round-trip back to iCloud.
660 - **No internal name collisions** — two iWork files in the same
661 folder normally share bare-pathed internal entries like
662 ``Data/Document.iwa``; unpacking both into the same parent dir
663 raises ``FileExistsError``. Skipping unpack avoids this entirely.
664 - **Lower inode footprint** — large bundles often expand to
665 hundreds of small files on disk.
667 Args:
668 config: Configuration dictionary (None ok).
670 Returns:
671 bool — True if packages should be kept as single files.
672 """
673 if not config:
674 return False
675 config_path = ["drive", "flatten_packages"]
676 return bool(
677 get_config_value_or_default(
678 config=config, config_path=config_path, default=False,
679 ),
680 )
683def get_drive_require_mount_marker(config: dict) -> bool:
684 """Return whether Drive sync requires the mount-failsafe marker file.
686 When True, ``sync._check_mount_marker`` refuses to start a Drive sync
687 until the marker file (see ``get_mount_marker_filename``) exists in
688 the Drive destination directory. Protects against silent bind-mount
689 failures dumping iCloud Drive content into the wrong place.
691 Default: False (preserves historical behaviour — no breaking change).
693 Args:
694 config: Configuration dictionary
696 Returns:
697 True if the marker is required before each Drive sync.
698 """
699 config_path = ["drive", "require_mount_marker"]
700 return bool(get_config_value_or_default(config=config, config_path=config_path, default=False))
703# =============================================================================
704# Photos Configuration Functions
705# =============================================================================
708def get_photos_enumeration_chunk_size(config: dict | None) -> int:
709 """Tasks to buffer before draining via execute_parallel_downloads.
711 Smaller = lower peak memory, more per-chunk HTTP setup overhead.
712 Larger = higher peak memory, fewer chunks. Default 1000 keeps
713 resident set at ~10 MB on typical libraries while still amortising
714 connection setup. Tested empirically on a 111K-photo library:
715 1000 sustained < 1 GB resident through the full enumeration.
717 Args:
718 config: Configuration dictionary (None falls back to default).
720 Returns:
721 Positive integer chunk size, defaulting to 1000.
722 """
723 raw = get_config_value_or_default(
724 config=config or {},
725 config_path=["photos", "enumeration_chunk_size"],
726 default=DEFAULT_ENUMERATION_CHUNK_SIZE,
727 )
728 try:
729 value = int(raw)
730 except (TypeError, ValueError):
731 return DEFAULT_ENUMERATION_CHUNK_SIZE
732 return value if value > 0 else DEFAULT_ENUMERATION_CHUNK_SIZE
735def get_photos_destination_path(config: dict) -> str:
736 """Get photos destination path from config without creating directory.
738 Args:
739 config: Configuration dictionary
741 Returns:
742 Photos destination path string
743 """
744 config_path = ["photos", "destination"]
745 photos_destination = get_config_value_or_default(
746 config=config,
747 config_path=config_path,
748 default=DEFAULT_PHOTOS_DESTINATION,
749 )
751 if not traverse_config_path(config=config, config_path=config_path):
752 log_config_not_found_warning(
753 config_path,
754 f"destination is missing. Using default photos destination: {config_path_to_string(config_path)}",
755 )
757 return photos_destination
760def prepare_photos_destination(config: dict) -> str:
761 """Prepare photos destination path by creating directory if needed.
763 Args:
764 config: Configuration dictionary
766 Returns:
767 Absolute path to photos destination directory
768 """
769 log_config_debug("Checking photos destination ...")
770 root_path = prepare_root_destination(config=config)
771 photos_destination = get_photos_destination_path(config)
772 return join_and_ensure_path(root_path, photos_destination)
775def get_photos_all_albums(config: dict) -> bool:
776 """Return flag to download all albums from config.
778 Args:
779 config: Configuration dictionary
781 Returns:
782 True if all albums should be synced, False otherwise
783 """
784 config_path = ["photos", "all_albums"]
785 download_all = get_config_value_or_default(
786 config=config, config_path=config_path, default=False,
787 )
789 if download_all:
790 log_config_found_info("Syncing all albums.")
792 return download_all
795def get_photos_use_hardlinks(config: dict, log_messages: bool = True) -> bool:
796 """Return flag to use hard links for duplicate photos from config.
798 Args:
799 config: Configuration dictionary
800 log_messages: Whether to log informational messages (default: True)
802 Returns:
803 True if hard links should be used, False otherwise
804 """
805 config_path = ["photos", "use_hardlinks"]
806 use_hardlinks = get_config_value_or_default(
807 config=config, config_path=config_path, default=False,
808 )
810 if use_hardlinks and log_messages:
811 log_config_found_info("Using hard links for duplicate photos.")
813 return use_hardlinks
816def get_photos_remove_obsolete(config: dict) -> bool:
817 """Return photos remove obsolete flag from config.
819 Args:
820 config: Configuration dictionary
822 Returns:
823 True if obsolete files should be removed, False otherwise
824 """
825 config_path = ["photos", "remove_obsolete"]
826 photos_remove_obsolete = get_config_value_or_default(
827 config=config, config_path=config_path, default=False,
828 )
830 if not photos_remove_obsolete:
831 _log_config_warning_once(
832 config_path,
833 "remove_obsolete is not found. Not removing the obsolete files and folders.",
834 )
835 else:
836 log_config_debug(
837 f"{'R' if photos_remove_obsolete else 'Not R'}emoving obsolete files and folders ...",
838 )
840 return photos_remove_obsolete
843def get_photos_require_mount_marker(config: dict) -> bool:
844 """Return whether Photos sync requires the mount-failsafe marker file.
846 When True, ``sync._check_mount_marker`` refuses to start a Photos sync
847 until the marker file (see ``get_mount_marker_filename``) exists in
848 the Photos destination directory. Protects against silent bind-mount
849 failures dumping the entire iCloud photo library into the wrong place.
851 Default: False (preserves historical behaviour — no breaking change).
853 Args:
854 config: Configuration dictionary
856 Returns:
857 True if the marker is required before each Photos sync.
858 """
859 config_path = ["photos", "require_mount_marker"]
860 return bool(get_config_value_or_default(config=config, config_path=config_path, default=False))
863def get_photos_folder_format(config: dict) -> str | None:
864 """Return filename format or None.
866 Args:
867 config: Configuration dictionary
869 Returns:
870 Folder format string if configured, None otherwise
871 """
872 config_path = ["photos", "folder_format"]
873 fmt = get_config_value_or_none(config=config, config_path=config_path)
875 if fmt:
876 log_config_found_info(f"Using format {fmt}.")
878 return fmt
881def get_photos_request_timeout(config: dict) -> int:
882 """Return the HTTP read timeout for photo downloads, in seconds.
884 Drive downloads have had ``drive.request_timeout`` for a while; photo
885 downloads had no timeout at all, so a stalled connection to Apple's CDN
886 blocked its worker thread forever. The process stays alive and the
887 container stays "healthy" while the sync never finishes.
889 This is a between-bytes timeout, not a total-transfer budget, so large
890 videos are unaffected -- it only fires when the connection goes quiet.
891 """
892 config_path = ["photos", "request_timeout"]
894 if not traverse_config_path(config=config, config_path=config_path):
895 return DEFAULT_REQUEST_TIMEOUT_SEC
897 return get_config_value(config=config, config_path=config_path)
900def get_photos_obsolete_delete_limit_percent(config: dict) -> int:
901 """Share of a destination obsolete-cleanup may delete in one run.
903 Cleanup is the only destructive step in a sync and it infers deletions
904 from the absence of a path in the run's tracked-file set, so any bug that
905 leaves paths untracked is acted on at full speed. Above this share the
906 run reports instead of deleting.
908 Args:
909 config: Configuration dictionary
911 Returns:
912 Percentage of a destination a single cleanup run may delete.
913 ``DEFAULT_OBSOLETE_DELETE_LIMIT_PERCENT`` when unset; 0 disables
914 the limit entirely.
915 """
916 config_path = ["photos", "obsolete_delete_limit_percent"]
918 if not traverse_config_path(config=config, config_path=config_path):
919 return DEFAULT_OBSOLETE_DELETE_LIMIT_PERCENT
921 return get_config_value(config=config, config_path=config_path)
924# =============================================================================
925# Photos Filter Configuration Functions
926# =============================================================================
929# Friendly spellings for the canonical version keys, borrowed from icloudpd's
930# ``--size`` vocabulary: ``adjusted`` is Apple's rendered edit (``full``) and
931# ``alternative`` is the RAW/JPEG pair alternate (``original_alt``), which is
932# not an edit.
933FILE_SIZE_ALIASES = {
934 "adjusted": "full",
935 "alternative": "original_alt",
936}
939def validate_file_sizes(file_sizes: list[str]) -> list[str]:
940 """Validate and filter file sizes against valid options.
942 Accepts any key in ``PhotoAsset.PHOTO_VERSION_LOOKUP``, including the
943 ``live_video_*`` keys: add ``live_video_original`` to ``file_sizes`` to
944 download the paired ``.mov`` of a Live Photo (and ``live_video_medium`` /
945 ``live_video_thumb`` for smaller variants). Photos that aren't Live Photos
946 simply don't have those versions and are skipped.
948 Two friendly aliases are normalized to their canonical keys, mirroring
949 icloudpd's vocabulary: ``adjusted`` -> ``full`` (Apple's rendered edit)
950 and ``alternative`` -> ``original_alt`` (the RAW/alternate rendition,
951 *not* an edit). Duplicates that appear after normalization are collapsed,
952 so ``[original, adjusted, full]`` downloads the edit once.
954 Args:
955 file_sizes: List of file size strings to validate
957 Returns:
958 List of valid file sizes (defaults to ["original"] if all invalid)
959 """
960 valid_file_sizes = list(PhotoAsset.PHOTO_VERSION_LOOKUP.keys())
961 validated_sizes = []
963 for file_size in file_sizes:
964 canonical = FILE_SIZE_ALIASES.get(file_size, file_size)
965 if canonical not in valid_file_sizes:
966 log_invalid_config_value(
967 ["photos", "filters", "file_sizes"],
968 file_size,
969 ",".join(valid_file_sizes),
970 )
971 elif canonical not in validated_sizes:
972 validated_sizes.append(canonical)
974 return validated_sizes if validated_sizes else ["original"]
977def get_photos_filename_format(config: dict) -> str:
978 """Filename naming convention for downloaded photos.
980 - ``"metadata"`` (default, backward-compatible): ``name__filesize__base64id.ext``
981 — mandarons' historical format. Encodes the CloudKit asset ID into the
982 filename so the same source photo can be unambiguously round-tripped to
983 its iCloud record from the local file alone.
984 - ``"simple"``: ``name.ext`` — the boredazfcuk/docker-icloudpd convention
985 (and Apple's own download-from-icloud.com convention). Drops the
986 metadata suffix entirely. Trade-off: rename collisions are possible
987 (two distinct iCloud photos sharing the same human filename land on
988 the same path on disk; ``collect_download_task`` detects this and
989 falls back to the metadata-suffix name for the colliding photo so
990 both files coexist).
991 Useful for users migrating from boredazfcuk who want to point
992 mandarons at their existing photo tree without triggering a full
993 re-download — ``check_photo_exists`` compares by file size, not by
994 filename suffix, so matching simple-format files are correctly
995 treated as already-present.
997 Returns:
998 Either ``"metadata"`` (default) or ``"simple"``.
999 """
1000 config_path = ["photos", "filename_format"]
1001 value = get_config_value_or_none(config=config, config_path=config_path)
1002 if value is None:
1003 return "metadata"
1004 value = str(value).lower().strip()
1005 if value not in ("metadata", "simple"):
1006 log_config_not_found_warning(
1007 config_path, f"unknown value {value!r}; falling back to 'metadata'",
1008 )
1009 return "metadata"
1010 return value
1013def get_photos_file_format(config: dict) -> str | None:
1014 """Single filename template applied to all versions (``photos.file_format``).
1016 Mirrors ``folder_format`` but for the filename. Uses ``${photo.*}`` tokens
1017 (see ``photo_path_utils.render_filename_template``). When set it overrides
1018 ``filename_format``. Returns ``None`` (use ``filename_format``) when unset.
1019 """
1020 config_path = ["photos", "file_format"]
1021 value = get_config_value_or_none(config=config, config_path=config_path)
1022 if value is None:
1023 return None
1024 if not isinstance(value, str) or not value.strip():
1025 log_config_not_found_warning(config_path, "must be a non-empty template string; ignoring")
1026 return None
1027 return value
1030def get_photos_variant_separator(config: dict) -> str:
1031 """Separator inserted before the variant in ``${photo.variant_suffix}``.
1033 Defaults to ``"_"``. Only appears when a version is non-primary (not
1034 original/full), so primaries stay un-suffixed.
1035 """
1036 config_path = ["photos", "variant_separator"]
1037 value = get_config_value_or_none(config=config, config_path=config_path)
1038 return str(value) if value is not None else "_"
1041def get_photos_preserve_originals_as_bak(config: dict) -> bool:
1042 """Whether to hide untouched-original copies of edited photos via ``.original.bak``.
1044 When True AND ``photos.filters.file_sizes`` contains BOTH ``original``
1045 AND ``full`` (the edit), an edited photo lands as TWO files on disk:
1047 - ``IMG_1234.jpeg`` — the edited "current view" (visible)
1048 - ``IMG_1234.HEIC.original.bak`` — the untouched original (invisible
1049 to Plex / Photos.app / Synology Photos / any photo browser, since
1050 no app recognises ``.bak`` as an image extension), but filesystem-
1051 recoverable
1053 Unedited photos (no ``full`` version) are unaffected. ``original_alt`` is
1054 the RAW/JPEG pair alternate, not an edit, and no longer activates this.
1056 Default False (backward-compatible — no behaviour change for existing
1057 mandarons users).
1058 """
1059 config_path = ["photos", "preserve_originals_as_bak"]
1060 value = get_config_value_or_none(config=config, config_path=config_path)
1061 return bool(value) if value is not None else False
1064def get_recycle_bin_enabled(config: dict) -> bool:
1065 """Whether obsolete-file cleanup moves things to the recycle bin.
1067 ``app.recycle_bin.enabled``. Off by default: cleanup deletes, as before.
1068 The bin never widens what cleanup removes; ``remove_obsolete`` still
1069 decides that.
1070 """
1071 value = get_config_value_or_none(config=config, config_path=["app", "recycle_bin", "enabled"])
1072 return bool(value) if value is not None else False
1075def get_recycle_bin_retention_days(config: dict) -> int | None:
1076 """Days the recycle bin keeps what cleanup moved there; None keeps forever.
1078 ``app.recycle_bin.retention_days``. Anything but a positive whole number
1079 is reported and treated as unset -- keeping forever is the safe reading
1080 of a value that cannot be trusted.
1081 """
1082 config_path = ["app", "recycle_bin", "retention_days"]
1083 value = get_config_value_or_none(config=config, config_path=config_path)
1084 if value is None:
1085 return None
1086 if isinstance(value, bool) or not isinstance(value, int) or value <= 0:
1087 log_invalid_config_value(config_path, value, "a positive number of days; keeping deleted files forever")
1088 return None
1089 return value
1092def get_photos_library_destinations(config: dict) -> dict[str, str]:
1093 """Get per-library destination subdirectory mapping from photos config.
1095 Optional config block (under top-level ``photos``):
1097 .. code-block:: yaml
1099 photos:
1100 destination: photos
1101 library_destinations:
1102 PrimarySync: personal
1103 SharedLibrary: shared
1105 When set, photos from each library are written to
1106 ``<photos.destination>/<library_destinations[library]>/...`` instead of
1107 sharing one destination tree. When unset (the default), all libraries
1108 share the single ``photos.destination`` path — preserving the historical
1109 behaviour of mandarons/icloud-docker.
1111 Returns:
1112 Dict mapping library name → subdirectory relative to ``photos.destination``.
1113 Returns ``{}`` if not configured (backward-compatible default).
1114 """
1115 config_path = ["photos", "library_destinations"]
1116 mapping = get_config_value_or_none(config=config, config_path=config_path)
1117 if not mapping or not isinstance(mapping, dict):
1118 return {}
1119 return {str(k): str(v) for k, v in mapping.items()}
1122def get_photos_libraries_filter(config: dict, base_config_path: list[str]) -> list[str] | None:
1123 """Get libraries filter from photos config.
1125 Args:
1126 config: Configuration dictionary
1127 base_config_path: Base path to filters section
1129 Returns:
1130 List of library names if configured, None otherwise
1131 """
1132 config_path = base_config_path + ["libraries"]
1133 libraries = get_config_value_or_none(config=config, config_path=config_path)
1135 if not libraries or len(libraries) == 0:
1136 log_config_not_found_warning(
1137 config_path, "not found. Downloading all libraries ...",
1138 )
1139 return None
1141 return libraries
1144def get_photos_albums_filter(
1145 config: dict, base_config_path: list[str],
1146) -> list[str] | None:
1147 """Get albums filter from photos config.
1149 Args:
1150 config: Configuration dictionary
1151 base_config_path: Base path to filters section
1153 Returns:
1154 List of album names if configured, None otherwise
1155 """
1156 config_path = base_config_path + ["albums"]
1157 albums = get_config_value_or_none(config=config, config_path=config_path)
1159 if not albums or len(albums) == 0:
1160 log_config_not_found_warning(
1161 config_path, "not found. Downloading all albums ...",
1162 )
1163 return None
1165 return albums
1168def get_photos_file_sizes_filter(
1169 config: dict, base_config_path: list[str],
1170) -> list[str]:
1171 """Get file sizes filter from photos config.
1173 Args:
1174 config: Configuration dictionary
1175 base_config_path: Base path to filters section
1177 Returns:
1178 List of file size options (defaults to ["original"])
1179 """
1180 config_path = base_config_path + ["file_sizes"]
1182 if not traverse_config_path(config=config, config_path=config_path):
1183 log_config_not_found_warning(
1184 config_path, "not found. Downloading original size photos ...",
1185 )
1186 return ["original"]
1188 file_sizes = get_config_value(config=config, config_path=config_path)
1189 return validate_file_sizes(file_sizes)
1192def get_photos_fallback_to_original(config: dict) -> bool:
1193 """Whether an absent requested size falls back to ``original``.
1195 ``photos.filters.fallback_to_original`` (default false, backward
1196 compatible). When true, a photo that has no copy of a requested version
1197 is downloaded as its ``original`` instead of being skipped, which is how
1198 icloudpd behaves. That makes ``file_sizes: [adjusted]`` mean "the edit,
1199 or the untouched original if the photo was never edited".
1201 ``live_video_*`` requests never fall back: those versions are QuickTime
1202 movies, and substituting the still image would write JPEG bytes under a
1203 ``.mov`` name.
1205 Args:
1206 config: Configuration dictionary
1208 Returns:
1209 True when the fallback is enabled, False otherwise.
1210 """
1211 config_path = ["photos", "filters", "fallback_to_original"]
1212 value = get_config_value_or_none(config=config, config_path=config_path)
1213 return bool(value) if value is not None else False
1216def get_photos_extensions_filter(
1217 config: dict, base_config_path: list[str],
1218) -> list[str] | None:
1219 """Get extensions filter from photos config.
1221 Args:
1222 config: Configuration dictionary
1223 base_config_path: Base path to filters section
1225 Returns:
1226 List of file extensions if configured, None otherwise
1227 """
1228 config_path = base_config_path + ["extensions"]
1229 extensions = get_config_value_or_none(config=config, config_path=config_path)
1231 if not extensions or len(extensions) == 0:
1232 log_config_not_found_warning(
1233 config_path, "not found. Downloading all extensions ...",
1234 )
1235 return None
1237 return extensions
1240def get_photos_filters(config: dict) -> dict[str, Any]:
1241 """Return photos filters from config.
1243 Args:
1244 config: Configuration dictionary
1246 Returns:
1247 Dictionary containing filter configuration for photos
1248 """
1249 photos_filters = {
1250 "libraries": None,
1251 "albums": None,
1252 "file_sizes": ["original"],
1253 "extensions": None,
1254 "fallback_to_original": False,
1255 }
1257 base_config_path = ["photos", "filters"]
1259 # Check for filters section existence
1260 if not traverse_config_path(config=config, config_path=base_config_path):
1261 log_config_not_found_warning(
1262 base_config_path,
1263 "not found. Downloading all libraries and albums with original size ...",
1264 )
1265 return photos_filters
1267 # Parse individual filter components
1268 photos_filters["libraries"] = get_photos_libraries_filter(config, base_config_path)
1269 photos_filters["albums"] = get_photos_albums_filter(config, base_config_path)
1270 photos_filters["file_sizes"] = get_photos_file_sizes_filter(
1271 config, base_config_path,
1272 )
1273 photos_filters["extensions"] = get_photos_extensions_filter(
1274 config, base_config_path,
1275 )
1276 photos_filters["fallback_to_original"] = get_photos_fallback_to_original(
1277 config=config,
1278 )
1280 return photos_filters
1283# =============================================================================
1284# SMTP Configuration Functions
1285# =============================================================================
1288def get_smtp_config_value(
1289 config: dict, key: str, warn_if_missing: bool = True,
1290) -> str | None:
1291 """Get SMTP configuration value with optional warning.
1293 Common helper for SMTP config retrieval to reduce duplication.
1295 Args:
1296 config: Configuration dictionary
1297 key: SMTP config key name
1298 warn_if_missing: Whether to log warning if not found
1300 Returns:
1301 Config value if found, None otherwise
1302 """
1303 config_path = ["app", "smtp", key]
1304 value = get_config_value_or_none(config=config, config_path=config_path)
1306 if value is None and warn_if_missing:
1307 log_config_not_found_warning(config_path, f"{key} is not found.")
1309 return value
1312def get_smtp_email(config: dict) -> str | None:
1313 """Return smtp from email from config.
1315 Args:
1316 config: Configuration dictionary
1318 Returns:
1319 SMTP email address if configured, None otherwise
1320 """
1321 return get_smtp_config_value(config, "email", warn_if_missing=False)
1324def get_smtp_username(config: dict) -> str | None:
1325 """Return smtp username from the config, if set.
1327 Args:
1328 config: Configuration dictionary
1330 Returns:
1331 SMTP username if configured, None otherwise
1332 """
1333 return get_smtp_config_value(config, "username", warn_if_missing=False)
1336def get_smtp_to_email(config: dict) -> str | None:
1337 """Return smtp to email from config, defaults to from email.
1339 Args:
1340 config: Configuration dictionary
1342 Returns:
1343 SMTP 'to' email address, falling back to 'from' email if not specified
1344 """
1345 to_email = get_smtp_config_value(config, "to", warn_if_missing=False)
1346 return to_email if to_email else get_smtp_email(config=config)
1349def get_smtp_password(config: dict) -> str | None:
1350 """Return smtp password from config.
1352 Args:
1353 config: Configuration dictionary
1355 Returns:
1356 SMTP password if configured, None otherwise
1357 """
1358 return get_smtp_config_value(config, "password", warn_if_missing=True)
1361def get_smtp_host(config: dict) -> str | None:
1362 """Return smtp host from config.
1364 Args:
1365 config: Configuration dictionary
1367 Returns:
1368 SMTP host if configured, None otherwise
1369 """
1370 return get_smtp_config_value(config, "host", warn_if_missing=True)
1373def get_smtp_port(config: dict) -> int | None:
1374 """Return smtp port from config.
1376 Args:
1377 config: Configuration dictionary
1379 Returns:
1380 SMTP port if configured, None otherwise
1381 """
1382 return get_smtp_config_value(config, "port", warn_if_missing=True) # type: ignore[return-value]
1385def get_smtp_no_tls(config: dict) -> bool:
1386 """Return smtp no_tls flag from config.
1388 Args:
1389 config: Configuration dictionary
1391 Returns:
1392 True if TLS should be disabled, False otherwise
1393 """
1394 no_tls = get_smtp_config_value(config, "no_tls", warn_if_missing=True)
1395 return no_tls if no_tls is not None else False # type: ignore[return-value]
1398# =============================================================================
1399# Notification Service Configuration Functions
1400# =============================================================================
1403def get_notification_config_value(config: dict, service: str, key: str) -> str | None:
1404 """Get notification service configuration value.
1406 Common helper for notification service config retrieval.
1408 Args:
1409 config: Configuration dictionary
1410 service: Service name (telegram, discord, pushover)
1411 key: Config key name
1413 Returns:
1414 Config value if found, None otherwise
1415 """
1416 config_path = ["app", service, key]
1417 value = get_config_value_or_none(config=config, config_path=config_path)
1419 if value is None:
1420 log_config_not_found_warning(config_path, f"{key} is not found.")
1422 return value
1425def get_telegram_bot_token(config: dict) -> str | None:
1426 """Return telegram bot token from config.
1428 Args:
1429 config: Configuration dictionary
1431 Returns:
1432 Telegram bot token if configured, None otherwise
1433 """
1434 return get_notification_config_value(config, "telegram", "bot_token")
1437def get_telegram_chat_id(config: dict) -> str | None:
1438 """Return telegram chat id from config.
1440 Args:
1441 config: Configuration dictionary
1443 Returns:
1444 Telegram chat ID if configured, None otherwise
1445 """
1446 return get_notification_config_value(config, "telegram", "chat_id")
1449def get_telegram_listen_enabled(config: dict) -> bool:
1450 """Whether to poll Telegram for inbound replies during a 2FA wait.
1452 Opt-in (default False). When True, the 2FA wait gap is replaced with a
1453 Telegram ``getUpdates`` poll: the user replies the auth keyword to have a
1454 code pushed to their Apple devices, then replies the 6-digit code -- so
1455 re-authentication can be completed from a phone, headless.
1457 Reuses ``app.telegram.bot_token`` / ``chat_id`` from the existing outbound
1458 config; ``chat_id`` filtering means only messages from the user's own chat
1459 are honoured (no separate auth surface).
1460 """
1461 return bool(
1462 get_config_value_or_none(
1463 config=config,
1464 config_path=["app", "telegram", "listen"],
1465 ),
1466 )
1469def get_telegram_auth_keyword(config: dict) -> str:
1470 """The reply keyword that triggers a 2FA push (default ``auth``).
1472 Customisable via ``app.telegram.auth_keyword`` so users running multiple
1473 containers in one chat can target a specific one (e.g. ``auth-photos``).
1474 Matched case-insensitively; returns the lowercased word.
1475 """
1476 # Optional key with a sane default -- quiet lookup so the absent case does
1477 # not log a "not found" warning on every 2FA wait.
1478 value = get_config_value_or_none(
1479 config=config,
1480 config_path=["app", "telegram", "auth_keyword"],
1481 )
1482 return str(value).strip().lower() if value else "auth"
1485def get_discord_webhook_url(config: dict) -> str | None:
1486 """Return discord webhook_url from config.
1488 Args:
1489 config: Configuration dictionary
1491 Returns:
1492 Discord webhook URL if configured, None otherwise
1493 """
1494 return get_notification_config_value(config, "discord", "webhook_url")
1497def get_discord_username(config: dict) -> str | None:
1498 """Return discord username from config.
1500 Args:
1501 config: Configuration dictionary
1503 Returns:
1504 Discord username if configured, None otherwise
1505 """
1506 return get_notification_config_value(config, "discord", "username")
1509def get_pushover_user_key(config: dict) -> str | None:
1510 """Return Pushover user key from config.
1512 Args:
1513 config: Configuration dictionary
1515 Returns:
1516 Pushover user key if configured, None otherwise
1517 """
1518 return get_notification_config_value(config, "pushover", "user_key")
1521def get_pushover_api_token(config: dict) -> str | None:
1522 """Return Pushover API token from config.
1524 Args:
1525 config: Configuration dictionary
1527 Returns:
1528 Pushover API token if configured, None otherwise
1529 """
1530 return get_notification_config_value(config, "pushover", "api_token")
1533def get_pushover_notification_priority(config: dict) -> int | None:
1534 """Return Pushover notification priority from config.
1536 Args:
1537 config: Configuration dictionary
1539 Returns:
1540 Pushover notification priority if configured, None otherwise
1541 """
1542 config_path = ["app", "pushover", "priority"]
1543 return get_config_value_or_none(config=config, config_path=config_path)
1546def get_webhook_url(config: dict, event: str) -> str | None:
1547 """Return the ping URL configured for a sync-lifecycle event.
1549 ``app.webhooks.start`` / ``success`` / ``failure`` are each optional and
1550 independent. Looked up quietly rather than through
1551 ``get_notification_config_value``: every event is consulted on every sync
1552 cycle, so the "not found" warning the other notification getters log would
1553 become three lines per cycle for anyone not using webhooks.
1555 Args:
1556 config: Configuration dictionary
1557 event: Event name -- ``start``, ``success`` or ``failure``
1559 Returns:
1560 The URL if configured and non-blank, None otherwise
1561 """
1562 config_path = ["app", "webhooks", event]
1563 value = get_config_value_or_none(config=config, config_path=config_path)
1564 if not value:
1565 return None
1566 return str(value).strip() or None
1569def get_webhook_post_url(config: dict) -> str | None:
1570 """Return ``app.webhooks.url`` -- the endpoint every event is POSTed to.
1572 Distinct from the ``start``/``success``/``failure`` GET pings: those report
1573 a cycle boundary to a monitor, this one receives the whole notification
1574 stream as JSON. Either, both or neither may be configured.
1576 Args:
1577 config: Configuration dictionary
1579 Returns:
1580 The URL if configured and non-blank, None otherwise
1581 """
1582 return get_webhook_url(config=config, event="url")
1585def get_webhook_events(config: dict) -> list[str] | None:
1586 """Return the ``app.webhooks.events`` allow-list, or None for "all".
1588 None and a configured list mean different things: no key at all sends
1589 every event, while an explicit (even empty) list sends only what it
1590 names. A non-list value is treated as absent rather than as an empty
1591 filter, so a typo cannot silently mute the endpoint.
1593 Args:
1594 config: Configuration dictionary
1596 Returns:
1597 List of event names, or None when unfiltered
1598 """
1599 value = get_config_value_or_none(config=config, config_path=["app", "webhooks", "events"])
1600 if not isinstance(value, list):
1601 return None
1602 return [str(event).strip() for event in value]
1605def get_webhook_headers(config: dict) -> dict[str, str]:
1606 """Return the extra request headers for ``app.webhooks.url``.
1608 Receivers behind an authenticating proxy need a bearer token or an API
1609 key, which has no other home in the config. Values are credentials and
1610 are never logged.
1612 Args:
1613 config: Configuration dictionary
1615 A key left without a value in YAML parses as None; such an entry is
1616 dropped rather than sent as the string "None".
1618 Returns:
1619 Mapping of header name to value; empty when unconfigured
1620 """
1621 value = get_config_value_or_none(config=config, config_path=["app", "webhooks", "headers"])
1622 if not isinstance(value, dict):
1623 return {}
1624 return {
1625 str(name): str(header_value)
1626 for name, header_value in value.items()
1627 if header_value is not None
1628 }
1631# =============================================================================
1632# Sync Summary Notification Configuration Functions
1633# =============================================================================
1636def get_sync_summary_enabled(config: dict) -> bool:
1637 """Return whether sync summary notifications are enabled.
1639 Args:
1640 config: Configuration dictionary
1642 Returns:
1643 True if sync summary is enabled, False otherwise (default: False)
1644 """
1645 config_path = ["app", "notifications", "sync_summary", "enabled"]
1646 if not traverse_config_path(config=config, config_path=config_path):
1647 return False
1649 value = get_config_value(config=config, config_path=config_path)
1650 return bool(value) if value is not None else False
1653def get_sync_summary_on_success(config: dict) -> bool:
1654 """Return whether to send summary on successful syncs.
1656 Args:
1657 config: Configuration dictionary
1659 Returns:
1660 True if should send on success, False otherwise (default: True)
1661 """
1662 config_path = ["app", "notifications", "sync_summary", "on_success"]
1663 if not traverse_config_path(config=config, config_path=config_path):
1664 return True # Default to True if not configured
1666 value = get_config_value(config=config, config_path=config_path)
1667 return bool(value) if value is not None else True
1670def get_sync_summary_on_error(config: dict) -> bool:
1671 """Return whether to send summary when errors occur.
1673 Args:
1674 config: Configuration dictionary
1676 Returns:
1677 True if should send on error, False otherwise (default: True)
1678 """
1679 config_path = ["app", "notifications", "sync_summary", "on_error"]
1680 if not traverse_config_path(config=config, config_path=config_path):
1681 return True # Default to True if not configured
1683 value = get_config_value(config=config, config_path=config_path)
1684 return bool(value) if value is not None else True
1687def get_sync_summary_min_downloads(config: dict) -> int:
1688 """Return minimum downloads required to trigger notification.
1690 Args:
1691 config: Configuration dictionary
1693 Returns:
1694 Minimum downloads threshold (default: 1)
1695 """
1696 config_path = ["app", "notifications", "sync_summary", "min_downloads"]
1697 if not traverse_config_path(config=config, config_path=config_path):
1698 return 1 # Default to 1 if not configured
1700 value = get_config_value(config=config, config_path=config_path)
1701 return int(value) if value is not None else 1