Coverage for src/config_parser.py: 100%
307 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-08-05 00:26 +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_PHOTOS_DESTINATION,
19 DEFAULT_REQUEST_TIMEOUT_SEC,
20 DEFAULT_RETRY_LOGIN_INTERVAL_SEC,
21 DEFAULT_ROOT_DESTINATION,
22 DEFAULT_SYNC_INTERVAL_SEC,
23 configure_icloudpy_logging,
24 get_logger,
25)
26from src.config_logging import (
27 log_config_debug,
28 log_config_error,
29 log_config_found_info,
30 log_config_not_found_warning,
31 log_invalid_config_value,
32)
33from src.config_utils import (
34 config_path_to_string,
35 get_config_value,
36 get_config_value_or_default,
37 get_config_value_or_none,
38 traverse_config_path,
39)
40from src.filesystem_utils import ensure_directory_exists, join_and_ensure_path
42# Configure icloudpy logging immediately after import
43configure_icloudpy_logging()
45LOGGER = get_logger()
47# Cache for config values to prevent repeated warnings
48_config_warning_cache = set()
51def _log_config_warning_once(config_path: list, message: str) -> None:
52 """Log a configuration warning only once for the given config path.
54 Args:
55 config_path: Configuration path as list
56 message: Warning message to log
57 """
58 config_path_key = config_path_to_string(config_path)
59 if config_path_key not in _config_warning_cache:
60 _config_warning_cache.add(config_path_key)
61 log_config_not_found_warning(config_path, message)
64def clear_config_warning_cache() -> None:
65 """Clear the configuration warning cache.
67 This function is primarily intended for testing purposes to ensure
68 clean test isolation.
69 """
70 _config_warning_cache.clear()
73# =============================================================================
74# String Processing Functions
75# =============================================================================
78def validate_and_strip_username(username: str, config_path: list[str]) -> str | None:
79 """Validate and strip username string.
81 Args:
82 username: Raw username string from config
83 config_path: Config path for error logging
85 Returns:
86 Stripped username if valid, None if empty
87 """
88 username = username.strip()
89 if len(username) == 0:
90 log_config_error(config_path, "username is empty")
91 return None
92 return username
95# =============================================================================
96# Credential Configuration Functions
97# =============================================================================
100def get_username(config: dict) -> str | None:
101 """Get username from config.
103 Args:
104 config: Configuration dictionary
106 Returns:
107 Username string if found and valid, None otherwise
108 """
109 config_path = ["app", "credentials", "username"]
110 if not traverse_config_path(config=config, config_path=config_path):
111 log_config_error(config_path, "username is missing. Please set the username.")
112 return None
114 username = get_config_value(config=config, config_path=config_path)
115 return validate_and_strip_username(username, config_path)
118def get_retry_login_interval(config: dict) -> int:
119 """Return retry login interval from config.
121 Args:
122 config: Configuration dictionary
124 Returns:
125 Retry login interval in seconds
126 """
127 config_path = ["app", "credentials", "retry_login_interval"]
129 if not traverse_config_path(config=config, config_path=config_path):
130 retry_login_interval = DEFAULT_RETRY_LOGIN_INTERVAL_SEC
131 log_config_not_found_warning(
132 config_path,
133 f"not found. Using default {retry_login_interval} seconds ...",
134 )
135 else:
136 retry_login_interval = get_config_value(config=config, config_path=config_path)
137 log_config_found_info(f"Retrying login every {retry_login_interval} seconds.")
139 return retry_login_interval
142def get_region(config: dict) -> str:
143 """Return region from config.
145 Args:
146 config: Configuration dictionary
148 Returns:
149 Region string ('global' or 'china')
150 """
151 config_path = ["app", "region"]
152 region = get_config_value_or_default(
153 config=config, config_path=config_path, default="global",
154 )
156 if region == "global" and not traverse_config_path(
157 config=config, config_path=config_path,
158 ):
159 log_config_not_found_warning(
160 config_path, "not found. Using default value - global ...",
161 )
162 elif region not in ["global", "china"]:
163 log_config_error(
164 config_path,
165 "is invalid. Valid values are - global or china. Using default value - global ...",
166 )
167 region = "global"
169 return region
172# =============================================================================
173# Sync Interval Configuration Functions
174# =============================================================================
177def get_sync_interval(
178 config: dict, config_path: list[str], service_name: str, log_messages: bool = True,
179) -> int:
180 """Get sync interval for a service (drive or photos).
182 Extracted common logic for retrieving sync intervals.
184 Args:
185 config: Configuration dictionary
186 config_path: Path to sync_interval config
187 service_name: Name of service for logging ("drive" or "photos")
188 log_messages: Whether to log informational messages (default: True)
190 Returns:
191 Sync interval in seconds
192 """
193 sync_interval = get_config_value_or_default(
194 config=config,
195 config_path=config_path,
196 default=DEFAULT_SYNC_INTERVAL_SEC,
197 )
199 if log_messages:
200 if sync_interval == DEFAULT_SYNC_INTERVAL_SEC:
201 log_config_not_found_warning(
202 config_path,
203 f"is not found. Using default sync_interval: {sync_interval} seconds ...",
204 )
205 else:
206 log_config_found_info(
207 f"Syncing {service_name} every {sync_interval} seconds.",
208 )
210 return sync_interval
213def get_drive_sync_interval(config: dict, log_messages: bool = True) -> int:
214 """Return drive sync interval from config.
216 Args:
217 config: Configuration dictionary
218 log_messages: Whether to log informational messages (default: True)
220 Returns:
221 Drive sync interval in seconds
222 """
223 config_path = ["drive", "sync_interval"]
224 return get_sync_interval(
225 config=config,
226 config_path=config_path,
227 service_name="drive",
228 log_messages=log_messages,
229 )
232def get_drive_request_timeout(config: dict) -> int:
233 """Return drive request timeout from config.
235 Args:
236 config: Configuration dictionary
238 Returns:
239 Request timeout in seconds (default: DEFAULT_REQUEST_TIMEOUT_SEC)
240 """
241 config_path = ["drive", "request_timeout"]
242 return get_config_value_or_default(
243 config=config,
244 config_path=config_path,
245 default=DEFAULT_REQUEST_TIMEOUT_SEC,
246 )
249def get_photos_sync_interval(config: dict, log_messages: bool = True) -> int:
250 """Return photos sync interval from config.
252 Args:
253 config: Configuration dictionary
254 log_messages: Whether to log informational messages (default: True)
256 Returns:
257 Photos sync interval in seconds
258 """
259 config_path = ["photos", "sync_interval"]
260 return get_sync_interval(
261 config=config,
262 config_path=config_path,
263 service_name="photos",
264 log_messages=log_messages,
265 )
268# =============================================================================
269# Thread Configuration Functions
270# =============================================================================
273def calculate_default_max_threads() -> int:
274 """Calculate default maximum threads based on CPU cores.
276 Returns:
277 Default max threads (min of CPU count and 8)
278 """
279 return min(multiprocessing.cpu_count(), 8)
282def parse_max_threads_value(max_threads_config: Any, default_max_threads: int) -> int:
283 """Parse and validate max_threads configuration value.
285 Args:
286 max_threads_config: Raw config value (string "auto" or integer)
287 default_max_threads: Default value to use
289 Returns:
290 Validated max threads value (capped at 16)
291 """
292 # Handle "auto" value
293 if isinstance(max_threads_config, str) and max_threads_config.lower() == "auto":
294 max_threads = default_max_threads
295 log_config_found_info(
296 f"Using automatic thread count: {max_threads} threads (based on CPU cores).",
297 )
298 elif isinstance(max_threads_config, int) and max_threads_config >= 1:
299 max_threads = min(
300 max_threads_config, 16,
301 ) # Cap at 16 to avoid overwhelming servers
302 log_config_found_info(f"Using configured max_threads: {max_threads}.")
303 else:
304 log_invalid_config_value(
305 ["app", "max_threads"],
306 max_threads_config,
307 "'auto' or integer >= 1",
308 )
309 max_threads = default_max_threads
311 return max_threads
314def get_web_ui_enabled(config: dict) -> bool:
315 """Return whether the embedded web UI should start on container boot.
317 Default: **False** — opt-in. Existing mandarons installs see no
318 behaviour change; only users who explicitly set
319 ``app.web_ui.enabled: true`` open the port.
320 """
321 return bool(
322 get_config_value_or_default(
323 config=config,
324 config_path=["app", "web_ui", "enabled"],
325 default=False,
326 ),
327 )
330def get_web_ui_host(config: dict) -> str:
331 """Web UI bind address.
333 Default ``127.0.0.1`` — the web UI accepts the user's Apple ID
334 password on POST /auth/password with no built-in authentication and
335 no CSRF token (the feature assumes a reverse-proxy trust boundary
336 in front of it). Defaulting to loopback means the credential-
337 accepting form is never exposed to LAN/public on a vanilla install.
338 Users who run behind a reverse proxy or want explicit LAN exposure
339 set ``app.web_ui.host: 0.0.0.0`` consciously.
340 """
341 return str(
342 get_config_value_or_default(
343 config=config,
344 config_path=["app", "web_ui", "host"],
345 default="127.0.0.1",
346 ),
347 )
350def get_web_ui_port(config: dict) -> int:
351 """Web UI TCP port. Default ``8080``. Coexists with mandarons' legacy
352 ``EXPOSE 80`` (unused) — no port collision."""
353 return int(
354 get_config_value_or_default(
355 config=config,
356 config_path=["app", "web_ui", "port"],
357 default=8080,
358 ),
359 )
362def get_web_ui_public_url(config: dict) -> str | None:
363 """Public-facing URL for the web UI, used in notifications.
365 The daemon only knows its local bind host:port; the user-facing URL
366 (e.g. ``https://icloud.zosia.io`` behind a reverse proxy) must be
367 declared explicitly. When unset, notifications fall back to
368 ``http://{host}:{port}`` and log a one-shot warning at startup.
369 """
370 value = get_config_value_or_default(
371 config=config,
372 config_path=["app", "web_ui", "public_url"],
373 default=None,
374 )
375 if value is None:
376 return None
377 return str(value).rstrip("/")
380def get_trust_expiry_warn_days(config: dict) -> int:
381 """Warn this many days before Apple's trust cookie expires.
383 Default 7. The check + notification fires once per cookie-lifetime
384 when ``trust_days_remaining`` first drops below this threshold so
385 the user can tap refresh-trust *before* the sync loop hits a
386 failed-auth state.
387 """
388 return int(
389 get_config_value_or_default(
390 config=config,
391 config_path=["app", "trust_expiry_warn_days"],
392 default=7,
393 ),
394 )
397def get_app_max_threads(config: dict) -> int:
398 """Return app-level max threads from config with support for 'auto' value.
400 Args:
401 config: Configuration dictionary
403 Returns:
404 Maximum number of threads for parallel operations
405 """
406 default_max_threads = calculate_default_max_threads()
407 config_path = ["app", "max_threads"]
409 if not traverse_config_path(config=config, config_path=config_path):
410 log_config_debug(
411 f"max_threads is not found in {config_path_to_string(config_path=config_path)}. "
412 f"Using default max_threads: {default_max_threads} (auto) ...",
413 )
414 return default_max_threads
416 max_threads_config = get_config_value(config=config, config_path=config_path)
417 return parse_max_threads_value(max_threads_config, default_max_threads)
420def get_mount_marker_filename(config: dict) -> str:
421 """Return the filename used as the mount-failsafe marker (default `.mounted`).
423 The marker is the empty/sentinel file the user touches in a destination
424 directory to assert "this path is correctly bind-mounted". The mount
425 checker (``sync._check_mount_marker``) refuses to sync into a destination
426 that lacks this file when ``require_mount_marker`` is enabled.
428 Args:
429 config: Configuration dictionary
431 Returns:
432 Filename string (relative to each destination directory).
433 """
434 config_path = ["app", "mount_marker_filename"]
435 return str(get_config_value_or_default(config=config, config_path=config_path, default=".mounted"))
438def get_usage_tracking_enabled(config: dict) -> bool:
439 """Get usage tracking enabled setting from configuration.
441 Args:
442 config: Configuration dictionary
444 Returns:
445 True if usage tracking is enabled (default), False if disabled
446 """
447 config_path = ["app", "usage_tracking", "enabled"]
448 if not traverse_config_path(config=config, config_path=config_path):
449 return True # Default to enabled if not configured
451 value = get_config_value(config=config, config_path=config_path)
452 if isinstance(value, bool):
453 return value
455 # Handle string values for backwards compatibility
456 if isinstance(value, str):
457 return value.lower() not in ("false", "no", "0", "disabled", "off")
459 # Default to enabled for any other type
460 return True
463# =============================================================================
464# Root Destination Functions
465# =============================================================================
468def get_root_destination_path(config: dict) -> str:
469 """Get root destination path from config without creating directory.
471 Args:
472 config: Configuration dictionary
474 Returns:
475 Root destination path string
476 """
477 config_path = ["app", "root"]
478 root_destination = get_config_value_or_default(
479 config=config,
480 config_path=config_path,
481 default=DEFAULT_ROOT_DESTINATION,
482 )
484 if not traverse_config_path(config=config, config_path=config_path):
485 log_config_not_found_warning(
486 config_path,
487 f"root destination is missing. Using default root destination: {root_destination}",
488 )
490 return root_destination
493def prepare_root_destination(config: dict) -> str:
494 """Prepare root destination by creating directory if needed.
496 Args:
497 config: Configuration dictionary
499 Returns:
500 Absolute path to root destination directory
501 """
502 log_config_debug("Checking root destination ...")
503 root_destination = get_root_destination_path(config)
504 return ensure_directory_exists(root_destination)
507# =============================================================================
508# Drive Configuration Functions
509# =============================================================================
512def get_drive_destination_path(config: dict) -> str:
513 """Get drive destination path from config without creating directory.
515 Args:
516 config: Configuration dictionary
518 Returns:
519 Drive destination path string
520 """
521 config_path = ["drive", "destination"]
522 drive_destination = get_config_value_or_default(
523 config=config,
524 config_path=config_path,
525 default=DEFAULT_DRIVE_DESTINATION,
526 )
528 if not traverse_config_path(config=config, config_path=config_path):
529 log_config_not_found_warning(
530 config_path,
531 f"destination is missing. Using default drive destination: {drive_destination}.",
532 )
534 return drive_destination
537def prepare_drive_destination(config: dict) -> str:
538 """Prepare drive destination path by creating directory if needed.
540 Args:
541 config: Configuration dictionary
543 Returns:
544 Absolute path to drive destination directory
545 """
546 log_config_debug("Checking drive destination ...")
547 root_path = prepare_root_destination(config=config)
548 drive_destination = get_drive_destination_path(config)
549 return join_and_ensure_path(root_path, drive_destination)
552def get_drive_remove_obsolete(config: dict) -> bool:
553 """Return drive remove obsolete flag from config.
555 Args:
556 config: Configuration dictionary
558 Returns:
559 True if obsolete files should be removed, False otherwise
560 """
561 config_path = ["drive", "remove_obsolete"]
562 drive_remove_obsolete = get_config_value_or_default(
563 config=config, config_path=config_path, default=False,
564 )
566 if not drive_remove_obsolete:
567 _log_config_warning_once(
568 config_path,
569 "remove_obsolete is not found. Not removing the obsolete files and folders.",
570 )
571 else:
572 log_config_debug(
573 f"{'R' if drive_remove_obsolete else 'Not R'}emoving obsolete files and folders ...",
574 )
576 return drive_remove_obsolete
579def get_drive_require_mount_marker(config: dict) -> bool:
580 """Return whether Drive sync requires the mount-failsafe marker file.
582 When True, ``sync._check_mount_marker`` refuses to start a Drive sync
583 until the marker file (see ``get_mount_marker_filename``) exists in
584 the Drive destination directory. Protects against silent bind-mount
585 failures dumping iCloud Drive content into the wrong place.
587 Default: False (preserves historical behaviour — no breaking change).
589 Args:
590 config: Configuration dictionary
592 Returns:
593 True if the marker is required before each Drive sync.
594 """
595 config_path = ["drive", "require_mount_marker"]
596 return bool(get_config_value_or_default(config=config, config_path=config_path, default=False))
599# =============================================================================
600# Photos Configuration Functions
601# =============================================================================
604def get_photos_enumeration_chunk_size(config: dict | None) -> int:
605 """Tasks to buffer before draining via execute_parallel_downloads.
607 Smaller = lower peak memory, more per-chunk HTTP setup overhead.
608 Larger = higher peak memory, fewer chunks. Default 1000 keeps
609 resident set at ~10 MB on typical libraries while still amortising
610 connection setup. Tested empirically on a 111K-photo library:
611 1000 sustained < 1 GB resident through the full enumeration.
613 Args:
614 config: Configuration dictionary (None falls back to default).
616 Returns:
617 Positive integer chunk size, defaulting to 1000.
618 """
619 raw = get_config_value_or_default(
620 config=config or {},
621 config_path=["photos", "enumeration_chunk_size"],
622 default=DEFAULT_ENUMERATION_CHUNK_SIZE,
623 )
624 try:
625 value = int(raw)
626 except (TypeError, ValueError):
627 return DEFAULT_ENUMERATION_CHUNK_SIZE
628 return value if value > 0 else DEFAULT_ENUMERATION_CHUNK_SIZE
631def get_photos_destination_path(config: dict) -> str:
632 """Get photos destination path from config without creating directory.
634 Args:
635 config: Configuration dictionary
637 Returns:
638 Photos destination path string
639 """
640 config_path = ["photos", "destination"]
641 photos_destination = get_config_value_or_default(
642 config=config,
643 config_path=config_path,
644 default=DEFAULT_PHOTOS_DESTINATION,
645 )
647 if not traverse_config_path(config=config, config_path=config_path):
648 log_config_not_found_warning(
649 config_path,
650 f"destination is missing. Using default photos destination: {config_path_to_string(config_path)}",
651 )
653 return photos_destination
656def prepare_photos_destination(config: dict) -> str:
657 """Prepare photos destination path by creating directory if needed.
659 Args:
660 config: Configuration dictionary
662 Returns:
663 Absolute path to photos destination directory
664 """
665 log_config_debug("Checking photos destination ...")
666 root_path = prepare_root_destination(config=config)
667 photos_destination = get_photos_destination_path(config)
668 return join_and_ensure_path(root_path, photos_destination)
671def get_photos_all_albums(config: dict) -> bool:
672 """Return flag to download all albums from config.
674 Args:
675 config: Configuration dictionary
677 Returns:
678 True if all albums should be synced, False otherwise
679 """
680 config_path = ["photos", "all_albums"]
681 download_all = get_config_value_or_default(
682 config=config, config_path=config_path, default=False,
683 )
685 if download_all:
686 log_config_found_info("Syncing all albums.")
688 return download_all
691def get_photos_use_hardlinks(config: dict, log_messages: bool = True) -> bool:
692 """Return flag to use hard links for duplicate photos from config.
694 Args:
695 config: Configuration dictionary
696 log_messages: Whether to log informational messages (default: True)
698 Returns:
699 True if hard links should be used, False otherwise
700 """
701 config_path = ["photos", "use_hardlinks"]
702 use_hardlinks = get_config_value_or_default(
703 config=config, config_path=config_path, default=False,
704 )
706 if use_hardlinks and log_messages:
707 log_config_found_info("Using hard links for duplicate photos.")
709 return use_hardlinks
712def get_photos_remove_obsolete(config: dict) -> bool:
713 """Return photos remove obsolete flag from config.
715 Args:
716 config: Configuration dictionary
718 Returns:
719 True if obsolete files should be removed, False otherwise
720 """
721 config_path = ["photos", "remove_obsolete"]
722 photos_remove_obsolete = get_config_value_or_default(
723 config=config, config_path=config_path, default=False,
724 )
726 if not photos_remove_obsolete:
727 _log_config_warning_once(
728 config_path,
729 "remove_obsolete is not found. Not removing the obsolete files and folders.",
730 )
731 else:
732 log_config_debug(
733 f"{'R' if photos_remove_obsolete else 'Not R'}emoving obsolete files and folders ...",
734 )
736 return photos_remove_obsolete
739def get_photos_require_mount_marker(config: dict) -> bool:
740 """Return whether Photos sync requires the mount-failsafe marker file.
742 When True, ``sync._check_mount_marker`` refuses to start a Photos sync
743 until the marker file (see ``get_mount_marker_filename``) exists in
744 the Photos destination directory. Protects against silent bind-mount
745 failures dumping the entire iCloud photo library into the wrong place.
747 Default: False (preserves historical behaviour — no breaking change).
749 Args:
750 config: Configuration dictionary
752 Returns:
753 True if the marker is required before each Photos sync.
754 """
755 config_path = ["photos", "require_mount_marker"]
756 return bool(get_config_value_or_default(config=config, config_path=config_path, default=False))
759def get_photos_folder_format(config: dict) -> str | None:
760 """Return filename format or None.
762 Args:
763 config: Configuration dictionary
765 Returns:
766 Folder format string if configured, None otherwise
767 """
768 config_path = ["photos", "folder_format"]
769 fmt = get_config_value_or_none(config=config, config_path=config_path)
771 if fmt:
772 log_config_found_info(f"Using format {fmt}.")
774 return fmt
777# =============================================================================
778# Photos Filter Configuration Functions
779# =============================================================================
782def validate_file_sizes(file_sizes: list[str]) -> list[str]:
783 """Validate and filter file sizes against valid options.
785 Accepts any key in ``PhotoAsset.PHOTO_VERSION_LOOKUP``, including the
786 ``live_video_*`` keys: add ``live_video_original`` to ``file_sizes`` to
787 download the paired ``.mov`` of a Live Photo (and ``live_video_medium`` /
788 ``live_video_thumb`` for smaller variants). Photos that aren't Live Photos
789 simply don't have those versions and are skipped.
791 Args:
792 file_sizes: List of file size strings to validate
794 Returns:
795 List of valid file sizes (defaults to ["original"] if all invalid)
796 """
797 valid_file_sizes = list(PhotoAsset.PHOTO_VERSION_LOOKUP.keys())
798 validated_sizes = []
800 for file_size in file_sizes:
801 if file_size not in valid_file_sizes:
802 log_invalid_config_value(
803 ["photos", "filters", "file_sizes"],
804 file_size,
805 ",".join(valid_file_sizes),
806 )
807 else:
808 validated_sizes.append(file_size)
810 return validated_sizes if validated_sizes else ["original"]
813def get_photos_library_destinations(config: dict) -> dict[str, str]:
814 """Get per-library destination subdirectory mapping from photos config.
816 Optional config block (under top-level ``photos``):
818 .. code-block:: yaml
820 photos:
821 destination: photos
822 library_destinations:
823 PrimarySync: personal
824 SharedLibrary: shared
826 When set, photos from each library are written to
827 ``<photos.destination>/<library_destinations[library]>/...`` instead of
828 sharing one destination tree. When unset (the default), all libraries
829 share the single ``photos.destination`` path — preserving the historical
830 behaviour of mandarons/icloud-docker.
832 Returns:
833 Dict mapping library name → subdirectory relative to ``photos.destination``.
834 Returns ``{}`` if not configured (backward-compatible default).
835 """
836 config_path = ["photos", "library_destinations"]
837 mapping = get_config_value_or_none(config=config, config_path=config_path)
838 if not mapping or not isinstance(mapping, dict):
839 return {}
840 return {str(k): str(v) for k, v in mapping.items()}
843def get_photos_libraries_filter(config: dict, base_config_path: list[str]) -> list[str] | None:
844 """Get libraries filter from photos config.
846 Args:
847 config: Configuration dictionary
848 base_config_path: Base path to filters section
850 Returns:
851 List of library names if configured, None otherwise
852 """
853 config_path = base_config_path + ["libraries"]
854 libraries = get_config_value_or_none(config=config, config_path=config_path)
856 if not libraries or len(libraries) == 0:
857 log_config_not_found_warning(
858 config_path, "not found. Downloading all libraries ...",
859 )
860 return None
862 return libraries
865def get_photos_albums_filter(
866 config: dict, base_config_path: list[str],
867) -> list[str] | None:
868 """Get albums filter from photos config.
870 Args:
871 config: Configuration dictionary
872 base_config_path: Base path to filters section
874 Returns:
875 List of album names if configured, None otherwise
876 """
877 config_path = base_config_path + ["albums"]
878 albums = get_config_value_or_none(config=config, config_path=config_path)
880 if not albums or len(albums) == 0:
881 log_config_not_found_warning(
882 config_path, "not found. Downloading all albums ...",
883 )
884 return None
886 return albums
889def get_photos_file_sizes_filter(
890 config: dict, base_config_path: list[str],
891) -> list[str]:
892 """Get file sizes filter from photos config.
894 Args:
895 config: Configuration dictionary
896 base_config_path: Base path to filters section
898 Returns:
899 List of file size options (defaults to ["original"])
900 """
901 config_path = base_config_path + ["file_sizes"]
903 if not traverse_config_path(config=config, config_path=config_path):
904 log_config_not_found_warning(
905 config_path, "not found. Downloading original size photos ...",
906 )
907 return ["original"]
909 file_sizes = get_config_value(config=config, config_path=config_path)
910 return validate_file_sizes(file_sizes)
913def get_photos_extensions_filter(
914 config: dict, base_config_path: list[str],
915) -> list[str] | None:
916 """Get extensions filter from photos config.
918 Args:
919 config: Configuration dictionary
920 base_config_path: Base path to filters section
922 Returns:
923 List of file extensions if configured, None otherwise
924 """
925 config_path = base_config_path + ["extensions"]
926 extensions = get_config_value_or_none(config=config, config_path=config_path)
928 if not extensions or len(extensions) == 0:
929 log_config_not_found_warning(
930 config_path, "not found. Downloading all extensions ...",
931 )
932 return None
934 return extensions
937def get_photos_filters(config: dict) -> dict[str, Any]:
938 """Return photos filters from config.
940 Args:
941 config: Configuration dictionary
943 Returns:
944 Dictionary containing filter configuration for photos
945 """
946 photos_filters = {
947 "libraries": None,
948 "albums": None,
949 "file_sizes": ["original"],
950 "extensions": None,
951 }
953 base_config_path = ["photos", "filters"]
955 # Check for filters section existence
956 if not traverse_config_path(config=config, config_path=base_config_path):
957 log_config_not_found_warning(
958 base_config_path,
959 "not found. Downloading all libraries and albums with original size ...",
960 )
961 return photos_filters
963 # Parse individual filter components
964 photos_filters["libraries"] = get_photos_libraries_filter(config, base_config_path)
965 photos_filters["albums"] = get_photos_albums_filter(config, base_config_path)
966 photos_filters["file_sizes"] = get_photos_file_sizes_filter(
967 config, base_config_path,
968 )
969 photos_filters["extensions"] = get_photos_extensions_filter(
970 config, base_config_path,
971 )
973 return photos_filters
976# =============================================================================
977# SMTP Configuration Functions
978# =============================================================================
981def get_smtp_config_value(
982 config: dict, key: str, warn_if_missing: bool = True,
983) -> str | None:
984 """Get SMTP configuration value with optional warning.
986 Common helper for SMTP config retrieval to reduce duplication.
988 Args:
989 config: Configuration dictionary
990 key: SMTP config key name
991 warn_if_missing: Whether to log warning if not found
993 Returns:
994 Config value if found, None otherwise
995 """
996 config_path = ["app", "smtp", key]
997 value = get_config_value_or_none(config=config, config_path=config_path)
999 if value is None and warn_if_missing:
1000 log_config_not_found_warning(config_path, f"{key} is not found.")
1002 return value
1005def get_smtp_email(config: dict) -> str | None:
1006 """Return smtp from email from config.
1008 Args:
1009 config: Configuration dictionary
1011 Returns:
1012 SMTP email address if configured, None otherwise
1013 """
1014 return get_smtp_config_value(config, "email", warn_if_missing=False)
1017def get_smtp_username(config: dict) -> str | None:
1018 """Return smtp username from the config, if set.
1020 Args:
1021 config: Configuration dictionary
1023 Returns:
1024 SMTP username if configured, None otherwise
1025 """
1026 return get_smtp_config_value(config, "username", warn_if_missing=False)
1029def get_smtp_to_email(config: dict) -> str | None:
1030 """Return smtp to email from config, defaults to from email.
1032 Args:
1033 config: Configuration dictionary
1035 Returns:
1036 SMTP 'to' email address, falling back to 'from' email if not specified
1037 """
1038 to_email = get_smtp_config_value(config, "to", warn_if_missing=False)
1039 return to_email if to_email else get_smtp_email(config=config)
1042def get_smtp_password(config: dict) -> str | None:
1043 """Return smtp password from config.
1045 Args:
1046 config: Configuration dictionary
1048 Returns:
1049 SMTP password if configured, None otherwise
1050 """
1051 return get_smtp_config_value(config, "password", warn_if_missing=True)
1054def get_smtp_host(config: dict) -> str | None:
1055 """Return smtp host from config.
1057 Args:
1058 config: Configuration dictionary
1060 Returns:
1061 SMTP host if configured, None otherwise
1062 """
1063 return get_smtp_config_value(config, "host", warn_if_missing=True)
1066def get_smtp_port(config: dict) -> int | None:
1067 """Return smtp port from config.
1069 Args:
1070 config: Configuration dictionary
1072 Returns:
1073 SMTP port if configured, None otherwise
1074 """
1075 return get_smtp_config_value(config, "port", warn_if_missing=True) # type: ignore[return-value]
1078def get_smtp_no_tls(config: dict) -> bool:
1079 """Return smtp no_tls flag from config.
1081 Args:
1082 config: Configuration dictionary
1084 Returns:
1085 True if TLS should be disabled, False otherwise
1086 """
1087 no_tls = get_smtp_config_value(config, "no_tls", warn_if_missing=True)
1088 return no_tls if no_tls is not None else False # type: ignore[return-value]
1091# =============================================================================
1092# Notification Service Configuration Functions
1093# =============================================================================
1096def get_notification_config_value(config: dict, service: str, key: str) -> str | None:
1097 """Get notification service configuration value.
1099 Common helper for notification service config retrieval.
1101 Args:
1102 config: Configuration dictionary
1103 service: Service name (telegram, discord, pushover)
1104 key: Config key name
1106 Returns:
1107 Config value if found, None otherwise
1108 """
1109 config_path = ["app", service, key]
1110 value = get_config_value_or_none(config=config, config_path=config_path)
1112 if value is None:
1113 log_config_not_found_warning(config_path, f"{key} is not found.")
1115 return value
1118def get_telegram_bot_token(config: dict) -> str | None:
1119 """Return telegram bot token from config.
1121 Args:
1122 config: Configuration dictionary
1124 Returns:
1125 Telegram bot token if configured, None otherwise
1126 """
1127 return get_notification_config_value(config, "telegram", "bot_token")
1130def get_telegram_chat_id(config: dict) -> str | None:
1131 """Return telegram chat id from config.
1133 Args:
1134 config: Configuration dictionary
1136 Returns:
1137 Telegram chat ID if configured, None otherwise
1138 """
1139 return get_notification_config_value(config, "telegram", "chat_id")
1142def get_discord_webhook_url(config: dict) -> str | None:
1143 """Return discord webhook_url from config.
1145 Args:
1146 config: Configuration dictionary
1148 Returns:
1149 Discord webhook URL if configured, None otherwise
1150 """
1151 return get_notification_config_value(config, "discord", "webhook_url")
1154def get_discord_username(config: dict) -> str | None:
1155 """Return discord username from config.
1157 Args:
1158 config: Configuration dictionary
1160 Returns:
1161 Discord username if configured, None otherwise
1162 """
1163 return get_notification_config_value(config, "discord", "username")
1166def get_pushover_user_key(config: dict) -> str | None:
1167 """Return Pushover user key from config.
1169 Args:
1170 config: Configuration dictionary
1172 Returns:
1173 Pushover user key if configured, None otherwise
1174 """
1175 return get_notification_config_value(config, "pushover", "user_key")
1178def get_pushover_api_token(config: dict) -> str | None:
1179 """Return Pushover API token from config.
1181 Args:
1182 config: Configuration dictionary
1184 Returns:
1185 Pushover API token if configured, None otherwise
1186 """
1187 return get_notification_config_value(config, "pushover", "api_token")
1190def get_pushover_notification_priority(config: dict) -> int | None:
1191 """Return Pushover notification priority from config.
1193 Args:
1194 config: Configuration dictionary
1196 Returns:
1197 Pushover notification priority if configured, None otherwise
1198 """
1199 config_path = ["app", "pushover", "priority"]
1200 return get_config_value_or_none(config=config, config_path=config_path)
1203# =============================================================================
1204# Sync Summary Notification Configuration Functions
1205# =============================================================================
1208def get_sync_summary_enabled(config: dict) -> bool:
1209 """Return whether sync summary notifications are enabled.
1211 Args:
1212 config: Configuration dictionary
1214 Returns:
1215 True if sync summary is enabled, False otherwise (default: False)
1216 """
1217 config_path = ["app", "notifications", "sync_summary", "enabled"]
1218 if not traverse_config_path(config=config, config_path=config_path):
1219 return False
1221 value = get_config_value(config=config, config_path=config_path)
1222 return bool(value) if value is not None else False
1225def get_sync_summary_on_success(config: dict) -> bool:
1226 """Return whether to send summary on successful syncs.
1228 Args:
1229 config: Configuration dictionary
1231 Returns:
1232 True if should send on success, False otherwise (default: True)
1233 """
1234 config_path = ["app", "notifications", "sync_summary", "on_success"]
1235 if not traverse_config_path(config=config, config_path=config_path):
1236 return True # Default to True if not configured
1238 value = get_config_value(config=config, config_path=config_path)
1239 return bool(value) if value is not None else True
1242def get_sync_summary_on_error(config: dict) -> bool:
1243 """Return whether to send summary when errors occur.
1245 Args:
1246 config: Configuration dictionary
1248 Returns:
1249 True if should send on error, False otherwise (default: True)
1250 """
1251 config_path = ["app", "notifications", "sync_summary", "on_error"]
1252 if not traverse_config_path(config=config, config_path=config_path):
1253 return True # Default to True if not configured
1255 value = get_config_value(config=config, config_path=config_path)
1256 return bool(value) if value is not None else True
1259def get_sync_summary_min_downloads(config: dict) -> int:
1260 """Return minimum downloads required to trigger notification.
1262 Args:
1263 config: Configuration dictionary
1265 Returns:
1266 Minimum downloads threshold (default: 1)
1267 """
1268 config_path = ["app", "notifications", "sync_summary", "min_downloads"]
1269 if not traverse_config_path(config=config, config_path=config_path):
1270 return 1 # Default to 1 if not configured
1272 value = get_config_value(config=config, config_path=config_path)
1273 return int(value) if value is not None else 1