Coverage for pyPhotoAlbum/autosave_manager.py: 93%

1""" 

2Autosave and checkpoint management for pyPhotoAlbum. 

3 

4This module provides automatic checkpoint creation and recovery functionality 

5to prevent data loss from crashes or unexpected exits. 

6""" 

7 

8import os 

9import json 

10import shutil 

11from pathlib import Path 

12from datetime import datetime, timedelta 

13from typing import Dict, Optional, List, Tuple 

14from pyPhotoAlbum.project_serializer import save_to_zip, load_from_zip 

15 

16 

17class AutosaveManager: 

18 """Manages autosave checkpoints for projects.""" 

19 

20 CHECKPOINT_DIR = Path.home() / ".pyphotoalbum" / "checkpoints" 

21 CHECKPOINT_PREFIX = "checkpoint_" 

22 CHECKPOINT_EXTENSION = ".ppz" 

23 

24 def __init__(self): 

25 """Initialize the autosave manager.""" 

26 self._ensure_checkpoint_directory() 

27 

28 def _ensure_checkpoint_directory(self): 

29 """Ensure the checkpoint directory exists.""" 

30 self.CHECKPOINT_DIR.mkdir(parents=True, exist_ok=True) 

31 

32 def _get_checkpoint_path(self, project_name: str, timestamp: Optional[datetime] = None) -> Path: 

33 """ 

34 Get the path for a checkpoint file. 

35 

36 Args: 

37 project_name: Name of the project 

38 timestamp: Optional timestamp, defaults to current time 

39 

40 Returns: 

41 Path to the checkpoint file 

42 """ 

43 if timestamp is None: 

44 timestamp = datetime.now() 

45 

46 # Sanitize project name for filename 

47 safe_name = "".join(c if c.isalnum() or c in "-_" else "_" for c in project_name) 

48 timestamp_str = timestamp.strftime("%Y%m%d_%H%M%S") 

49 filename = f"{self.CHECKPOINT_PREFIX}{safe_name}_{timestamp_str}{self.CHECKPOINT_EXTENSION}" 

50 

51 return self.CHECKPOINT_DIR / filename 

52 

53 def create_checkpoint(self, project) -> Tuple[bool, str]: 

54 """ 

55 Create a checkpoint for the given project. 

56 

57 Args: 

58 project: Project instance to checkpoint 

59 

60 Returns: 

61 Tuple of (success: bool, message: str) 

62 """ 

63 try: 

64 checkpoint_path = self._get_checkpoint_path(project.name) 

65 success, message = save_to_zip(project, str(checkpoint_path)) 

66 

67 if success: 

68 # Also save metadata about this checkpoint 

69 self._save_checkpoint_metadata(project, checkpoint_path) 

70 return True, f"Checkpoint created: {checkpoint_path.name}" 

71 else: 

72 return False, f"Checkpoint failed: {message}" 

73 

74 except Exception as e: 

75 return False, f"Checkpoint error: {str(e)}" 

76 

77 def _save_checkpoint_metadata(self, project, checkpoint_path: Path): 

78 """ 

79 Save metadata about a checkpoint. 

80 

81 Args: 

82 project: Project instance 

83 checkpoint_path: Path to the checkpoint file 

84 """ 

85 metadata = { 

86 "project_name": project.name, 

87 "timestamp": datetime.now().isoformat(), 

88 "checkpoint_path": str(checkpoint_path), 

89 "original_path": getattr(project, "file_path", None), 

90 } 

91 

92 metadata_path = checkpoint_path.with_suffix(".json") 

93 with open(metadata_path, "w") as f: 

94 json.dump(metadata, f, indent=2) 

95 

96 def list_checkpoints(self, project_name: Optional[str] = None) -> List[Tuple[Path, dict]]: 

97 """ 

98 List available checkpoints. 

99 

100 Args: 

101 project_name: Optional filter by project name 

102 

103 Returns: 

104 List of tuples (checkpoint_path, metadata) 

105 """ 

106 checkpoints = [] 

107 

108 for checkpoint_file in self.CHECKPOINT_DIR.glob(f"{self.CHECKPOINT_PREFIX}*{self.CHECKPOINT_EXTENSION}"): 

109 metadata_file = checkpoint_file.with_suffix(".json") 

110 

111 # Try to load metadata 

112 metadata = {} 

113 if metadata_file.exists(): 

114 try: 

115 with open(metadata_file, "r") as f: 

116 metadata = json.load(f) 

117 except: 

118 pass 

119 

120 # Filter by project name if specified 

