Coverage for src/config_parser.py: 100%

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_REQUEST_TIMEOUT_SEC,

19 DEFAULT_RETRY_LOGIN_INTERVAL_SEC,

20 DEFAULT_ROOT_DESTINATION,

21 DEFAULT_SYNC_INTERVAL_SEC,

22 configure_icloudpy_logging,

23 get_logger,

24)

25from src.config_logging import (

26 log_config_debug,

27 log_config_error,

28 log_config_found_info,

29 log_config_not_found_warning,

30 log_invalid_config_value,

31)

32from src.config_utils import (

33 config_path_to_string,

34 get_config_value,

35 get_config_value_or_default,

36 get_config_value_or_none,

37 traverse_config_path,

38)

39from src.filesystem_utils import ensure_directory_exists, join_and_ensure_path

41# Configure icloudpy logging immediately after import

42configure_icloudpy_logging()

44LOGGER = get_logger()

46# Cache for config values to prevent repeated warnings

47_config_warning_cache = set()

50def _log_config_warning_once(config_path: list, message: str) -> None:

51 """Log a configuration warning only once for the given config path.

53 Args:

54 config_path: Configuration path as list

55 message: Warning message to log

56 """

57 config_path_key = config_path_to_string(config_path)

58 if config_path_key not in _config_warning_cache:

59 _config_warning_cache.add(config_path_key)

60 log_config_not_found_warning(config_path, message)

63def clear_config_warning_cache() -> None:

64 """Clear the configuration warning cache.

66 This function is primarily intended for testing purposes to ensure

67 clean test isolation.

68 """

69 _config_warning_cache.clear()

72# =============================================================================

73# String Processing Functions

74# =============================================================================

77def validate_and_strip_username(username: str, config_path: list[str]) -> str | None:

78 """Validate and strip username string.

80 Args:

81 username: Raw username string from config

82 config_path: Config path for error logging

84 Returns:

85 Stripped username if valid, None if empty

86 """

87 username = username.strip()

88 if len(username) == 0:

89 log_config_error(config_path, "username is empty")

90 return None

91 return username

94# =============================================================================

95# Credential Configuration Functions

96# =============================================================================

99def get_username(config: dict) -> str | None:

100 """Get username from config.

101

102 Args:

103 config: Configuration dictionary

104

105 Returns:

106 Username string if found and valid, None otherwise

107 """

108 config_path = ["app", "credentials", "username"]

109 if not traverse_config_path(config=config, config_path=config_path):

110 log_config_error(config_path, "username is missing. Please set the username.")

111 return None

112

113 username = get_config_value(config=config, config_path=config_path)

114 return validate_and_strip_username(username, config_path)

115

116

117def get_retry_login_interval(config: dict) -> int:

118 """Return retry login interval from config.

119

120 Args:

121 config: Configuration dictionary

122

123 Returns:

124 Retry login interval in seconds

125 """

126 config_path = ["app", "credentials", "retry_login_interval"]

127

128 if not traverse_config_path(config=config, config_path=config_path):

129 retry_login_interval = DEFAULT_RETRY_LOGIN_INTERVAL_SEC

130 log_config_not_found_warning(

131 config_path,

132 f"not found. Using default {retry_login_interval} seconds ...",

133 )

134 else:

135 retry_login_interval = get_config_value(config=config, config_path=config_path)

136 log_config_found_info(f"Retrying login every {retry_login_interval} seconds.")

137

138 return retry_login_interval

139

140

141def get_region(config: dict) -> str:

142 """Return region from config.

143

144 Args:

145 config: Configuration dictionary

146

147 Returns:

148 Region string ('global' or 'china')

149 """

150 config_path = ["app", "region"]

151 region = get_config_value_or_default(config=config, config_path=config_path, default="global")

152

153 if region == "global" and not traverse_config_path(config=config, config_path=config_path):

154 log_config_not_found_warning(config_path, "not found. Using default value - global ...")

155 elif region not in ["global", "china"]:

156 log_config_error(

157 config_path,

158 "is invalid. Valid values are - global or china. Using default value - global ...",

159 )

160 region = "global"

161

162 return region

163

164

165# =============================================================================

