Coverage for src/config_parser.py: 100%
266 statements
« prev ^ index » next coverage.py v7.10.7, created at 2025-10-16 04:41 +0000
« prev ^ index » next coverage.py v7.10.7, created at 2025-10-16 04:41 +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_PHOTOS_DESTINATION,
18 DEFAULT_RETRY_LOGIN_INTERVAL_SEC,
19 DEFAULT_ROOT_DESTINATION,
20 DEFAULT_SYNC_INTERVAL_SEC,
21 configure_icloudpy_logging,
22 get_logger,
23)
24from src.config_logging import (
25 log_config_debug,
26 log_config_error,
27 log_config_found_info,
28 log_config_not_found_warning,
29 log_invalid_config_value,
30)
31from src.config_utils import (
32 config_path_to_string,
33 get_config_value,
34 get_config_value_or_default,
35 get_config_value_or_none,
36 traverse_config_path,
37)
38from src.filesystem_utils import ensure_directory_exists, join_and_ensure_path
40# Configure icloudpy logging immediately after import
41configure_icloudpy_logging()
43LOGGER = get_logger()
45# Cache for config values to prevent repeated warnings
46_config_warning_cache = set()
49def _log_config_warning_once(config_path: list, message: str) -> None:
50 """Log a configuration warning only once for the given config path.
52 Args:
53 config_path: Configuration path as list
54 message: Warning message to log
55 """
56 config_path_key = config_path_to_string(config_path)
57 if config_path_key not in _config_warning_cache:
58 _config_warning_cache.add(config_path_key)
59 log_config_not_found_warning(config_path, message)
62def clear_config_warning_cache() -> None:
63 """Clear the configuration warning cache.
65 This function is primarily intended for testing purposes to ensure
66 clean test isolation.
67 """
68 _config_warning_cache.clear()
71# =============================================================================
72# String Processing Functions
73# =============================================================================
76def validate_and_strip_username(username: str, config_path: list[str]) -> str | None:
77 """Validate and strip username string.
79 Args:
80 username: Raw username string from config
81 config_path: Config path for error logging
83 Returns:
84 Stripped username if valid, None if empty
85 """
86 username = username.strip()
87 if len(username) == 0:
88 log_config_error(config_path, "username is empty")
89 return None
90 return username
93# =============================================================================
94# Credential Configuration Functions
95# =============================================================================
98def get_username(config: dict) -> str | None:
99 """Get username from config.
101 Args:
102 config: Configuration dictionary
104 Returns:
105 Username string if found and valid, None otherwise
106 """
107 config_path = ["app", "credentials", "username"]
108 if not traverse_config_path(config=config, config_path=config_path):
109 log_config_error(config_path, "username is missing. Please set the username.")
110 return None
112 username = get_config_value(config=config, config_path=config_path)
113 return validate_and_strip_username(username, config_path)
116def get_retry_login_interval(config: dict) -> int:
117 """Return retry login interval from config.
119 Args:
120 config: Configuration dictionary
122 Returns:
123 Retry login interval in seconds
124 """
125 config_path = ["app", "credentials", "retry_login_interval"]
127 if not traverse_config_path(config=config, config_path=config_path):
128 retry_login_interval = DEFAULT_RETRY_LOGIN_INTERVAL_SEC
129 log_config_not_found_warning(
130 config_path,
131 f"not found. Using default {retry_login_interval} seconds ...",
132 )
133 else:
134 retry_login_interval = get_config_value(config=config, config_path=config_path)
135 log_config_found_info(f"Retrying login every {retry_login_interval} seconds.")
137 return retry_login_interval
140def get_region(config: dict) -> str:
141 """Return region from config.
143 Args:
144 config: Configuration dictionary
146 Returns:
147 Region string ('global' or 'china')
148 """
149 config_path = ["app", "region"]
150 region = get_config_value_or_default(config=config, config_path=config_path, default="global")
152 if region == "global" and not traverse_config_path(config=config, config_path=config_path):
153 log_config_not_found_warning(config_path, "not found. Using default value - global ...")
154 elif region not in ["global", "china"]:
155 log_config_error(
156 config_path,
157 "is invalid. Valid values are - global or china. Using default value - global ...",
158 )
159 region = "global"
161 return region
164# =============================================================================
165# Sync Interval Configuration Functions
166# =============================================================================
169def get_sync_interval(config: dict, config_path: list[str], service_name: str, log_messages: bool = True) -> int:
170 """Get sync interval for a service (drive or photos).
172 Extracted common logic for retrieving sync intervals.
174 Args:
175 config: Configuration dictionary
176 config_path: Path to sync_interval config
177 service_name: Name of service for logging ("drive" or "photos")
178 log_messages: Whether to log informational messages (default: True)
180 Returns:
181 Sync interval in seconds
182 """
183 sync_interval = get_config_value_or_default(
184 config=config,
185 config_path=config_path,
186 default=DEFAULT_SYNC_INTERVAL_SEC,
187 )
189 if log_messages:
190 if sync_interval == DEFAULT_SYNC_INTERVAL_SEC:
191 log_config_not_found_warning(
192 config_path,
193 f"is not found. Using default sync_interval: {sync_interval} seconds ...",
194 )
195 else:
196 log_config_found_info(f"Syncing {service_name} every {sync_interval} seconds.")
198 return sync_interval
201def get_drive_sync_interval(config: dict, log_messages: bool = True) -> int:
202 """Return drive sync interval from config.
204 Args:
205 config: Configuration dictionary
206 log_messages: Whether to log informational messages (default: True)
208 Returns:
209 Drive sync interval in seconds
210 """
211 config_path = ["drive", "sync_interval"]
212 return get_sync_interval(config=config, config_path=config_path, service_name="drive", log_messages=log_messages)
215def get_photos_sync_interval(config: dict, log_messages: bool = True) -> int:
216 """Return photos sync interval from config.
218 Args:
219 config: Configuration dictionary
220 log_messages: Whether to log informational messages (default: True)
222 Returns:
223 Photos sync interval in seconds
224 """
225 config_path = ["photos", "sync_interval"]
226 return get_sync_interval(config=config, config_path=config_path, service_name="photos", log_messages=log_messages)
229# =============================================================================
230# Thread Configuration Functions
231# =============================================================================
234def calculate_default_max_threads() -> int:
235 """Calculate default maximum threads based on CPU cores.
237 Returns:
238 Default max threads (min of CPU count and 8)
239 """
240 return min(multiprocessing.cpu_count(), 8)
243def parse_max_threads_value(max_threads_config: Any, default_max_threads: int) -> int:
244 """Parse and validate max_threads configuration value.
246 Args:
247 max_threads_config: Raw config value (string "auto" or integer)
248 default_max_threads: Default value to use
250 Returns:
251 Validated max threads value (capped at 16)
252 """
253 # Handle "auto" value
254 if isinstance(max_threads_config, str) and max_threads_config.lower() == "auto":
255 max_threads = default_max_threads
256 log_config_found_info(f"Using automatic thread count: {max_threads} threads (based on CPU cores).")
257 elif isinstance(max_threads_config, int) and max_threads_config >= 1:
258 max_threads = min(max_threads_config, 16) # Cap at 16 to avoid overwhelming servers
259 log_config_found_info(f"Using configured max_threads: {max_threads}.")
260 else:
261 log_invalid_config_value(
262 ["app", "max_threads"],
263 max_threads_config,
264 "'auto' or integer >= 1",
265 )
266 max_threads = default_max_threads
268 return max_threads
271def get_app_max_threads(config: dict) -> int:
272 """Return app-level max threads from config with support for 'auto' value.
274 Args:
275 config: Configuration dictionary
277 Returns:
278 Maximum number of threads for parallel operations
279 """
280 default_max_threads = calculate_default_max_threads()
281 config_path = ["app", "max_threads"]
283 if not traverse_config_path(config=config, config_path=config_path):
284 log_config_debug(
285 f"max_threads is not found in {config_path_to_string(config_path=config_path)}. "
286 f"Using default max_threads: {default_max_threads} (auto) ...",
287 )
288 return default_max_threads
290 max_threads_config = get_config_value(config=config, config_path=config_path)
291 return parse_max_threads_value(max_threads_config, default_max_threads)
294def get_usage_tracking_enabled(config: dict) -> bool:
295 """Get usage tracking enabled setting from configuration.
297 Args:
298 config: Configuration dictionary
300 Returns:
301 True if usage tracking is enabled (default), False if disabled
302 """
303 config_path = ["app", "usage_tracking", "enabled"]
304 if not traverse_config_path(config=config, config_path=config_path):
305 return True # Default to enabled if not configured
307 value = get_config_value(config=config, config_path=config_path)
308 if isinstance(value, bool):
309 return value
311 # Handle string values for backwards compatibility
312 if isinstance(value, str):
313 return value.lower() not in ("false", "no", "0", "disabled", "off")
315 # Default to enabled for any other type
316 return True
319# =============================================================================
320# Root Destination Functions
321# =============================================================================
324def get_root_destination_path(config: dict) -> str:
325 """Get root destination path from config without creating directory.
327 Args:
328 config: Configuration dictionary
330 Returns:
331 Root destination path string
332 """
333 config_path = ["app", "root"]
334 root_destination = get_config_value_or_default(
335 config=config,
336 config_path=config_path,
337 default=DEFAULT_ROOT_DESTINATION,
338 )
340 if not traverse_config_path(config=config, config_path=config_path):
341 log_config_not_found_warning(
342 config_path,
343 f"root destination is missing. Using default root destination: {root_destination}",
344 )
346 return root_destination
349def prepare_root_destination(config: dict) -> str:
350 """Prepare root destination by creating directory if needed.
352 Args:
353 config: Configuration dictionary
355 Returns:
356 Absolute path to root destination directory
357 """
358 log_config_debug("Checking root destination ...")
359 root_destination = get_root_destination_path(config)
360 return ensure_directory_exists(root_destination)
363# =============================================================================
364# Drive Configuration Functions
365# =============================================================================
368def get_drive_destination_path(config: dict) -> str:
369 """Get drive destination path from config without creating directory.
371 Args:
372 config: Configuration dictionary
374 Returns:
375 Drive destination path string
376 """
377 config_path = ["drive", "destination"]
378 drive_destination = get_config_value_or_default(
379 config=config,
380 config_path=config_path,
381 default=DEFAULT_DRIVE_DESTINATION,
382 )
384 if not traverse_config_path(config=config, config_path=config_path):
385 log_config_not_found_warning(
386 config_path,
387 f"destination is missing. Using default drive destination: {drive_destination}.",
388 )
390 return drive_destination
393def prepare_drive_destination(config: dict) -> str:
394 """Prepare drive destination path by creating directory if needed.
396 Args:
397 config: Configuration dictionary
399 Returns:
400 Absolute path to drive destination directory
401 """
402 log_config_debug("Checking drive destination ...")
403 root_path = prepare_root_destination(config=config)
404 drive_destination = get_drive_destination_path(config)
405 return join_and_ensure_path(root_path, drive_destination)
408def get_drive_remove_obsolete(config: dict) -> bool:
409 """Return drive remove obsolete flag from config.
411 Args:
412 config: Configuration dictionary
414 Returns:
415 True if obsolete files should be removed, False otherwise
416 """
417 config_path = ["drive", "remove_obsolete"]
418 drive_remove_obsolete = get_config_value_or_default(config=config, config_path=config_path, default=False)
420 if not drive_remove_obsolete:
421 _log_config_warning_once(
422 config_path,
423 "remove_obsolete is not found. Not removing the obsolete files and folders.",
424 )
425 else:
426 log_config_debug(f"{'R' if drive_remove_obsolete else 'Not R'}emoving obsolete files and folders ...")
428 return drive_remove_obsolete
431# =============================================================================
432# Photos Configuration Functions
433# =============================================================================
436def get_photos_destination_path(config: dict) -> str:
437 """Get photos destination path from config without creating directory.
439 Args:
440 config: Configuration dictionary
442 Returns:
443 Photos destination path string
444 """
445 config_path = ["photos", "destination"]
446 photos_destination = get_config_value_or_default(
447 config=config,
448 config_path=config_path,
449 default=DEFAULT_PHOTOS_DESTINATION,
450 )
452 if not traverse_config_path(config=config, config_path=config_path):
453 log_config_not_found_warning(
454 config_path,
455 f"destination is missing. Using default photos destination: {config_path_to_string(config_path)}",
456 )
458 return photos_destination
461def prepare_photos_destination(config: dict) -> str:
462 """Prepare photos destination path by creating directory if needed.
464 Args:
465 config: Configuration dictionary
467 Returns:
468 Absolute path to photos destination directory
469 """
470 log_config_debug("Checking photos destination ...")
471 root_path = prepare_root_destination(config=config)
472 photos_destination = get_photos_destination_path(config)
473 return join_and_ensure_path(root_path, photos_destination)
476def get_photos_all_albums(config: dict) -> bool:
477 """Return flag to download all albums from config.
479 Args:
480 config: Configuration dictionary
482 Returns:
483 True if all albums should be synced, False otherwise
484 """
485 config_path = ["photos", "all_albums"]
486 download_all = get_config_value_or_default(config=config, config_path=config_path, default=False)
488 if download_all:
489 log_config_found_info("Syncing all albums.")
491 return download_all
494def get_photos_use_hardlinks(config: dict, log_messages: bool = True) -> bool:
495 """Return flag to use hard links for duplicate photos from config.
497 Args:
498 config: Configuration dictionary
499 log_messages: Whether to log informational messages (default: True)
501 Returns:
502 True if hard links should be used, False otherwise
503 """
504 config_path = ["photos", "use_hardlinks"]
505 use_hardlinks = get_config_value_or_default(config=config, config_path=config_path, default=False)
507 if use_hardlinks and log_messages:
508 log_config_found_info("Using hard links for duplicate photos.")
510 return use_hardlinks
513def get_photos_remove_obsolete(config: dict) -> bool:
514 """Return photos remove obsolete flag from config.
516 Args:
517 config: Configuration dictionary
519 Returns:
520 True if obsolete files should be removed, False otherwise
521 """
522 config_path = ["photos", "remove_obsolete"]
523 photos_remove_obsolete = get_config_value_or_default(config=config, config_path=config_path, default=False)
525 if not photos_remove_obsolete:
526 _log_config_warning_once(
527 config_path,
528 "remove_obsolete is not found. Not removing the obsolete files and folders.",
529 )
530 else:
531 log_config_debug(f"{'R' if photos_remove_obsolete else 'Not R'}emoving obsolete files and folders ...")
533 return photos_remove_obsolete
536def get_photos_folder_format(config: dict) -> str | None:
537 """Return filename format or None.
539 Args:
540 config: Configuration dictionary
542 Returns:
543 Folder format string if configured, None otherwise
544 """
545 config_path = ["photos", "folder_format"]
546 fmt = get_config_value_or_none(config=config, config_path=config_path)
548 if fmt:
549 log_config_found_info(f"Using format {fmt}.")
551 return fmt
554# =============================================================================
555# Photos Filter Configuration Functions
556# =============================================================================
559def validate_file_sizes(file_sizes: list[str]) -> list[str]:
560 """Validate and filter file sizes against valid options.
562 Args:
563 file_sizes: List of file size strings to validate
565 Returns:
566 List of valid file sizes (defaults to ["original"] if all invalid)
567 """
568 valid_file_sizes = list(PhotoAsset.PHOTO_VERSION_LOOKUP.keys())
569 validated_sizes = []
571 for file_size in file_sizes:
572 if file_size not in valid_file_sizes:
573 log_invalid_config_value(
574 ["photos", "filters", "file_sizes"],
575 file_size,
576 ",".join(valid_file_sizes),
577 )
578 else:
579 validated_sizes.append(file_size)
581 return validated_sizes if validated_sizes else ["original"]
584def get_photos_libraries_filter(config: dict, base_config_path: list[str]) -> list[str] | None:
585 """Get libraries filter from photos config.
587 Args:
588 config: Configuration dictionary
589 base_config_path: Base path to filters section
591 Returns:
592 List of library names if configured, None otherwise
593 """
594 config_path = base_config_path + ["libraries"]
595 libraries = get_config_value_or_none(config=config, config_path=config_path)
597 if not libraries or len(libraries) == 0:
598 log_config_not_found_warning(config_path, "not found. Downloading all libraries ...")
599 return None
601 return libraries
604def get_photos_albums_filter(config: dict, base_config_path: list[str]) -> list[str] | None:
605 """Get albums filter from photos config.
607 Args:
608 config: Configuration dictionary
609 base_config_path: Base path to filters section
611 Returns:
612 List of album names if configured, None otherwise
613 """
614 config_path = base_config_path + ["albums"]
615 albums = get_config_value_or_none(config=config, config_path=config_path)
617 if not albums or len(albums) == 0:
618 log_config_not_found_warning(config_path, "not found. Downloading all albums ...")
619 return None
621 return albums
624def get_photos_file_sizes_filter(config: dict, base_config_path: list[str]) -> list[str]:
625 """Get file sizes filter from photos config.
627 Args:
628 config: Configuration dictionary
629 base_config_path: Base path to filters section
631 Returns:
632 List of file size options (defaults to ["original"])
633 """
634 config_path = base_config_path + ["file_sizes"]
636 if not traverse_config_path(config=config, config_path=config_path):
637 log_config_not_found_warning(config_path, "not found. Downloading original size photos ...")
638 return ["original"]
640 file_sizes = get_config_value(config=config, config_path=config_path)
641 return validate_file_sizes(file_sizes)
644def get_photos_extensions_filter(config: dict, base_config_path: list[str]) -> list[str] | None:
645 """Get extensions filter from photos config.
647 Args:
648 config: Configuration dictionary
649 base_config_path: Base path to filters section
651 Returns:
652 List of file extensions if configured, None otherwise
653 """
654 config_path = base_config_path + ["extensions"]
655 extensions = get_config_value_or_none(config=config, config_path=config_path)
657 if not extensions or len(extensions) == 0:
658 log_config_not_found_warning(config_path, "not found. Downloading all extensions ...")
659 return None
661 return extensions
664def get_photos_filters(config: dict) -> dict[str, Any]:
665 """Return photos filters from config.
667 Args:
668 config: Configuration dictionary
670 Returns:
671 Dictionary containing filter configuration for photos
672 """
673 photos_filters = {
674 "libraries": None,
675 "albums": None,
676 "file_sizes": ["original"],
677 "extensions": None,
678 }
680 base_config_path = ["photos", "filters"]
682 # Check for filters section existence
683 if not traverse_config_path(config=config, config_path=base_config_path):
684 log_config_not_found_warning(
685 base_config_path,
686 "not found. Downloading all libraries and albums with original size ...",
687 )
688 return photos_filters
690 # Parse individual filter components
691 photos_filters["libraries"] = get_photos_libraries_filter(config, base_config_path)
692 photos_filters["albums"] = get_photos_albums_filter(config, base_config_path)
693 photos_filters["file_sizes"] = get_photos_file_sizes_filter(config, base_config_path)
694 photos_filters["extensions"] = get_photos_extensions_filter(config, base_config_path)
696 return photos_filters
699# =============================================================================
700# SMTP Configuration Functions
701# =============================================================================
704def get_smtp_config_value(config: dict, key: str, warn_if_missing: bool = True) -> str | None:
705 """Get SMTP configuration value with optional warning.
707 Common helper for SMTP config retrieval to reduce duplication.
709 Args:
710 config: Configuration dictionary
711 key: SMTP config key name
712 warn_if_missing: Whether to log warning if not found
714 Returns:
715 Config value if found, None otherwise
716 """
717 config_path = ["app", "smtp", key]
718 value = get_config_value_or_none(config=config, config_path=config_path)
720 if value is None and warn_if_missing:
721 log_config_not_found_warning(config_path, f"{key} is not found.")
723 return value
726def get_smtp_email(config: dict) -> str | None:
727 """Return smtp from email from config.
729 Args:
730 config: Configuration dictionary
732 Returns:
733 SMTP email address if configured, None otherwise
734 """
735 return get_smtp_config_value(config, "email", warn_if_missing=False)
738def get_smtp_username(config: dict) -> str | None:
739 """Return smtp username from the config, if set.
741 Args:
742 config: Configuration dictionary
744 Returns:
745 SMTP username if configured, None otherwise
746 """
747 return get_smtp_config_value(config, "username", warn_if_missing=False)
750def get_smtp_to_email(config: dict) -> str | None:
751 """Return smtp to email from config, defaults to from email.
753 Args:
754 config: Configuration dictionary
756 Returns:
757 SMTP 'to' email address, falling back to 'from' email if not specified
758 """
759 to_email = get_smtp_config_value(config, "to", warn_if_missing=False)
760 return to_email if to_email else get_smtp_email(config=config)
763def get_smtp_password(config: dict) -> str | None:
764 """Return smtp password from config.
766 Args:
767 config: Configuration dictionary
769 Returns:
770 SMTP password if configured, None otherwise
771 """
772 return get_smtp_config_value(config, "password", warn_if_missing=True)
775def get_smtp_host(config: dict) -> str | None:
776 """Return smtp host from config.
778 Args:
779 config: Configuration dictionary
781 Returns:
782 SMTP host if configured, None otherwise
783 """
784 return get_smtp_config_value(config, "host", warn_if_missing=True)
787def get_smtp_port(config: dict) -> int | None:
788 """Return smtp port from config.
790 Args:
791 config: Configuration dictionary
793 Returns:
794 SMTP port if configured, None otherwise
795 """
796 return get_smtp_config_value(config, "port", warn_if_missing=True) # type: ignore[return-value]
799def get_smtp_no_tls(config: dict) -> bool:
800 """Return smtp no_tls flag from config.
802 Args:
803 config: Configuration dictionary
805 Returns:
806 True if TLS should be disabled, False otherwise
807 """
808 no_tls = get_smtp_config_value(config, "no_tls", warn_if_missing=True)
809 return no_tls if no_tls is not None else False # type: ignore[return-value]
812# =============================================================================
813# Notification Service Configuration Functions
814# =============================================================================
817def get_notification_config_value(config: dict, service: str, key: str) -> str | None:
818 """Get notification service configuration value.
820 Common helper for notification service config retrieval.
822 Args:
823 config: Configuration dictionary
824 service: Service name (telegram, discord, pushover)
825 key: Config key name
827 Returns:
828 Config value if found, None otherwise
829 """
830 config_path = ["app", service, key]
831 value = get_config_value_or_none(config=config, config_path=config_path)
833 if value is None:
834 log_config_not_found_warning(config_path, f"{key} is not found.")
836 return value
839def get_telegram_bot_token(config: dict) -> str | None:
840 """Return telegram bot token from config.
842 Args:
843 config: Configuration dictionary
845 Returns:
846 Telegram bot token if configured, None otherwise
847 """
848 return get_notification_config_value(config, "telegram", "bot_token")
851def get_telegram_chat_id(config: dict) -> str | None:
852 """Return telegram chat id from config.
854 Args:
855 config: Configuration dictionary
857 Returns:
858 Telegram chat ID if configured, None otherwise
859 """
860 return get_notification_config_value(config, "telegram", "chat_id")
863def get_discord_webhook_url(config: dict) -> str | None:
864 """Return discord webhook_url from config.
866 Args:
867 config: Configuration dictionary
869 Returns:
870 Discord webhook URL if configured, None otherwise
871 """
872 return get_notification_config_value(config, "discord", "webhook_url")
875def get_discord_username(config: dict) -> str | None:
876 """Return discord username from config.
878 Args:
879 config: Configuration dictionary
881 Returns:
882 Discord username if configured, None otherwise
883 """
884 return get_notification_config_value(config, "discord", "username")
887def get_pushover_user_key(config: dict) -> str | None:
888 """Return Pushover user key from config.
890 Args:
891 config: Configuration dictionary
893 Returns:
894 Pushover user key if configured, None otherwise
895 """
896 return get_notification_config_value(config, "pushover", "user_key")
899def get_pushover_api_token(config: dict) -> str | None:
900 """Return Pushover API token from config.
902 Args:
903 config: Configuration dictionary
905 Returns:
906 Pushover API token if configured, None otherwise
907 """
908 return get_notification_config_value(config, "pushover", "api_token")
911# =============================================================================
912# Sync Summary Notification Configuration Functions
913# =============================================================================
916def get_sync_summary_enabled(config: dict) -> bool:
917 """Return whether sync summary notifications are enabled.
919 Args:
920 config: Configuration dictionary
922 Returns:
923 True if sync summary is enabled, False otherwise (default: False)
924 """
925 config_path = ["app", "notifications", "sync_summary", "enabled"]
926 if not traverse_config_path(config=config, config_path=config_path):
927 return False
929 value = get_config_value(config=config, config_path=config_path)
930 return bool(value) if value is not None else False
933def get_sync_summary_on_success(config: dict) -> bool:
934 """Return whether to send summary on successful syncs.
936 Args:
937 config: Configuration dictionary
939 Returns:
940 True if should send on success, False otherwise (default: True)
941 """
942 config_path = ["app", "notifications", "sync_summary", "on_success"]
943 if not traverse_config_path(config=config, config_path=config_path):
944 return True # Default to True if not configured
946 value = get_config_value(config=config, config_path=config_path)
947 return bool(value) if value is not None else True
950def get_sync_summary_on_error(config: dict) -> bool:
951 """Return whether to send summary when errors occur.
953 Args:
954 config: Configuration dictionary
956 Returns:
957 True if should send on error, False otherwise (default: True)
958 """
959 config_path = ["app", "notifications", "sync_summary", "on_error"]
960 if not traverse_config_path(config=config, config_path=config_path):
961 return True # Default to True if not configured
963 value = get_config_value(config=config, config_path=config_path)
964 return bool(value) if value is not None else True
967def get_sync_summary_min_downloads(config: dict) -> int:
968 """Return minimum downloads required to trigger notification.
970 Args:
971 config: Configuration dictionary
973 Returns:
974 Minimum downloads threshold (default: 1)
975 """
976 config_path = ["app", "notifications", "sync_summary", "min_downloads"]
977 if not traverse_config_path(config=config, config_path=config_path):
978 return 1 # Default to 1 if not configured
980 value = get_config_value(config=config, config_path=config_path)
981 return int(value) if value is not None else 1