Coverage for src/sync.py: 100%

1"""Sync module."""

3__author__ = "Mandar Patil <mandarons@pm.me>"

4import datetime

5import os

6from time import sleep

8from icloudpy import ICloudPyService, exceptions, utils

10from src import (

11 DEFAULT_CONFIG_FILE_PATH,

12 DEFAULT_COOKIE_DIRECTORY,

13 ENV_CONFIG_FILE_PATH_KEY,

14 ENV_ICLOUD_PASSWORD_KEY,

15 config_parser,

16 configure_icloudpy_logging,

17 get_logger,

18 notify,

19 read_config,

20 sync_drive,

21 sync_photos,

22)

23from src.sync_stats import SyncSummary

24from src.usage import alive

26# Configure icloudpy logging immediately after import

27configure_icloudpy_logging()

29LOGGER = get_logger()

32def get_api_instance(

33 username: str,

34 password: str,

35 cookie_directory: str = DEFAULT_COOKIE_DIRECTORY,

36 server_region: str = "global",

37) -> ICloudPyService:

38 """

39 Create and return an iCloud API client instance.

41 Args:

42 username: iCloud username/Apple ID

43 password: iCloud password

44 cookie_directory: Directory to store authentication cookies

45 server_region: Server region ("china" or "global")

47 Returns:

48 Configured ICloudPyService instance

49 """

50 return (

51 ICloudPyService(

52 apple_id=username,

53 password=password,

54 cookie_directory=cookie_directory,

55 home_endpoint="https://www.icloud.com.cn",

56 setup_endpoint="https://setup.icloud.com.cn/setup/ws/1",

57 )

58 if server_region == "china"

59 else ICloudPyService(

60 apple_id=username,

61 password=password,

62 cookie_directory=cookie_directory,

63 )

64 )

67class SyncState:

68 """

69 Maintains synchronization state for drive and photos.

71 This class encapsulates the countdown timers and sync flags to avoid

72 passing multiple variables between functions.

73 """

75 def __init__(self):

76 """Initialize sync state with default values."""

77 self.drive_time_remaining = 0

78 self.photos_time_remaining = 0

79 self.enable_sync_drive = True

80 self.enable_sync_photos = True

81 self.last_send = None

84def _load_configuration():

85 """

86 Load configuration from file or environment.

88 Returns:

89 Configuration dictionary

90 """

91 config_path = os.environ.get(ENV_CONFIG_FILE_PATH_KEY, DEFAULT_CONFIG_FILE_PATH)

92 return read_config(config_path=config_path)

95def _extract_sync_intervals(config, log_messages: bool = False):

96 """

97 Extract drive and photos sync intervals from configuration.

99 Args:

100 config: Configuration dictionary

101 log_messages: Whether to log informational messages (default: False for loop usage)

102

103 Returns:

104 tuple: (drive_sync_interval, photos_sync_interval)

105 """

106 drive_sync_interval = 0

107 photos_sync_interval = 0

108

109 if config and "drive" in config:

110 drive_sync_interval = config_parser.get_drive_sync_interval(config=config, log_messages=log_messages)

111 if config and "photos" in config:

112 photos_sync_interval = config_parser.get_photos_sync_interval(config=config, log_messages=log_messages)

113

114 return drive_sync_interval, photos_sync_interval

115

116

117def _retrieve_password(username: str):

118 """

119 Retrieve password from environment or keyring.

120

121 Args:

122 username: iCloud username

123

124 Returns:

125 Password string or None if not found

126

127 Raises:

128 ICloudPyNoStoredPasswordAvailableException: If password not available

129 """

130 if ENV_ICLOUD_PASSWORD_KEY in os.environ:

131 password = os.environ.get(ENV_ICLOUD_PASSWORD_KEY)

132 utils.store_password_in_keyring(username=username, password=password)

133 return password

134 else:

135 return utils.get_password_from_keyring(username=username)

136

137

138def _authenticate_and_get_api(config, username: str):