166# Sync Interval Configuration Functions

167# =============================================================================

168

169

170def get_sync_interval(config: dict, config_path: list[str], service_name: str, log_messages: bool = True) -> int:

171 """Get sync interval for a service (drive or photos).

172

173 Extracted common logic for retrieving sync intervals.

174

175 Args:

176 config: Configuration dictionary

177 config_path: Path to sync_interval config

178 service_name: Name of service for logging ("drive" or "photos")

179 log_messages: Whether to log informational messages (default: True)

180

181 Returns:

182 Sync interval in seconds

183 """

184 sync_interval = get_config_value_or_default(

185 config=config,

186 config_path=config_path,

187 default=DEFAULT_SYNC_INTERVAL_SEC,

188 )

189

190 if log_messages:

191 if sync_interval == DEFAULT_SYNC_INTERVAL_SEC:

192 log_config_not_found_warning(

193 config_path,

194 f"is not found. Using default sync_interval: {sync_interval} seconds ...",

195 )

196 else:

197 log_config_found_info(f"Syncing {service_name} every {sync_interval} seconds.")

198

199 return sync_interval

200

201

202def get_drive_sync_interval(config: dict, log_messages: bool = True) -> int:

203 """Return drive sync interval from config.

204

205 Args:

206 config: Configuration dictionary

207 log_messages: Whether to log informational messages (default: True)

208

209 Returns:

210 Drive sync interval in seconds

211 """

212 config_path = ["drive", "sync_interval"]

213 return get_sync_interval(config=config, config_path=config_path, service_name="drive", log_messages=log_messages)

214

215

216def get_drive_request_timeout(config: dict) -> int:

217 """Return drive request timeout from config.

218

219 Args:

220 config: Configuration dictionary

221

222 Returns:

223 Request timeout in seconds (default: DEFAULT_REQUEST_TIMEOUT_SEC)

224 """

225 config_path = ["drive", "request_timeout"]

226 return get_config_value_or_default(

227 config=config,

228 config_path=config_path,

229 default=DEFAULT_REQUEST_TIMEOUT_SEC,

230 )

231

232

233def get_photos_sync_interval(config: dict, log_messages: bool = True) -> int:

234 """Return photos sync interval from config.

235

236 Args:

237 config: Configuration dictionary

238 log_messages: Whether to log informational messages (default: True)

239

240 Returns:

241 Photos sync interval in seconds

242 """

243 config_path = ["photos", "sync_interval"]

244 return get_sync_interval(config=config, config_path=config_path, service_name="photos", log_messages=log_messages)

245

246

247# =============================================================================

248# Thread Configuration Functions

249# =============================================================================

250

251

252def calculate_default_max_threads() -> int:

253 """Calculate default maximum threads based on CPU cores.

254

255 Returns:

256 Default max threads (min of CPU count and 8)

257 """

258 return min(multiprocessing.cpu_count(), 8)

259

260

261def parse_max_threads_value(max_threads_config: Any, default_max_threads: int) -> int:

262 """Parse and validate max_threads configuration value.

263

264 Args:

265 max_threads_config: Raw config value (string "auto" or integer)

266 default_max_threads: Default value to use

267

268 Returns:

269 Validated max threads value (capped at 16)

270 """

271 # Handle "auto" value

272 if isinstance(max_threads_config, str) and max_threads_config.lower() == "auto":

273 max_threads = default_max_threads

274 log_config_found_info(f"Using automatic thread count: {max_threads} threads (based on CPU cores).")

275 elif isinstance(max_threads_config, int) and max_threads_config >= 1:

276 max_threads = min(max_threads_config, 16) # Cap at 16 to avoid overwhelming servers

277 log_config_found_info(f"Using configured max_threads: {max_threads}.")

278 else:

279 log_invalid_config_value(

280 ["app", "max_threads"],

281 max_threads_config,

282 "'auto' or integer >= 1",

283 )

284 max_threads = default_max_threads

285

286 return max_threads

287

288

289def get_app_max_threads(config: dict) -> int:

290 """Return app-level max threads from config with support for 'auto' value.

291

292 Args:

293 config: Configuration dictionary

294

295 Returns:

296 Maximum number of threads for parallel operations

297 """

