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_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.

100

101 Args:

102 config: Configuration dictionary

103

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

111

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

113 return validate_and_strip_username(username, config_path)

114

115

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

117 """Return retry login interval from config.

118

119 Args:

120 config: Configuration dictionary

121

122 Returns:

123 Retry login interval in seconds

124 """

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

126

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.")

136

137 return retry_login_interval

138

139

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

141 """Return region from config.

142

143 Args:

144 config: Configuration dictionary

145

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")

151

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"

160

161 return region

162

163

164# =============================================================================

165# Sync Interval Configuration Functions

166# =============================================================================

167

168

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).

171

172 Extracted common logic for retrieving sync intervals.

173

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)

179

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 )

188

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.")

197

198 return sync_interval

199

200

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

202 """Return drive sync interval from config.

203

204 Args:

205 config: Configuration dictionary

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

207

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)

213

214

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

216 """Return photos sync interval from config.

217

218 Args:

219 config: Configuration dictionary

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

221

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)

227

228

229# =============================================================================

230# Thread Configuration Functions

231# =============================================================================

232

233

234def calculate_default_max_threads() -> int:

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

236

237 Returns:

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

239 """

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

241

242

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

244 """Parse and validate max_threads configuration value.

245

246 Args:

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

248 default_max_threads: Default value to use

249

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

267

268 return max_threads

269

270

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

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

273

274 Args:

275 config: Configuration dictionary

276

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"]

282

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

289

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)

292

293

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

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

296

297 Args:

298 config: Configuration dictionary

299

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

306

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

308 if isinstance(value, bool):

309 return value

310

311 # Handle string values for backwards compatibility

312 if isinstance(value, str):

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

314

315 # Default to enabled for any other type

316 return True

317

318

319# =============================================================================

320# Root Destination Functions

321# =============================================================================

322

323

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

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

326

327 Args:

328 config: Configuration dictionary

329

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 )

339

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 )

345

346 return root_destination

347

348

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

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

351

352 Args:

353 config: Configuration dictionary

354

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)

361

362

363# =============================================================================

364# Drive Configuration Functions

365# =============================================================================

366

367

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

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

370

371 Args:

372 config: Configuration dictionary

373

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 )

383

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 )

389

390 return drive_destination

391

392

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

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

395

396 Args:

397 config: Configuration dictionary

398

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)

406

407

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

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

410

411 Args:

412 config: Configuration dictionary

413

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)

419

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 ...")

427

428 return drive_remove_obsolete

429

430

431# =============================================================================

432# Photos Configuration Functions

433# =============================================================================

434

435

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

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

438

439 Args:

440 config: Configuration dictionary

441

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 )

451

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 )

457

458 return photos_destination

459

460

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

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

463

464 Args:

465 config: Configuration dictionary

466

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)

474

475

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

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

478

479 Args:

480 config: Configuration dictionary

481

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)

487

488 if download_all:

489 log_config_found_info("Syncing all albums.")

490

491 return download_all

492

493

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

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

496

497 Args:

498 config: Configuration dictionary

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

500

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)

506

507 if use_hardlinks and log_messages:

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

509

510 return use_hardlinks

511

512

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

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

515

516 Args:

517 config: Configuration dictionary

518

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)

524

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 ...")

532

533 return photos_remove_obsolete

534

535

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

537 """Return filename format or None.

538

539 Args:

540 config: Configuration dictionary

541

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)

547

548 if fmt:

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

550

551 return fmt

552

553

554# =============================================================================

555# Photos Filter Configuration Functions

556# =============================================================================

557

558

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

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

561

562 Args:

563 file_sizes: List of file size strings to validate

564

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 = []

570

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)

580

581 return validated_sizes if validated_sizes else ["original"]

582

583

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

585 """Get libraries filter from photos config.

586

587 Args:

588 config: Configuration dictionary

589 base_config_path: Base path to filters section

590

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)

596

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

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

599 return None

600

601 return libraries

602

603

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

605 """Get albums filter from photos config.

606

607 Args:

608 config: Configuration dictionary

609 base_config_path: Base path to filters section

610

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)