139 """

140 Authenticate user and return iCloud API instance.

141

142 Args:

143 config: Configuration dictionary

144 username: iCloud username

145

146 Returns:

147 ICloudPyService instance

148

149 Raises:

150 ICloudPyNoStoredPasswordAvailableException: If password not available

151 """

152 server_region = config_parser.get_region(config=config)

153 password = _retrieve_password(username)

154 return get_api_instance(username=username, password=password, server_region=server_region)

155

156

157def _perform_drive_sync(config, api, sync_state: SyncState, drive_sync_interval: int):

158 """

159 Execute drive synchronization if enabled.

160

161 Args:

162 config: Configuration dictionary

163 api: iCloud API instance

164 sync_state: Current sync state

165 drive_sync_interval: Drive sync interval in seconds

166

167 Returns:

168 DriveStats object if sync was performed, None otherwise

169 """

170 if config and "drive" in config and sync_state.enable_sync_drive:

171 import time

172

173 from src.sync_stats import DriveStats

174

175 start_time = time.time()

176 stats = DriveStats()

177

178 destination_path = config_parser.prepare_drive_destination(config=config)

179

180 # Count files before sync

181 files_before = set()

182 if os.path.exists(destination_path):

183 try:

184 for root, _dirs, file_list in os.walk(destination_path):

185 for file in file_list:

186 files_before.add(os.path.join(root, file))

187 except Exception:

188 pass

189

190 LOGGER.info("Syncing drive...")

191 files_after = sync_drive.sync_drive(config=config, drive=api.drive)

192 LOGGER.info("Drive synced")

193

194 # Calculate statistics

195 stats.duration_seconds = time.time() - start_time

196

197 # Handle case where sync_drive returns None (e.g., in tests)

198 if files_after is not None:

199 # Count newly downloaded files

200 new_files = files_after - files_before

201 stats.files_downloaded = len(new_files)

202

203 # Count skipped files

204 stats.files_skipped = len(files_before & files_after)

205

206 # Count removed files

207 if config_parser.get_drive_remove_obsolete(config=config):

208 stats.files_removed = len(files_before - files_after)

209

210 # Calculate bytes downloaded

211 try:

212 for file_path in new_files:

213 if os.path.exists(file_path) and os.path.isfile(file_path):

214 stats.bytes_downloaded += os.path.getsize(file_path)

215 except Exception:

216 pass

217

218 # Reset countdown timer to the configured interval

219 sync_state.drive_time_remaining = drive_sync_interval

220 return stats

221 return None

222

223

224def _perform_photos_sync(config, api, sync_state: SyncState, photos_sync_interval: int):

225 """

226 Execute photos synchronization if enabled.

227

228 Args:

229 config: Configuration dictionary

230 api: iCloud API instance

231 sync_state: Current sync state

232 photos_sync_interval: Photos sync interval in seconds

233

234 Returns:

235 PhotoStats object if sync was performed, None otherwise

236 """

237 if config and "photos" in config and sync_state.enable_sync_photos:

238 import time

239

240 from src.sync_stats import PhotoStats

241

242 start_time = time.time()

243 stats = PhotoStats()

244

245 destination_path = config_parser.prepare_photos_destination(config=config)

246

247 # Count files before sync

248 files_before = set()

249 if os.path.exists(destination_path):

250 try:

251 for root, _dirs, file_list in os.walk(destination_path):

252 for file in file_list:

253 files_before.add(os.path.join(root, file))

254 except Exception:

255 pass

256

257 LOGGER.info("Syncing photos...")

258 sync_photos.sync_photos(config=config, photos=api.photos)

259 LOGGER.info("Photos synced")

260

261 # Count files after sync

262 files_after = set()

263 if os.path.exists(destination_path):

264 try:

265 for root, _dirs, file_list in os.walk(destination_path):

266 for file in file_list:

267 files_after.add(os.path.join(root, file))

268 except Exception:

269 pass

270

271 # Calculate statistics

272 stats.duration_seconds = time.time() - start_time

273