298 default_max_threads = calculate_default_max_threads()

299 config_path = ["app", "max_threads"]

300

301 if not traverse_config_path(config=config, config_path=config_path):

302 log_config_debug(

303 f"max_threads is not found in {config_path_to_string(config_path=config_path)}. "

304 f"Using default max_threads: {default_max_threads} (auto) ...",

305 )

306 return default_max_threads

307

308 max_threads_config = get_config_value(config=config, config_path=config_path)

309 return parse_max_threads_value(max_threads_config, default_max_threads)

310

311

312def get_usage_tracking_enabled(config: dict) -> bool:

313 """Get usage tracking enabled setting from configuration.

314

315 Args:

316 config: Configuration dictionary

317

318 Returns:

319 True if usage tracking is enabled (default), False if disabled

320 """

321 config_path = ["app", "usage_tracking", "enabled"]

322 if not traverse_config_path(config=config, config_path=config_path):

323 return True # Default to enabled if not configured

324

325 value = get_config_value(config=config, config_path=config_path)

326 if isinstance(value, bool):

327 return value

328

329 # Handle string values for backwards compatibility

330 if isinstance(value, str):

331 return value.lower() not in ("false", "no", "0", "disabled", "off")

332

333 # Default to enabled for any other type

334 return True

335

336

337# =============================================================================

338# Root Destination Functions

339# =============================================================================

340

341

342def get_root_destination_path(config: dict) -> str:

343 """Get root destination path from config without creating directory.

344

345 Args:

346 config: Configuration dictionary

347

348 Returns:

349 Root destination path string

350 """

351 config_path = ["app", "root"]

352 root_destination = get_config_value_or_default(

353 config=config,

354 config_path=config_path,

355 default=DEFAULT_ROOT_DESTINATION,

356 )

357

358 if not traverse_config_path(config=config, config_path=config_path):

359 log_config_not_found_warning(

360 config_path,

361 f"root destination is missing. Using default root destination: {root_destination}",

362 )

363

364 return root_destination

365

366

367def prepare_root_destination(config: dict) -> str:

368 """Prepare root destination by creating directory if needed.

369

370 Args:

371 config: Configuration dictionary

372

373 Returns:

374 Absolute path to root destination directory

375 """

376 log_config_debug("Checking root destination ...")

377 root_destination = get_root_destination_path(config)

378 return ensure_directory_exists(root_destination)

379

380

381# =============================================================================

382# Drive Configuration Functions

383# =============================================================================

384

385

386def get_drive_destination_path(config: dict) -> str:

387 """Get drive destination path from config without creating directory.

388

389 Args:

390 config: Configuration dictionary

391

392 Returns:

393 Drive destination path string

394 """

395 config_path = ["drive", "destination"]

396 drive_destination = get_config_value_or_default(

397 config=config,

398 config_path=config_path,

399 default=DEFAULT_DRIVE_DESTINATION,

400 )

401

402 if not traverse_config_path(config=config, config_path=config_path):

403 log_config_not_found_warning(

404 config_path,

405 f"destination is missing. Using default drive destination: {drive_destination}.",

406 )

407

408 return drive_destination

409

410

411def prepare_drive_destination(config: dict) -> str:

412 """Prepare drive destination path by creating directory if needed.

413

414 Args:

415 config: Configuration dictionary

416

417 Returns:

418 Absolute path to drive destination directory

419 """

420 log_config_debug("Checking drive destination ...")

421 root_path = prepare_root_destination(config=config)

422 drive_destination = get_drive_destination_path(config)

423 return join_and_ensure_path(root_path, drive_destination)

424

425

426def get_drive_remove_obsolete(config: dict) -> bool:

427 """Return drive remove obsolete flag from config.

428

429 Args:

430 config: Configuration dictionary

431

432 Returns:

433 True if obsolete files should be removed, False otherwise

434 """

435 config_path = ["drive", "remove_obsolete"]

436 drive_remove_obsolete = get_config_value_or_default(config=config, config_path=config_path, default=False)

437

438 if not drive_remove_obsolete:

439 _log_config_warning_once(

440 config_path,

441 "remove_obsolete is not found. Not removing the obsolete files and folders.",

442 )

443 else:

444 log_config_debug(f"{'R' if drive_remove_obsolete else 'Not R'}emoving obsolete files and folders ...")

445

446 return drive_remove_obsolete

447

448

449# =============================================================================

450# Photos Configuration Functions

451# =============================================================================

452

453

454def get_photos_destination_path(config: dict) -> str:

455 """Get photos destination path from config without creating directory.

456

457 Args:

458 config: Configuration dictionary

459

460 Returns:

461 Photos destination path string

462 """

463 config_path = ["photos", "destination"]

464 photos_destination = get_config_value_or_default(

465 config=config,

466 config_path=config_path,

467 default=DEFAULT_PHOTOS_DESTINATION,

468 )

469

470 if not traverse_config_path(config=config, config_path=config_path):

471 log_config_not_found_warning(

472 config_path,

473 f"destination is missing. Using default photos destination: {config_path_to_string(config_path)}",

474 )

475

476 return photos_destination

477

478

479def prepare_photos_destination(config: dict) -> str:

480 """Prepare photos destination path by creating directory if needed.

481

482 Args:

483 config: Configuration dictionary

484

485 Returns:

486 Absolute path to photos destination directory

487 """

488 log_config_debug("Checking photos destination ...")

489 root_path = prepare_root_destination(config=config)

490 photos_destination = get_photos_destination_path(config)

491 return join_and_ensure_path(root_path, photos_destination)

492

493

494def get_photos_all_albums(config: dict) -> bool:

495 """Return flag to download all albums from config.

496

497 Args:

498 config: Configuration dictionary

499

500 Returns:

501 True if all albums should be synced, False otherwise

502 """

503 config_path = ["photos", "all_albums"]

504 download_all = get_config_value_or_default(config=config, config_path=config_path, default=False)

505

506 if download_all:

507 log_config_found_info("Syncing all albums.")

508

509 return download_all

510

511

512def get_photos_use_hardlinks(config: dict, log_messages: bool = True) -> bool:

513 """Return flag to use hard links for duplicate photos from config.

514

515 Args:

516 config: Configuration dictionary

517 log_messages: Whether to log informational messages (default: True)

518

519 Returns:

520 True if hard links should be used, False otherwise

521 """

522 config_path = ["photos", "use_hardlinks"]

523 use_hardlinks = get_config_value_or_default(config=config, config_path=config_path, default=False)

524

525 if use_hardlinks and log_messages:

526 log_config_found_info("Using hard links for duplicate photos.")

527

528 return use_hardlinks

529

530

531def get_photos_remove_obsolete(config: dict) -> bool:

532 """Return photos remove obsolete flag from config.

533

534 Args:

535 config: Configuration dictionary

536

537 Returns:

538 True if obsolete files should be removed, False otherwise

539 """

540 config_path = ["photos", "remove_obsolete"]

541 photos_remove_obsolete = get_config_value_or_default(config=config, config_path=config_path, default=False)

542

543 if not photos_remove_obsolete:

544 _log_config_warning_once(

545 config_path,

546 "remove_obsolete is not found. Not removing the obsolete files and folders.",

547 )

548 else:

549 log_config_debug(f"{'R' if photos_remove_obsolete else 'Not R'}emoving obsolete files and folders ...")

550

551 return photos_remove_obsolete

552

553

554def get_photos_folder_format(config: dict) -> str | None:

555 """Return filename format or None.

556

557 Args:

558 config: Configuration dictionary

559

560 Returns:

561 Folder format string if configured, None otherwise

562 """

563 config_path = ["photos", "folder_format"]

564 fmt = get_config_value_or_none(config=config, config_path=config_path)

565

566 if fmt:

567 log_config_found_info(f"Using format {fmt}.")

568

569 return fmt

570

571

572# =============================================================================

573# Photos Filter Configuration Functions

574# =============================================================================

575

576

577def validate_file_sizes(file_sizes: list[str]) -> list[str]:

578 """Validate and filter file sizes against valid options.

579

580 Args:

581 file_sizes: List of file size strings to validate

582

583 Returns:

584 List of valid file sizes (defaults to ["original"] if all invalid)

585 """