616

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

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

619 return None

620

621 return albums

622

623

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

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

626

627 Args:

628 config: Configuration dictionary

629 base_config_path: Base path to filters section

630

631 Returns:

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

633 """

634 config_path = base_config_path + ["file_sizes"]

635

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"]

639

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

641 return validate_file_sizes(file_sizes)

642

643

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

645 """Get extensions filter from photos config.

646

647 Args:

648 config: Configuration dictionary

649 base_config_path: Base path to filters section

650

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)

656

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

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

659 return None

660

661 return extensions

662

663

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

665 """Return photos filters from config.

666

667 Args:

668 config: Configuration dictionary

669

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 }

679

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

681

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

689

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)

695

696 return photos_filters

697

698

699# =============================================================================

700# SMTP Configuration Functions

701# =============================================================================

702

703

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

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

706

707 Common helper for SMTP config retrieval to reduce duplication.

708

709 Args:

710 config: Configuration dictionary

711 key: SMTP config key name

712 warn_if_missing: Whether to log warning if not found

713

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)

719

720 if value is None and warn_if_missing:

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

722

723 return value

724

725

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

727 """Return smtp from email from config.

728

729 Args:

730 config: Configuration dictionary

731

732 Returns:

733 SMTP email address if configured, None otherwise

734 """

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

736

737

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

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

740

741 Args:

742 config: Configuration dictionary

743

744 Returns:

745 SMTP username if configured, None otherwise

746 """

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

748

749

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

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

752

753 Args:

754 config: Configuration dictionary

755

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)

761

762

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

764 """Return smtp password from config.

765

766 Args:

767 config: Configuration dictionary

768

769 Returns:

770 SMTP password if configured, None otherwise

771 """

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

773

774

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

776 """Return smtp host from config.

777

778 Args:

779 config: Configuration dictionary

780

781 Returns:

782 SMTP host if configured, None otherwise

783 """

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

785

786

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

788 """Return smtp port from config.

789

790 Args:

791 config: Configuration dictionary

792

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]

797

798

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

800 """Return smtp no_tls flag from config.

801

802 Args:

803 config: Configuration dictionary

804

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]

810

811

812# =============================================================================

813# Notification Service Configuration Functions

814# =============================================================================

815

816

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

818 """Get notification service configuration value.

819

820 Common helper for notification service config retrieval.

821

822 Args:

823 config: Configuration dictionary

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

825 key: Config key name

826

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)

832

833 if value is None:

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

835

836 return value

837

838

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

840 """Return telegram bot token from config.

841

842 Args:

843 config: Configuration dictionary

844

845 Returns:

846 Telegram bot token if configured, None otherwise

847 """

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

849

850

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

852 """Return telegram chat id from config.

853

854 Args:

855 config: Configuration dictionary

856

857 Returns:

858 Telegram chat ID if configured, None otherwise

859 """

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

861

862

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

864 """Return discord webhook_url from config.

865

866 Args:

867 config: Configuration dictionary

868

869 Returns:

870 Discord webhook URL if configured, None otherwise

871 """

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

873

874

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

876 """Return discord username from config.

877

878 Args:

879 config: Configuration dictionary

880

881 Returns:

882 Discord username if configured, None otherwise

883 """

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

885

886

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

888 """Return Pushover user key from config.

889

890 Args:

891 config: Configuration dictionary

892

893 Returns:

894 Pushover user key if configured, None otherwise

895 """

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

897

898

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

900 """Return Pushover API token from config.

901

902 Args:

903 config: Configuration dictionary

904

905 Returns:

906 Pushover API token if configured, None otherwise

907 """

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

909

910

911# =============================================================================

912# Sync Summary Notification Configuration Functions

913# =============================================================================

914

915

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

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

918

919 Args:

920 config: Configuration dictionary

921

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

928

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

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

931

932

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

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

935

936 Args:

937 config: Configuration dictionary

938

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

945

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

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

948

949

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

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

952

953 Args:

954 config: Configuration dictionary

955

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

962

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

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

965

966

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

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

969

970 Args:

971 config: Configuration dictionary

972

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

979

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

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