Coverage for src/sync.py: 100%

1"""Sync module.""" 

2 

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

4import datetime 

5import os 

6from time import sleep 

7 

8from icloudpy import ICloudPyService, exceptions, utils 

9 

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

17 notify, 

18 read_config, 

19 sync_drive, 

20 sync_photos, 

21) 

22from src.usage import alive 

23 

24LOGGER = get_logger() 

25 

26 

27def get_api_instance( 

28 username, 

29 password, 

30 cookie_directory=DEFAULT_COOKIE_DIRECTORY, 

31 server_region="global", 

32): 

33 """Get API client instance.""" 

34 return ( 

35 ICloudPyService( 

36 apple_id=username, 

37 password=password, 

38 cookie_directory=cookie_directory, 

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

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

41 ) 

42 if server_region == "china" 

43 else ICloudPyService( 

44 apple_id=username, 

45 password=password, 

46 cookie_directory=cookie_directory, 

47 ) 

48 ) 

49 

50 

51def sync(): 

52 """Sync data from server.""" 

53 last_send = None 

54 enable_sync_drive = True 

55 enable_sync_photos = True 

56 drive_sync_interval = 0 

57 photos_sync_interval = 0 

58 sleep_for = 10 

59 

60 while True: 

61 config = read_config(config_path=os.environ.get(ENV_CONFIG_FILE_PATH_KEY, DEFAULT_CONFIG_FILE_PATH)) 

62 alive(config=config) 

63 username = config_parser.get_username(config=config) 

64 if username: 

65 try: 

66 server_region = config_parser.get_region(config=config) 

67 if ENV_ICLOUD_PASSWORD_KEY in os.environ: 

68 password = os.environ.get(ENV_ICLOUD_PASSWORD_KEY) 

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

70 else: 

71 password = utils.get_password_from_keyring(username=username) 

72 api = get_api_instance(username=username, password=password, server_region=server_region) 

73 if not api.requires_2sa: 

74 if "drive" in config and enable_sync_drive: 

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

76 sync_drive.sync_drive(config=config, drive=api.drive) 

77 LOGGER.info("Drive synced") 

78 drive_sync_interval = config_parser.get_drive_sync_interval(config=config) 

79 if "photos" in config and enable_sync_photos: 

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

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

82 LOGGER.info("Photos synced") 

83 photos_sync_interval = config_parser.get_photos_sync_interval(config=config) 

84 if "drive" not in config and "photos" not in config: 

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

86 else: 

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

88 # Retry again 

89 sleep_for = config_parser.get_retry_login_interval(config=config) 

90 if sleep_for < 0: 

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

92 break 

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

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

95 last_send = notify.send(config=config, username=username, last_send=last_send, region=server_region) 

96 sleep(sleep_for) 

97 continue 

98 except exceptions.ICloudPyNoStoredPasswordAvailableException: 

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

100 sleep_for = config_parser.get_retry_login_interval(config=config) 

101 if sleep_for < 0: 

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

103 break 

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

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

106 last_send = notify.send(config=config, username=username, last_send=last_send, region=server_region) 

107 sleep(sleep_for) 

108 continue 

109 

110 if "drive" not in config and "photos" in config: 

111 sleep_for = photos_sync_interval 

112 enable_sync_drive = False 

113 enable_sync_photos = True 

114 elif "drive" in config and "photos" not in config: 

115 sleep_for = drive_sync_interval 

116 enable_sync_drive = True 

117 enable_sync_photos = False 

118 elif "drive" in config and "photos" in config and drive_sync_interval <= photos_sync_interval: 

119 sleep_for = photos_sync_interval - drive_sync_interval 

120 photos_sync_interval -= drive_sync_interval 

121 enable_sync_drive = True 

122 enable_sync_photos = False 

123 else: 

124 sleep_for = drive_sync_interval - photos_sync_interval 

125 drive_sync_interval -= photos_sync_interval 

126 enable_sync_drive = False 

127 enable_sync_photos = True 

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

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

130 if ( 

131 config_parser.get_drive_sync_interval(config=config) < 0 

132 if "drive" in config 

133 else True and config_parser.get_photos_sync_interval(config=config) < 0 

134 if "photos" in config 

135 else True 

136 ): 

137 break 

138 sleep(sleep_for)