Coverage for src/config_parser.py: 100%
402 statements
« prev ^ index » next coverage.py v7.16.2, created at 2026-10-08 21:05 +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.zosia.io`` 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# =============================================================================
929def validate_file_sizes(file_sizes: list[str]) -> list[str]:
930 """Validate and filter file sizes against valid options.
932 Accepts any key in ``PhotoAsset.PHOTO_VERSION_LOOKUP``, including the
933 ``live_video_*`` keys: add ``live_video_original`` to ``file_sizes`` to
934 download the paired ``.mov`` of a Live Photo (and ``live_video_medium`` /
935 ``live_video_thumb`` for smaller variants). Photos that aren't Live Photos
936 simply don't have those versions and are skipped.
938 Args:
939 file_sizes: List of file size strings to validate
941 Returns:
942 List of valid file sizes (defaults to ["original"] if all invalid)
943 """
944 valid_file_sizes = list(PhotoAsset.PHOTO_VERSION_LOOKUP.keys())
945 validated_sizes = []
947 for file_size in file_sizes:
948 if file_size not in valid_file_sizes:
949 log_invalid_config_value(
950 ["photos", "filters", "file_sizes"],
951 file_size,
952 ",".join(valid_file_sizes),
953 )
954 else:
955 validated_sizes.append(file_size)
957 return validated_sizes if validated_sizes else ["original"]
960def get_photos_filename_format(config: dict) -> str:
961 """Filename naming convention for downloaded photos.
963 - ``"metadata"`` (default, backward-compatible): ``name__filesize__base64id.ext``
964 — mandarons' historical format. Encodes the CloudKit asset ID into the
965 filename so the same source photo can be unambiguously round-tripped to
966 its iCloud record from the local file alone.
967 - ``"simple"``: ``name.ext`` — the boredazfcuk/docker-icloudpd convention
968 (and Apple's own download-from-icloud.com convention). Drops the
969 metadata suffix entirely. Trade-off: rename collisions are possible
970 (two distinct iCloud photos sharing the same human filename land on
971 the same path on disk; ``collect_download_task`` detects this and
972 falls back to the metadata-suffix name for the colliding photo so
973 both files coexist).
974 Useful for users migrating from boredazfcuk who want to point
975 mandarons at their existing photo tree without triggering a full
976 re-download — ``check_photo_exists`` compares by file size, not by
977 filename suffix, so matching simple-format files are correctly
978 treated as already-present.
980 Returns:
981 Either ``"metadata"`` (default) or ``"simple"``.
982 """
983 config_path = ["photos", "filename_format"]
984 value = get_config_value_or_none(config=config, config_path=config_path)
985 if value is None:
986 return "metadata"
987 value = str(value).lower().strip()
988 if value not in ("metadata", "simple"):
989 log_config_not_found_warning(
990 config_path, f"unknown value {value!r}; falling back to 'metadata'",
991 )
992 return "metadata"
993 return value
996def get_photos_file_format(config: dict) -> str | None:
997 """Single filename template applied to all versions (``photos.file_format``).
999 Mirrors ``folder_format`` but for the filename. Uses ``${photo.*}`` tokens
1000 (see ``photo_path_utils.render_filename_template``). When set it overrides
1001 ``filename_format``. Returns ``None`` (use ``filename_format``) when unset.
1002 """
1003 config_path = ["photos", "file_format"]
1004 value = get_config_value_or_none(config=config, config_path=config_path)
1005 if value is None:
1006 return None
1007 if not isinstance(value, str) or not value.strip():
1008 log_config_not_found_warning(config_path, "must be a non-empty template string; ignoring")
1009 return None
1010 return value
1013def get_photos_variant_separator(config: dict) -> str:
1014 """Separator inserted before the variant in ``${photo.variant_suffix}``.
1016 Defaults to ``"_"``. Only appears when a version is non-primary (not
1017 original/full), so primaries stay un-suffixed.
1018 """
1019 config_path = ["photos", "variant_separator"]
1020 value = get_config_value_or_none(config=config, config_path=config_path)
1021 return str(value) if value is not None else "_"
1024def get_photos_preserve_originals_as_bak(config: dict) -> bool:
1025 """Whether to hide untouched-original copies of edited photos via ``.original.bak``.
1027 When True AND ``photos.filters.file_sizes`` contains BOTH ``original``
1028 AND ``original_alt``, an edited photo lands as TWO files on disk:
1030 - ``IMG_1234.JPG`` — the edited "current view" (visible)
1031 - ``IMG_1234.HEIC.original.bak`` — the untouched original (invisible
1032 to Plex / Photos.app / Synology Photos / any photo browser, since
1033 no app recognises ``.bak`` as an image extension), but filesystem-
1034 recoverable
1036 Unedited photos (no ``original_alt`` for them) are unaffected.
1038 Default False (backward-compatible — no behaviour change for existing
1039 mandarons users).
1040 """
1041 config_path = ["photos", "preserve_originals_as_bak"]
1042 value = get_config_value_or_none(config=config, config_path=config_path)
1043 return bool(value) if value is not None else False
1046def get_recycle_bin_enabled(config: dict) -> bool:
1047 """Whether obsolete-file cleanup moves things to the recycle bin.
1049 ``app.recycle_bin.enabled``. Off by default: cleanup deletes, as before.
1050 The bin never widens what cleanup removes; ``remove_obsolete`` still
1051 decides that.
1052 """
1053 value = get_config_value_or_none(config=config, config_path=["app", "recycle_bin", "enabled"])
1054 return bool(value) if value is not None else False
1057def get_recycle_bin_retention_days(config: dict) -> int | None:
1058 """Days the recycle bin keeps what cleanup moved there; None keeps forever.
1060 ``app.recycle_bin.retention_days``. Anything but a positive whole number
1061 is reported and treated as unset -- keeping forever is the safe reading
1062 of a value that cannot be trusted.
1063 """
1064 config_path = ["app", "recycle_bin", "retention_days"]
1065 value = get_config_value_or_none(config=config, config_path=config_path)
1066 if value is None:
1067 return None
1068 if isinstance(value, bool) or not isinstance(value, int) or value <= 0:
1069 log_invalid_config_value(config_path, value, "a positive number of days; keeping deleted files forever")
1070 return None
1071 return value
1074def get_photos_library_destinations(config: dict) -> dict[str, str]:
1075 """Get per-library destination subdirectory mapping from photos config.
1077 Optional config block (under top-level ``photos``):
1079 .. code-block:: yaml
1081 photos:
1082 destination: photos
1083 library_destinations:
1084 PrimarySync: personal
1085 SharedLibrary: shared
1087 When set, photos from each library are written to
1088 ``<photos.destination>/<library_destinations[library]>/...`` instead of
1089 sharing one destination tree. When unset (the default), all libraries
1090 share the single ``photos.destination`` path — preserving the historical
1091 behaviour of mandarons/icloud-docker.
1093 Returns:
1094 Dict mapping library name → subdirectory relative to ``photos.destination``.
1095 Returns ``{}`` if not configured (backward-compatible default).
1096 """
1097 config_path = ["photos", "library_destinations"]
1098 mapping = get_config_value_or_none(config=config, config_path=config_path)
1099 if not mapping or not isinstance(mapping, dict):
1100 return {}
1101 return {str(k): str(v) for k, v in mapping.items()}
1104def get_photos_libraries_filter(config: dict, base_config_path: list[str]) -> list[str] | None:
1105 """Get libraries filter from photos config.
1107 Args:
1108 config: Configuration dictionary
1109 base_config_path: Base path to filters section
1111 Returns:
1112 List of library names if configured, None otherwise
1113 """
1114 config_path = base_config_path + ["libraries"]
1115 libraries = get_config_value_or_none(config=config, config_path=config_path)
1117 if not libraries or len(libraries) == 0:
1118 log_config_not_found_warning(
1119 config_path, "not found. Downloading all libraries ...",
1120 )
1121 return None
1123 return libraries
1126def get_photos_albums_filter(
1127 config: dict, base_config_path: list[str],
1128) -> list[str] | None:
1129 """Get albums filter from photos config.
1131 Args:
1132 config: Configuration dictionary
1133 base_config_path: Base path to filters section
1135 Returns:
1136 List of album names if configured, None otherwise
1137 """
1138 config_path = base_config_path + ["albums"]
1139 albums = get_config_value_or_none(config=config, config_path=config_path)
1141 if not albums or len(albums) == 0:
1142 log_config_not_found_warning(
1143 config_path, "not found. Downloading all albums ...",
1144 )
1145 return None
1147 return albums
1150def get_photos_file_sizes_filter(
1151 config: dict, base_config_path: list[str],
1152) -> list[str]:
1153 """Get file sizes filter from photos config.
1155 Args:
1156 config: Configuration dictionary
1157 base_config_path: Base path to filters section
1159 Returns:
1160 List of file size options (defaults to ["original"])
1161 """
1162 config_path = base_config_path + ["file_sizes"]
1164 if not traverse_config_path(config=config, config_path=config_path):
1165 log_config_not_found_warning(
1166 config_path, "not found. Downloading original size photos ...",
1167 )
1168 return ["original"]
1170 file_sizes = get_config_value(config=config, config_path=config_path)
1171 return validate_file_sizes(file_sizes)
1174def get_photos_extensions_filter(
1175 config: dict, base_config_path: list[str],
1176) -> list[str] | None:
1177 """Get extensions filter from photos config.
1179 Args:
1180 config: Configuration dictionary
1181 base_config_path: Base path to filters section
1183 Returns:
1184 List of file extensions if configured, None otherwise
1185 """
1186 config_path = base_config_path + ["extensions"]
1187 extensions = get_config_value_or_none(config=config, config_path=config_path)
1189 if not extensions or len(extensions) == 0:
1190 log_config_not_found_warning(
1191 config_path, "not found. Downloading all extensions ...",
1192 )
1193 return None
1195 return extensions
1198def get_photos_filters(config: dict) -> dict[str, Any]:
1199 """Return photos filters from config.
1201 Args:
1202 config: Configuration dictionary
1204 Returns:
1205 Dictionary containing filter configuration for photos
1206 """
1207 photos_filters = {
1208 "libraries": None,
1209 "albums": None,
1210 "file_sizes": ["original"],
1211 "extensions": None,
1212 }
1214 base_config_path = ["photos", "filters"]
1216 # Check for filters section existence
1217 if not traverse_config_path(config=config, config_path=base_config_path):
1218 log_config_not_found_warning(
1219 base_config_path,
1220 "not found. Downloading all libraries and albums with original size ...",
1221 )
1222 return photos_filters
1224 # Parse individual filter components
1225 photos_filters["libraries"] = get_photos_libraries_filter(config, base_config_path)
1226 photos_filters["albums"] = get_photos_albums_filter(config, base_config_path)
1227 photos_filters["file_sizes"] = get_photos_file_sizes_filter(
1228 config, base_config_path,
1229 )
1230 photos_filters["extensions"] = get_photos_extensions_filter(
1231 config, base_config_path,
1232 )
1234 return photos_filters
1237# =============================================================================
1238# SMTP Configuration Functions
1239# =============================================================================
1242def get_smtp_config_value(
1243 config: dict, key: str, warn_if_missing: bool = True,
1244) -> str | None:
1245 """Get SMTP configuration value with optional warning.
1247 Common helper for SMTP config retrieval to reduce duplication.
1249 Args:
1250 config: Configuration dictionary
1251 key: SMTP config key name
1252 warn_if_missing: Whether to log warning if not found
1254 Returns:
1255 Config value if found, None otherwise
1256 """
1257 config_path = ["app", "smtp", key]
1258 value = get_config_value_or_none(config=config, config_path=config_path)
1260 if value is None and warn_if_missing:
1261 log_config_not_found_warning(config_path, f"{key} is not found.")
1263 return value
1266def get_smtp_email(config: dict) -> str | None:
1267 """Return smtp from email from config.
1269 Args:
1270 config: Configuration dictionary
1272 Returns:
1273 SMTP email address if configured, None otherwise
1274 """
1275 return get_smtp_config_value(config, "email", warn_if_missing=False)
1278def get_smtp_username(config: dict) -> str | None:
1279 """Return smtp username from the config, if set.
1281 Args:
1282 config: Configuration dictionary
1284 Returns:
1285 SMTP username if configured, None otherwise
1286 """
1287 return get_smtp_config_value(config, "username", warn_if_missing=False)
1290def get_smtp_to_email(config: dict) -> str | None:
1291 """Return smtp to email from config, defaults to from email.
1293 Args:
1294 config: Configuration dictionary
1296 Returns:
1297 SMTP 'to' email address, falling back to 'from' email if not specified
1298 """
1299 to_email = get_smtp_config_value(config, "to", warn_if_missing=False)
1300 return to_email if to_email else get_smtp_email(config=config)
1303def get_smtp_password(config: dict) -> str | None:
1304 """Return smtp password from config.
1306 Args:
1307 config: Configuration dictionary
1309 Returns:
1310 SMTP password if configured, None otherwise
1311 """
1312 return get_smtp_config_value(config, "password", warn_if_missing=True)
1315def get_smtp_host(config: dict) -> str | None:
1316 """Return smtp host from config.
1318 Args:
1319 config: Configuration dictionary
1321 Returns:
1322 SMTP host if configured, None otherwise
1323 """
1324 return get_smtp_config_value(config, "host", warn_if_missing=True)
1327def get_smtp_port(config: dict) -> int | None:
1328 """Return smtp port from config.
1330 Args:
1331 config: Configuration dictionary
1333 Returns:
1334 SMTP port if configured, None otherwise
1335 """
1336 return get_smtp_config_value(config, "port", warn_if_missing=True) # type: ignore[return-value]
1339def get_smtp_no_tls(config: dict) -> bool:
1340 """Return smtp no_tls flag from config.
1342 Args:
1343 config: Configuration dictionary
1345 Returns:
1346 True if TLS should be disabled, False otherwise
1347 """
1348 no_tls = get_smtp_config_value(config, "no_tls", warn_if_missing=True)
1349 return no_tls if no_tls is not None else False # type: ignore[return-value]
1352# =============================================================================
1353# Notification Service Configuration Functions
1354# =============================================================================
1357def get_notification_config_value(config: dict, service: str, key: str) -> str | None:
1358 """Get notification service configuration value.
1360 Common helper for notification service config retrieval.
1362 Args:
1363 config: Configuration dictionary
1364 service: Service name (telegram, discord, pushover)
1365 key: Config key name
1367 Returns:
1368 Config value if found, None otherwise
1369 """
1370 config_path = ["app", service, key]
1371 value = get_config_value_or_none(config=config, config_path=config_path)
1373 if value is None:
1374 log_config_not_found_warning(config_path, f"{key} is not found.")
1376 return value
1379def get_telegram_bot_token(config: dict) -> str | None:
1380 """Return telegram bot token from config.
1382 Args:
1383 config: Configuration dictionary
1385 Returns:
1386 Telegram bot token if configured, None otherwise
1387 """
1388 return get_notification_config_value(config, "telegram", "bot_token")
1391def get_telegram_chat_id(config: dict) -> str | None:
1392 """Return telegram chat id from config.
1394 Args:
1395 config: Configuration dictionary
1397 Returns:
1398 Telegram chat ID if configured, None otherwise
1399 """
1400 return get_notification_config_value(config, "telegram", "chat_id")
1403def get_telegram_listen_enabled(config: dict) -> bool:
1404 """Whether to poll Telegram for inbound replies during a 2FA wait.
1406 Opt-in (default False). When True, the 2FA wait gap is replaced with a
1407 Telegram ``getUpdates`` poll: the user replies the auth keyword to have a
1408 code pushed to their Apple devices, then replies the 6-digit code -- so
1409 re-authentication can be completed from a phone, headless.
1411 Reuses ``app.telegram.bot_token`` / ``chat_id`` from the existing outbound
1412 config; ``chat_id`` filtering means only messages from the user's own chat
1413 are honoured (no separate auth surface).
1414 """
1415 return bool(
1416 get_config_value_or_none(
1417 config=config,
1418 config_path=["app", "telegram", "listen"],
1419 ),
1420 )
1423def get_telegram_auth_keyword(config: dict) -> str:
1424 """The reply keyword that triggers a 2FA push (default ``auth``).
1426 Customisable via ``app.telegram.auth_keyword`` so users running multiple
1427 containers in one chat can target a specific one (e.g. ``auth-photos``).
1428 Matched case-insensitively; returns the lowercased word.
1429 """
1430 # Optional key with a sane default -- quiet lookup so the absent case does
1431 # not log a "not found" warning on every 2FA wait.
1432 value = get_config_value_or_none(
1433 config=config,
1434 config_path=["app", "telegram", "auth_keyword"],
1435 )
1436 return str(value).strip().lower() if value else "auth"
1439def get_discord_webhook_url(config: dict) -> str | None:
1440 """Return discord webhook_url from config.
1442 Args:
1443 config: Configuration dictionary
1445 Returns:
1446 Discord webhook URL if configured, None otherwise
1447 """
1448 return get_notification_config_value(config, "discord", "webhook_url")
1451def get_discord_username(config: dict) -> str | None:
1452 """Return discord username from config.
1454 Args:
1455 config: Configuration dictionary
1457 Returns:
1458 Discord username if configured, None otherwise
1459 """
1460 return get_notification_config_value(config, "discord", "username")
1463def get_pushover_user_key(config: dict) -> str | None:
1464 """Return Pushover user key from config.
1466 Args:
1467 config: Configuration dictionary
1469 Returns:
1470 Pushover user key if configured, None otherwise
1471 """
1472 return get_notification_config_value(config, "pushover", "user_key")
1475def get_pushover_api_token(config: dict) -> str | None:
1476 """Return Pushover API token from config.
1478 Args:
1479 config: Configuration dictionary
1481 Returns:
1482 Pushover API token if configured, None otherwise
1483 """
1484 return get_notification_config_value(config, "pushover", "api_token")
1487def get_pushover_notification_priority(config: dict) -> int | None:
1488 """Return Pushover notification priority from config.
1490 Args:
1491 config: Configuration dictionary
1493 Returns:
1494 Pushover notification priority if configured, None otherwise
1495 """
1496 config_path = ["app", "pushover", "priority"]
1497 return get_config_value_or_none(config=config, config_path=config_path)
1500def get_webhook_url(config: dict, event: str) -> str | None:
1501 """Return the ping URL configured for a sync-lifecycle event.
1503 ``app.webhooks.start`` / ``success`` / ``failure`` are each optional and
1504 independent. Looked up quietly rather than through
1505 ``get_notification_config_value``: every event is consulted on every sync
1506 cycle, so the "not found" warning the other notification getters log would
1507 become three lines per cycle for anyone not using webhooks.
1509 Args:
1510 config: Configuration dictionary
1511 event: Event name -- ``start``, ``success`` or ``failure``
1513 Returns:
1514 The URL if configured and non-blank, None otherwise
1515 """
1516 config_path = ["app", "webhooks", event]
1517 value = get_config_value_or_none(config=config, config_path=config_path)
1518 if not value:
1519 return None
1520 return str(value).strip() or None
1523def get_webhook_post_url(config: dict) -> str | None:
1524 """Return ``app.webhooks.url`` -- the endpoint every event is POSTed to.
1526 Distinct from the ``start``/``success``/``failure`` GET pings: those report
1527 a cycle boundary to a monitor, this one receives the whole notification
1528 stream as JSON. Either, both or neither may be configured.
1530 Args:
1531 config: Configuration dictionary
1533 Returns:
1534 The URL if configured and non-blank, None otherwise
1535 """
1536 return get_webhook_url(config=config, event="url")
1539def get_webhook_events(config: dict) -> list[str] | None:
1540 """Return the ``app.webhooks.events`` allow-list, or None for "all".
1542 None and a configured list mean different things: no key at all sends
1543 every event, while an explicit (even empty) list sends only what it
1544 names. A non-list value is treated as absent rather than as an empty
1545 filter, so a typo cannot silently mute the endpoint.
1547 Args:
1548 config: Configuration dictionary
1550 Returns:
1551 List of event names, or None when unfiltered
1552 """
1553 value = get_config_value_or_none(config=config, config_path=["app", "webhooks", "events"])
1554 if not isinstance(value, list):
1555 return None
1556 return [str(event).strip() for event in value]
1559def get_webhook_headers(config: dict) -> dict[str, str]:
1560 """Return the extra request headers for ``app.webhooks.url``.
1562 Receivers behind an authenticating proxy need a bearer token or an API
1563 key, which has no other home in the config. Values are credentials and
1564 are never logged.
1566 Args:
1567 config: Configuration dictionary
1569 A key left without a value in YAML parses as None; such an entry is
1570 dropped rather than sent as the string "None".
1572 Returns:
1573 Mapping of header name to value; empty when unconfigured
1574 """
1575 value = get_config_value_or_none(config=config, config_path=["app", "webhooks", "headers"])
1576 if not isinstance(value, dict):
1577 return {}
1578 return {
1579 str(name): str(header_value)
1580 for name, header_value in value.items()
1581 if header_value is not None
1582 }
1585# =============================================================================
1586# Sync Summary Notification Configuration Functions
1587# =============================================================================
1590def get_sync_summary_enabled(config: dict) -> bool:
1591 """Return whether sync summary notifications are enabled.
1593 Args:
1594 config: Configuration dictionary
1596 Returns:
1597 True if sync summary is enabled, False otherwise (default: False)
1598 """
1599 config_path = ["app", "notifications", "sync_summary", "enabled"]
1600 if not traverse_config_path(config=config, config_path=config_path):
1601 return False
1603 value = get_config_value(config=config, config_path=config_path)
1604 return bool(value) if value is not None else False
1607def get_sync_summary_on_success(config: dict) -> bool:
1608 """Return whether to send summary on successful syncs.
1610 Args:
1611 config: Configuration dictionary
1613 Returns:
1614 True if should send on success, False otherwise (default: True)
1615 """
1616 config_path = ["app", "notifications", "sync_summary", "on_success"]
1617 if not traverse_config_path(config=config, config_path=config_path):
1618 return True # Default to True if not configured
1620 value = get_config_value(config=config, config_path=config_path)
1621 return bool(value) if value is not None else True
1624def get_sync_summary_on_error(config: dict) -> bool:
1625 """Return whether to send summary when errors occur.
1627 Args:
1628 config: Configuration dictionary
1630 Returns:
1631 True if should send on error, False otherwise (default: True)
1632 """
1633 config_path = ["app", "notifications", "sync_summary", "on_error"]
1634 if not traverse_config_path(config=config, config_path=config_path):
1635 return True # Default to True if not configured
1637 value = get_config_value(config=config, config_path=config_path)
1638 return bool(value) if value is not None else True
1641def get_sync_summary_min_downloads(config: dict) -> int:
1642 """Return minimum downloads required to trigger notification.
1644 Args:
1645 config: Configuration dictionary
1647 Returns:
1648 Minimum downloads threshold (default: 1)
1649 """
1650 config_path = ["app", "notifications", "sync_summary", "min_downloads"]
1651 if not traverse_config_path(config=config, config_path=config_path):
1652 return 1 # Default to 1 if not configured
1654 value = get_config_value(config=config, config_path=config_path)
1655 return int(value) if value is not None else 1