586 valid_file_sizes = list(PhotoAsset.PHOTO_VERSION_LOOKUP.keys())

587 validated_sizes = []

588

589 for file_size in file_sizes:

590 if file_size not in valid_file_sizes:

591 log_invalid_config_value(

592 ["photos", "filters", "file_sizes"],

593 file_size,

594 ",".join(valid_file_sizes),

595 )

596 else:

597 validated_sizes.append(file_size)

598

599 return validated_sizes if validated_sizes else ["original"]

600

601

602def get_photos_libraries_filter(config: dict, base_config_path: list[str]) -> list[str] | None:

603 """Get libraries filter from photos config.

604

605 Args:

606 config: Configuration dictionary

607 base_config_path: Base path to filters section

608

609 Returns:

610 List of library names if configured, None otherwise

611 """

612 config_path = base_config_path + ["libraries"]

613 libraries = get_config_value_or_none(config=config, config_path=config_path)

614

615 if not libraries or len(libraries) == 0:

616 log_config_not_found_warning(config_path, "not found. Downloading all libraries ...")

617 return None

618

619 return libraries

620

621

622def get_photos_albums_filter(config: dict, base_config_path: list[str]) -> list[str] | None:

623 """Get albums filter from photos config.

624

625 Args:

626 config: Configuration dictionary

627 base_config_path: Base path to filters section

628

629 Returns:

630 List of album names if configured, None otherwise

631 """

632 config_path = base_config_path + ["albums"]

633 albums = get_config_value_or_none(config=config, config_path=config_path)

634

635 if not albums or len(albums) == 0:

636 log_config_not_found_warning(config_path, "not found. Downloading all albums ...")

637 return None

638

639 return albums

640

641

642def get_photos_file_sizes_filter(config: dict, base_config_path: list[str]) -> list[str]:

643 """Get file sizes filter from photos config.

644

645 Args:

646 config: Configuration dictionary

647 base_config_path: Base path to filters section

648

649 Returns:

650 List of file size options (defaults to ["original"])

651 """

652 config_path = base_config_path + ["file_sizes"]

653

654 if not traverse_config_path(config=config, config_path=config_path):

655 log_config_not_found_warning(config_path, "not found. Downloading original size photos ...")

656 return ["original"]

657

658 file_sizes = get_config_value(config=config, config_path=config_path)

659 return validate_file_sizes(file_sizes)

660

661

662def get_photos_extensions_filter(config: dict, base_config_path: list[str]) -> list[str] | None:

663 """Get extensions filter from photos config.

664

665 Args:

666 config: Configuration dictionary

667 base_config_path: Base path to filters section

668

669 Returns:

670 List of file extensions if configured, None otherwise

671 """

672 config_path = base_config_path + ["extensions"]

673 extensions = get_config_value_or_none(config=config, config_path=config_path)

674

675 if not extensions or len(extensions) == 0:

676 log_config_not_found_warning(config_path, "not found. Downloading all extensions ...")

677 return None

678

679 return extensions

680

681

682def get_photos_filters(config: dict) -> dict[str, Any]:

683 """Return photos filters from config.

684

685 Args:

686 config: Configuration dictionary

687

688 Returns:

689 Dictionary containing filter configuration for photos

690 """

691 photos_filters = {

692 "libraries": None,

693 "albums": None,

694 "file_sizes": ["original"],

695 "extensions": None,

696 }

697

698 base_config_path = ["photos", "filters"]

699

700 # Check for filters section existence

701 if not traverse_config_path(config=config, config_path=base_config_path):

702 log_config_not_found_warning(

703 base_config_path,

704 "not found. Downloading all libraries and albums with original size ...",

705 )

706 return photos_filters

707

708 # Parse individual filter components

709 photos_filters["libraries"] = get_photos_libraries_filter(config, base_config_path)

710 photos_filters["albums"] = get_photos_albums_filter(config, base_config_path)

711 photos_filters["file_sizes"] = get_photos_file_sizes_filter(config, base_config_path)

712 photos_filters["extensions"] = get_photos_extensions_filter(config, base_config_path)