121 if project_name is None or metadata.get("project_name") == project_name: 

122 checkpoints.append((checkpoint_file, metadata)) 

123 

124 # Sort by timestamp (newest first) 

125 checkpoints.sort(key=lambda x: x[1].get("timestamp", ""), reverse=True) 

126 return checkpoints 

127 

128 def load_checkpoint(self, checkpoint_path: Path): 

129 """ 

130 Load a project from a checkpoint. 

131 

132 Args: 

133 checkpoint_path: Path to the checkpoint file 

134 

135 Returns: 

136 Tuple of (success: bool, project or error_message) 

137 """ 

138 try: 

139 project = load_from_zip(str(checkpoint_path)) 

140 return True, project 

141 except Exception as e: 

142 return False, f"Failed to load checkpoint: {str(e)}" 

143 

144 def delete_checkpoint(self, checkpoint_path: Path) -> bool: 

145 """ 

146 Delete a checkpoint file and its metadata. 

147 

148 Args: 

149 checkpoint_path: Path to the checkpoint file 

150 

151 Returns: 

152 True if successful 

153 """ 

154 try: 

155 # Delete checkpoint file 

156 if checkpoint_path.exists(): 

157 checkpoint_path.unlink() 

158 

159 # Delete metadata file 

160 metadata_path = checkpoint_path.with_suffix(".json") 

161 if metadata_path.exists(): 

162 metadata_path.unlink() 

163 

164 return True 

165 except Exception as e: 

166 print(f"Error deleting checkpoint: {e}") 

167 return False 

168 

169 def delete_all_checkpoints(self, project_name: Optional[str] = None): 

170 """ 

171 Delete all checkpoints, optionally filtered by project name. 

172 

173 Args: 

174 project_name: Optional filter by project name 

175 """ 

176 checkpoints = self.list_checkpoints(project_name) 

177 for checkpoint_path, _ in checkpoints: 

178 self.delete_checkpoint(checkpoint_path) 

179 

180 def cleanup_old_checkpoints(self, max_age_hours: int = 24 * 7, max_count: int = 50): 

181 """ 

182 Clean up old checkpoints to prevent unlimited growth. 

183 

184 Args: 

185 max_age_hours: Maximum age in hours (default: 7 days) 

186 max_count: Maximum number of checkpoints to keep per project 

187 """ 

188 now = datetime.now() 

189 checkpoints_by_project: Dict[str, List[Tuple[Path, dict]]] = {} 

190 

191 # Group checkpoints by project 

192 for checkpoint_path, metadata in self.list_checkpoints(): 

193 project_name = metadata.get("project_name", "unknown") 

194 if project_name not in checkpoints_by_project: 

195 checkpoints_by_project[project_name] = [] 

196 checkpoints_by_project[project_name].append((checkpoint_path, metadata)) 

197 

198 # Clean up each project's checkpoints 

199 for project_name, checkpoints in checkpoints_by_project.items(): 

200 # Sort by timestamp (newest first) 

201 checkpoints.sort(key=lambda x: x[1].get("timestamp", ""), reverse=True) 

202 

203 for idx, (checkpoint_path, metadata) in enumerate(checkpoints): 

204 # Delete if too old 

205 timestamp_str = metadata.get("timestamp") 

206 if timestamp_str: 

207 try: 

208 timestamp = datetime.fromisoformat(timestamp_str) 

209 age = now - timestamp 

210 if age > timedelta(hours=max_age_hours): 

211 self.delete_checkpoint(checkpoint_path) 

212 continue 

213 except: 

214 pass 

215 

216 # Delete if beyond max count 

217 if idx >= max_count: 

218 self.delete_checkpoint(checkpoint_path) 

219 

220 def has_checkpoints(self, project_name: Optional[str] = None) -> bool: 

221 """ 

222 Check if there are any checkpoints available. 

223 

224 Args: 

225 project_name: Optional filter by project name 

226 

227 Returns: 

228 True if checkpoints exist 

229 """ 

230 return len(self.list_checkpoints(project_name)) > 0 

231 

232 def get_latest_checkpoint(self, project_name: Optional[str] = None) -> Optional[Tuple[Path, dict]]: 

233 """ 

234 Get the most recent checkpoint. 

235 

236 Args: 

237 project_name: Optional filter by project name 

238 

239 Returns: 

240 Tuple of (checkpoint_path, metadata) or None 

241 """ 

242 checkpoints = self.list_checkpoints(project_name) 

243 if checkpoints: 

244 return checkpoints[0] # Already sorted newest first 

245 return None