274 # Count newly downloaded files

275 new_files = files_after - files_before

276 stats.photos_downloaded = len(new_files)

277

278 # Estimate hardlinked photos (approximate)

279 use_hardlinks = config_parser.get_photos_use_hardlinks(config=config, log_messages=False)

280 if use_hardlinks:

281 stats.photos_hardlinked = max(0, len(files_after) - len(files_before) - stats.photos_downloaded)

282

283 # Count skipped photos

284 stats.photos_skipped = len(files_before & files_after)

285

286 # Calculate bytes downloaded

287 try:

288 for file_path in new_files:

289 if os.path.exists(file_path) and os.path.isfile(file_path):

290 stats.bytes_downloaded += os.path.getsize(file_path)

291

292 # Estimate bytes saved by hardlinks

293 if use_hardlinks and stats.photos_hardlinked > 0:

294 for file_path in files_after:

295 if file_path not in new_files and os.path.isfile(file_path):

296 stats.bytes_saved_by_hardlinks += os.path.getsize(file_path)

297 except Exception:

298 pass

299

300 # Get list of synced albums (simple approximation based on directories)

301 try:

302 for item in os.listdir(destination_path):

303 item_path = os.path.join(destination_path, item)

304 if os.path.isdir(item_path):

305 stats.albums_synced.append(item)

306 except Exception:

307 pass

308

309 # Reset countdown timer to the configured interval

310 sync_state.photos_time_remaining = photos_sync_interval

311 return stats

312 return None

313

314

315def _check_services_configured(config):

316 """

317 Check if any sync services are configured.

318

319 Args:

320 config: Configuration dictionary

321

322 Returns:

323 bool: True if at least one service is configured

324 """

325

326 return "drive" in config or "photos" in config

327

328

329def _send_usage_statistics(config, summary: SyncSummary) -> None:

330 """Send anonymized usage statistics.

331

332 Args:

333 config: Configuration dictionary

334 summary: Sync summary with statistics

335 """

336

337 # Create anonymized usage data

338 usage_data = {

339 "sync_duration": (

340 (summary.sync_end_time - summary.sync_start_time).total_seconds() if summary.sync_end_time else 0

341 ),

342 "has_drive_activity": bool(summary.drive_stats and summary.drive_stats.has_activity()),

343 "has_photos_activity": bool(summary.photo_stats and summary.photo_stats.has_activity()),

344 "has_errors": summary.has_errors(),

345 "timestamp": summary.sync_end_time.isoformat() if summary.sync_end_time else None,

346 }

347

348 # Add aggregated statistics (no personal data)

349 if summary.drive_stats:

350 usage_data["drive"] = {

351 "files_count": summary.drive_stats.files_downloaded,

352 "bytes_count": summary.drive_stats.bytes_downloaded,

353 "has_errors": summary.drive_stats.has_errors(),

354 }

355

356 if summary.photo_stats:

357 usage_data["photos"] = {

358 "photos_count": summary.photo_stats.photos_downloaded,

359 "bytes_count": summary.photo_stats.bytes_downloaded,

360 "hardlinks_count": summary.photo_stats.photos_hardlinked,

361 "has_errors": summary.photo_stats.has_errors(),

362 }

363

364 # Send to usage tracking

365 alive(config=config, data=usage_data)

366

367

368def _handle_2fa_required(config, username: str, sync_state: SyncState):

369 """

370 Handle 2FA authentication requirement.

371

372 Args:

373 config: Configuration dictionary

374 username: iCloud username

375 sync_state: Current sync state

376

377 Returns:

378 bool: True if should continue (retry), False if should exit

379 """

380 LOGGER.error("Error: 2FA is required. Please log in.")

381 sleep_for = config_parser.get_retry_login_interval(config=config)

382

383 if sleep_for < 0:

384 LOGGER.info("retry_login_interval is < 0, exiting ...")

385 return False

386

387 _log_retry_time(sleep_for)

388 server_region = config_parser.get_region(config=config)