713

714 return photos_filters

715

716

717# =============================================================================

718# SMTP Configuration Functions

719# =============================================================================

720

721

722def get_smtp_config_value(config: dict, key: str, warn_if_missing: bool = True) -> str | None:

723 """Get SMTP configuration value with optional warning.

724

725 Common helper for SMTP config retrieval to reduce duplication.

726

727 Args:

728 config: Configuration dictionary

729 key: SMTP config key name

730 warn_if_missing: Whether to log warning if not found

731

732 Returns:

733 Config value if found, None otherwise

734 """

735 config_path = ["app", "smtp", key]

736 value = get_config_value_or_none(config=config, config_path=config_path)

737

738 if value is None and warn_if_missing:

739 log_config_not_found_warning(config_path, f"{key} is not found.")

740

741 return value

742

743

744def get_smtp_email(config: dict) -> str | None:

745 """Return smtp from email from config.

746

747 Args:

748 config: Configuration dictionary

749

750 Returns:

751 SMTP email address if configured, None otherwise

752 """

753 return get_smtp_config_value(config, "email", warn_if_missing=False)

754

755

756def get_smtp_username(config: dict) -> str | None:

757 """Return smtp username from the config, if set.

758

759 Args:

760 config: Configuration dictionary

761

762 Returns:

763 SMTP username if configured, None otherwise

764 """

765 return get_smtp_config_value(config, "username", warn_if_missing=False)

766

767

768def get_smtp_to_email(config: dict) -> str | None:

769 """Return smtp to email from config, defaults to from email.

770

771 Args:

772 config: Configuration dictionary

773

774 Returns:

775 SMTP 'to' email address, falling back to 'from' email if not specified

776 """

777 to_email = get_smtp_config_value(config, "to", warn_if_missing=False)

778 return to_email if to_email else get_smtp_email(config=config)

779

780

781def get_smtp_password(config: dict) -> str | None:

782 """Return smtp password from config.

783

784 Args:

785 config: Configuration dictionary

786

787 Returns:

788 SMTP password if configured, None otherwise

789 """

790 return get_smtp_config_value(config, "password", warn_if_missing=True)

791

792

793def get_smtp_host(config: dict) -> str | None:

794 """Return smtp host from config.

795

796 Args:

797 config: Configuration dictionary

798

799 Returns:

800 SMTP host if configured, None otherwise

801 """

802 return get_smtp_config_value(config, "host", warn_if_missing=True)

803

804

805def get_smtp_port(config: dict) -> int | None:

806 """Return smtp port from config.

807

808 Args:

809 config: Configuration dictionary

810

811 Returns:

812 SMTP port if configured, None otherwise

813 """

814 return get_smtp_config_value(config, "port", warn_if_missing=True) # type: ignore[return-value]

815

816

817def get_smtp_no_tls(config: dict) -> bool:

818 """Return smtp no_tls flag from config.

819

820 Args:

821 config: Configuration dictionary

822

823 Returns:

824 True if TLS should be disabled, False otherwise

825 """

826 no_tls = get_smtp_config_value(config, "no_tls", warn_if_missing=True)

827 return no_tls if no_tls is not None else False # type: ignore[return-value]

828

829

830# =============================================================================

831# Notification Service Configuration Functions

832# =============================================================================

833

834

835def get_notification_config_value(config: dict, service: str, key: str) -> str | None:

836 """Get notification service configuration value.

837

838 Common helper for notification service config retrieval.

839

840 Args:

841 config: Configuration dictionary

842 service: Service name (telegram, discord, pushover)

843 key: Config key name

844

845 Returns:

846 Config value if found, None otherwise

847 """

848 config_path = ["app", service, key]

849 value = get_config_value_or_none(config=config, config_path=config_path)

850

851 if value is None:

852 log_config_not_found_warning(config_path, f"{key} is not found.")

853

854 return value

855

856

857def get_telegram_bot_token(config: dict) -> str | None:

858 """Return telegram bot token from config.

859

860 Args:

861 config: Configuration dictionary

862

863 Returns:

864 Telegram bot token if configured, None otherwise

865 """

866 return get_notification_config_value(config, "telegram", "bot_token")

