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

1"""Config file parser. 

2 

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

7 

8__author__ = "Mandar Patil (mandarons@pm.me)" 

9 

10import multiprocessing 

11from typing import Any 

12 

13from icloudpy.services.photos import PhotoAsset 

14 

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 

39 

40# Configure icloudpy logging immediately after import 

41configure_icloudpy_logging() 

42 

43LOGGER = get_logger() 

44 

45# Cache for config values to prevent repeated warnings 

46_config_warning_cache = set() 

47 

48 

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

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

51 

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) 

60 

61 

62def clear_config_warning_cache() -> None: 

63 """Clear the configuration warning cache. 

64 

65 This function is primarily intended for testing purposes to ensure 

66 clean test isolation. 

67 """ 

68 _config_warning_cache.clear() 

69 

70 

71# ============================================================================= 

72# String Processing Functions 

73# ============================================================================= 

74 

75 

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

77 """Validate and strip username string. 

78 

79 Args: 

80 username: Raw username string from config 

81 config_path: Config path for error logging 

82 

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 

91 

92 

93# ============================================================================= 

94# Credential Configuration Functions 

95# ============================================================================= 

96 

97 

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