389 sync_state.last_send = notify.send(

390 config=config,

391 username=username,

392 last_send=sync_state.last_send,

393 region=server_region,

394 )

395 sleep(sleep_for)

396 return True

397

398

399def _handle_password_error(config, username: str, sync_state: SyncState):

400 """

401 Handle password not available error.

402

403 Args:

404 config: Configuration dictionary

405 username: iCloud username

406 sync_state: Current sync state

407

408 Returns:

409 bool: True if should continue (retry), False if should exit

410 """

411 LOGGER.error("Password is not stored in keyring. Please save the password in keyring.")

412 sleep_for = config_parser.get_retry_login_interval(config=config)

413

414 if sleep_for < 0:

415 LOGGER.info("retry_login_interval is < 0, exiting ...")

416 return False

417

418 _log_retry_time(sleep_for)

419 server_region = config_parser.get_region(config=config)

420 sync_state.last_send = notify.send(

421 config=config,

422 username=username,

423 last_send=sync_state.last_send,

424 region=server_region,

425 )

426 sleep(sleep_for)

427 return True

428

429

430def _log_retry_time(sleep_for: int):

431 """

432 Log the next retry time.

433

434 Args:

435 sleep_for: Sleep duration in seconds

436 """

437 next_sync = (datetime.datetime.now() + datetime.timedelta(seconds=sleep_for)).strftime("%c")

438 LOGGER.info(f"Retrying login at {next_sync} ...")

439

440

441def _calculate_next_sync_schedule(config, sync_state: SyncState):

442 """

443 Calculate next sync schedule and update sync state.

444

445 This function implements the adaptive scheduling algorithm that determines

446 which service should sync next based on countdown timers.

447

448 Args:

449 config: Configuration dictionary

450 sync_state: Current sync state

451

452 Returns:

453 int: Sleep duration in seconds

454 """

455 has_drive = config and "drive" in config

456 has_photos = config and "photos" in config

457

458 if not has_drive and has_photos:

459 sleep_for = sync_state.photos_time_remaining

460 sync_state.enable_sync_drive = False

461 sync_state.enable_sync_photos = True

462 elif has_drive and not has_photos:

463 sleep_for = sync_state.drive_time_remaining

464 sync_state.enable_sync_drive = True

465 sync_state.enable_sync_photos = False

466 elif has_drive and has_photos and sync_state.drive_time_remaining <= sync_state.photos_time_remaining:

467 # Special case: if both timers are equal and large (> 10 seconds), wait for the full interval

468 # This fixes the bug where equal large intervals cause immediate re-sync

469 if sync_state.drive_time_remaining == sync_state.photos_time_remaining and sync_state.drive_time_remaining > 10:

470 sleep_for = sync_state.drive_time_remaining

471 sync_state.enable_sync_drive = True

472 sync_state.enable_sync_photos = True

473 else:

474 sleep_for = sync_state.photos_time_remaining - sync_state.drive_time_remaining

475 sync_state.photos_time_remaining -= sync_state.drive_time_remaining

476 sync_state.enable_sync_drive = True

477 sync_state.enable_sync_photos = False

478 else:

479 sleep_for = sync_state.drive_time_remaining - sync_state.photos_time_remaining

480 sync_state.drive_time_remaining -= sync_state.photos_time_remaining

481 sync_state.enable_sync_drive = False

482 sync_state.enable_sync_photos = True

483

484 return sleep_for

485

486

487def _log_next_sync_time(sleep_for: int):

488 """

489 Log the next scheduled sync time.

490

491 Args:

492 sleep_for: Sleep duration in seconds

493 """

494 next_sync = (datetime.datetime.now() + datetime.timedelta(seconds=sleep_for)).strftime("%c")

495 LOGGER.info(f"Resyncing at {next_sync} ...")

496

497

498def _log_sync_intervals_at_startup(config):

499 """

500 Log sync intervals once at startup.

501

502 Args:

503 config: Configuration dictionary

504 """

505 if config and "drive" in config:

506 config_parser.get_drive_sync_interval(config=config, log_messages=True)

507 if config and "photos" in config:

508 config_parser.get_photos_sync_interval(config=config, log_messages=True)

509

510

511def _should_exit_oneshot_mode(config):

512 """

513 Check if should exit in oneshot mode.

514

515 Oneshot mode exits when ALL configured sync intervals are negative.

516

517 Args:

518 config: Configuration dictionary

519

520 Returns:

521 bool: True if should exit

522 """

523

524 should_exit_drive = ("drive" not in config) or (

525 config_parser.get_drive_sync_interval(config=config, log_messages=False) < 0

526 )

527 should_exit_photos = ("photos" not in config) or (

528 config_parser.get_photos_sync_interval(config=config, log_messages=False) < 0

529 )

530

531 return should_exit_drive and should_exit_photos

532

533

534def sync():

535 """

536 Main synchronization loop.

537

538 Orchestrates the entire sync process by delegating specific responsibilities

539 to focused helper functions. This function coordinates the high-level flow

540 while each helper handles a single concern.

541 """

542 sync_state = SyncState()

543 startup_logged = False

544

545 while True:

546 config = _load_configuration()

547 alive(config=config)

548

549 # Log sync intervals once at startup

550 if not startup_logged:

551 _log_sync_intervals_at_startup(config)

552 startup_logged = True

553

554 drive_sync_interval, photos_sync_interval = _extract_sync_intervals(config, log_messages=False)

555 username = config_parser.get_username(config=config) if config else None

556

557 if username:

558 try:

559 api = _authenticate_and_get_api(config, username)

560

561 if not api.requires_2sa:

562 # Create summary for this sync cycle

563 summary = SyncSummary()

564

565 # Perform syncs and collect statistics

566 drive_stats = _perform_drive_sync(config, api, sync_state, drive_sync_interval)

567 photos_stats = _perform_photos_sync(config, api, sync_state, photos_sync_interval)

568

569 # Populate summary with statistics

570 summary.drive_stats = drive_stats

571 summary.photo_stats = photos_stats

572 summary.sync_end_time = datetime.datetime.now()

573

574 # Send usage statistics (anonymized summary data)

575 try:

576 _send_usage_statistics(config, summary)

577 except Exception as e:

578 LOGGER.debug(f"Failed to send usage statistics: {e!s}")

579

580 # Send sync summary notification if configured

581 # Only send notification when both enabled services have synced in this cycle

582 # Gracefully handle notification failures to not break sync

583 has_drive_config = config and "drive" in config

584 has_photos_config = config and "photos" in config

585

586 should_send_notification = False

587 if has_drive_config and has_photos_config:

588 # Both services configured - send notification only when both have synced

589 should_send_notification = drive_stats is not None and photos_stats is not None

590 elif has_drive_config and not has_photos_config:

591 # Only drive configured - send when drive synced

592 should_send_notification = drive_stats is not None

593 elif has_photos_config and not has_drive_config:

594 # Only photos configured - send when photos synced

595 should_send_notification = photos_stats is not None

596

597 if should_send_notification:

598 try:

599 notify.send_sync_summary(config=config, summary=summary)

600 except Exception as e:

601 LOGGER.debug(f"Failed to send sync summary notification: {e!s}")

602

603 if not _check_services_configured(config):

604 LOGGER.warning("Nothing to sync. Please add drive: and/or photos: section in config.yaml file.")

605 else:

606 if not _handle_2fa_required(config, username, sync_state):

607 break

608 continue

609

610 except exceptions.ICloudPyNoStoredPasswordAvailableException:

611 if not _handle_password_error(config, username, sync_state):

612 break

613 continue

614

615 sleep_for = _calculate_next_sync_schedule(config, sync_state)

616 _log_next_sync_time(sleep_for)

617

618 if _should_exit_oneshot_mode(config):

619 LOGGER.info("All configured sync intervals are negative, exiting oneshot mode...")

620 break

621

622 sleep(sleep_for)