867

868

869def get_telegram_chat_id(config: dict) -> str | None:

870 """Return telegram chat id from config.

871

872 Args:

873 config: Configuration dictionary

874

875 Returns:

876 Telegram chat ID if configured, None otherwise

877 """

878 return get_notification_config_value(config, "telegram", "chat_id")

879

880

881def get_discord_webhook_url(config: dict) -> str | None:

882 """Return discord webhook_url from config.

883

884 Args:

885 config: Configuration dictionary

886

887 Returns:

888 Discord webhook URL if configured, None otherwise

889 """

890 return get_notification_config_value(config, "discord", "webhook_url")

891

892

893def get_discord_username(config: dict) -> str | None:

894 """Return discord username from config.

895

896 Args:

897 config: Configuration dictionary

898

899 Returns:

900 Discord username if configured, None otherwise

901 """

902 return get_notification_config_value(config, "discord", "username")

903

904

905def get_pushover_user_key(config: dict) -> str | None:

906 """Return Pushover user key from config.

907

908 Args:

909 config: Configuration dictionary

910

911 Returns:

912 Pushover user key if configured, None otherwise

913 """

914 return get_notification_config_value(config, "pushover", "user_key")

915

916

917def get_pushover_api_token(config: dict) -> str | None:

918 """Return Pushover API token from config.

919

920 Args:

921 config: Configuration dictionary

922

923 Returns:

924 Pushover API token if configured, None otherwise

925 """

926 return get_notification_config_value(config, "pushover", "api_token")

927

928def get_pushover_notification_priority(config: dict) -> int | None:

929 """Return Pushover notification priority from config.

930

931 Args:

932 config: Configuration dictionary

933

934 Returns:

935 Pushover notification priority if configured, None otherwise

936 """

937 config_path = ["app", "pushover", "priority"]

938 return get_config_value_or_none(config=config, config_path=config_path)

939

940

941# =============================================================================

942# Sync Summary Notification Configuration Functions

943# =============================================================================

944

945

946def get_sync_summary_enabled(config: dict) -> bool:

947 """Return whether sync summary notifications are enabled.

948

949 Args:

950 config: Configuration dictionary

951

952 Returns:

953 True if sync summary is enabled, False otherwise (default: False)

954 """

955 config_path = ["app", "notifications", "sync_summary", "enabled"]

956 if not traverse_config_path(config=config, config_path=config_path):

957 return False

958

959 value = get_config_value(config=config, config_path=config_path)

960 return bool(value) if value is not None else False

961

962

963def get_sync_summary_on_success(config: dict) -> bool:

964 """Return whether to send summary on successful syncs.

965

966 Args:

967 config: Configuration dictionary

968

969 Returns:

970 True if should send on success, False otherwise (default: True)

971 """

972 config_path = ["app", "notifications", "sync_summary", "on_success"]

973 if not traverse_config_path(config=config, config_path=config_path):

974 return True # Default to True if not configured

975

976 value = get_config_value(config=config, config_path=config_path)

977 return bool(value) if value is not None else True

978

979

980def get_sync_summary_on_error(config: dict) -> bool:

981 """Return whether to send summary when errors occur.

982

983 Args:

984 config: Configuration dictionary

985

986 Returns:

987 True if should send on error, False otherwise (default: True)

988 """

989 config_path = ["app", "notifications", "sync_summary", "on_error"]

990 if not traverse_config_path(config=config, config_path=config_path):

991 return True # Default to True if not configured

992

993 value = get_config_value(config=config, config_path=config_path)

994 return bool(value) if value is not None else True

995

996

997def get_sync_summary_min_downloads(config: dict) -> int:

998 """Return minimum downloads required to trigger notification.

999

1000 Args:

1001 config: Configuration dictionary

1002

1003 Returns:

1004 Minimum downloads threshold (default: 1)

1005 """

1006 config_path = ["app", "notifications", "sync_summary", "min_downloads"]

1007 if not traverse_config_path(config=config, config_path=config_path):

1008 return 1 # Default to 1 if not configured

1009

1010 value = get_config_value(config=config, config_path=config_path)

1011 return int(value) if value is not None else 1