Coverage for pyPhotoAlbum/project_serializer.py: 55%
204 statements
« prev ^ index » next coverage.py v7.14.1, created at 2026-06-20 12:55 +0000
1"""
2Project serialization to/from ZIP files for pyPhotoAlbum
3"""
5import os
6import json
7import zipfile
8import shutil
9import tempfile
10import threading
11from typing import Optional, Tuple, Callable
12from pathlib import Path
13from pyPhotoAlbum.project import Project
14from pyPhotoAlbum.version_manager import (
15 CURRENT_DATA_VERSION,
16 check_version_compatibility,
17 VersionCompatibility,
18 DataMigration,
19)
21# Legacy constant for backward compatibility
22SERIALIZATION_VERSION = CURRENT_DATA_VERSION
25def _import_external_images(project: Project):
26 """
27 Find and import any images that have external (absolute or non-assets) paths.
28 This ensures all images are in the assets folder before saving.
30 Args:
31 project: The Project instance to check
32 """
33 from pyPhotoAlbum.models import ImageData
35 imported_count = 0
37 for page in project.pages:
38 for element in page.layout.elements:
39 if isinstance(element, ImageData) and element.image_path:
40 # Check if this is an external path (absolute or not in assets/)
41 is_external = False
43 if os.path.isabs(element.image_path):
44 # Absolute path - definitely external
45 is_external = True
46 external_path = element.image_path
47 elif not element.image_path.startswith("assets/"):
48 # Relative path but not in assets folder
49 # Check if it exists relative to project folder
50 full_path = os.path.join(project.folder_path, element.image_path)
51 if os.path.exists(full_path) and not full_path.startswith(project.asset_manager.assets_folder):
52 is_external = True
53 external_path = full_path
54 else:
55 # Path doesn't exist - skip it (will be caught as missing asset)
56 continue
57 else:
58 # Already in assets/ folder
59 continue
61 # Import the external image
62 if is_external and os.path.exists(external_path):
63 try:
64 new_asset_path = project.asset_manager.import_asset(external_path)
65 element.image_path = new_asset_path
66 imported_count += 1
67 print(f"Auto-imported external image: {external_path} → {new_asset_path}")
68 except Exception as e:
69 print(f"Warning: Failed to import external image {external_path}: {e}")
71 if imported_count > 0:
72 print(f"Auto-imported {imported_count} external image(s) to assets folder")
75def _normalize_asset_paths(project: Project, project_folder: str):
76 """
77 Normalize asset paths in a loaded project to be relative to the project folder.
78 This fixes legacy projects that may have absolute paths or paths relative to old locations.
80 Args:
81 project: The Project instance to normalize
82 project_folder: The current project folder path
83 """
84 from pyPhotoAlbum.models import ImageData
86 normalized_count = 0
88 for page in project.pages:
89 for element in page.layout.elements:
90 if isinstance(element, ImageData) and element.image_path:
91 original_path = element.image_path
93 # Skip if already a simple relative path (assets/...)
94 if not os.path.isabs(original_path) and not original_path.startswith("./projects/"):
95 continue
97 # Try to extract just the filename or relative path from assets folder
98 # Pattern 1: "./projects/XXX/assets/filename.jpg" -> "assets/filename.jpg"
99 if "/assets/" in original_path:
100 parts = original_path.split("/assets/")
101 if len(parts) == 2:
102 new_path = os.path.join("assets", parts[1])
103 element.image_path = new_path
104 normalized_count += 1
105 print(f"Normalized path: {original_path} -> {new_path}")
106 continue
108 # Pattern 2: Absolute path - try to make it relative if it's in the extraction folder
109 if os.path.isabs(original_path):
110 try:
111 new_path = os.path.relpath(original_path, project_folder)
112 element.image_path = new_path
113 normalized_count += 1
114 print(f"Normalized absolute path: {original_path} -> {new_path}")
115 except ValueError:
116 # Can't make relative (different drives on Windows, etc.)
117 pass
119 if normalized_count > 0:
120 print(f"Normalized {normalized_count} asset paths")
123def save_to_zip(project: Project, zip_path: str) -> Tuple[bool, Optional[str]]:
124 """
125 Save a project to a ZIP file, including all assets.
127 Args:
128 project: The Project instance to save
129 zip_path: Path where the ZIP file should be created
131 Returns:
132 Tuple of (success: bool, error_message: Optional[str])
133 """
134 try:
135 # Ensure .ppz extension
136 if not zip_path.lower().endswith(".ppz"):
137 zip_path += ".ppz"
139 # Check for and import any external images before saving
140 _import_external_images(project)
142 # Serialize project to dictionary
143 project_data = project.serialize()
145 # Add version information
146 project_data["serialization_version"] = SERIALIZATION_VERSION # Legacy field
147 project_data["data_version"] = CURRENT_DATA_VERSION # New versioning system
149 # Create ZIP file
150 with zipfile.ZipFile(zip_path, "w", zipfile.ZIP_DEFLATED) as zipf:
151 # Write project.json with stable sorting for git-friendly diffs
152 project_json = json.dumps(project_data, indent=2, sort_keys=True)
153 zipf.writestr("project.json", project_json)
155 # Add all files from the assets folder
156 assets_folder = project.asset_manager.assets_folder
157 if os.path.exists(assets_folder):
158 for root, dirs, files in os.walk(assets_folder):
159 for file in files:
160 file_path = os.path.join(root, file)
161 # Store with relative path from project folder
162 arcname = os.path.relpath(file_path, project.folder_path)
163 zipf.write(file_path, arcname)
165 print(f"Project saved to {zip_path}")
166 return True, None
168 except Exception as e:
169 error_msg = f"Error saving project: {str(e)}"
170 print(error_msg)
171 return False, error_msg
174def save_to_zip_async(
175 project: Project,
176 zip_path: str,
177 on_complete: Optional[Callable[[bool, Optional[str]], None]] = None,
178 on_progress: Optional[Callable[[int, str], None]] = None,
179) -> threading.Thread:
180 """
181 Save a project to a ZIP file asynchronously in a background thread.
183 This provides instant UI responsiveness by:
184 1. Immediately serializing project.json to a temp folder (fast)
185 2. Creating the ZIP file in a background thread (slow)
186 3. Calling on_complete when done
188 Args:
189 project: The Project instance to save
190 zip_path: Path where the ZIP file should be created
191 on_complete: Optional callback(success: bool, error_msg: Optional[str])
192 called when save completes
193 on_progress: Optional callback(progress: int, message: str) where
194 progress is 0-100 and message describes current step
196 Returns:
197 The background thread (already started)
198 """
199 # Ensure .ppz extension
200 final_zip_path = zip_path
201 if not final_zip_path.lower().endswith(".ppz"):
202 final_zip_path += ".ppz"
204 # ---- Work done on the CALLING (main) thread ----
205 # Serialization is pure Python and holds the GIL, but it's fast.
206 # Doing it here keeps the background thread to file-I/O only, which
207 # releases the GIL and keeps the UI responsive.
208 if on_progress:
209 on_progress(0, "Preparing to save...")
211 _import_external_images(project)
213 if on_progress:
214 on_progress(10, "Serializing project data...")
215 project_data = project.serialize()
216 project_data["serialization_version"] = SERIALIZATION_VERSION
217 project_data["data_version"] = CURRENT_DATA_VERSION
218 project_json_str = json.dumps(project_data, indent=2, sort_keys=True)
220 # Collect the asset file list now so the background thread doesn't
221 # need to touch the (potentially temporary) project folder.
222 assets_folder = project.asset_manager.assets_folder
223 folder_path = project.folder_path
224 asset_files: list[tuple[str, str]] = []
225 if os.path.exists(assets_folder):
226 for root, _dirs, files in os.walk(assets_folder):
227 for file in files:
228 file_path = os.path.join(root, file)
229 arcname = os.path.relpath(file_path, folder_path)
230 asset_files.append((file_path, arcname))
232 total_files = 1 + len(asset_files) # project.json + assets
234 if on_progress:
235 on_progress(20, f"Starting background write ({total_files} files)...")
237 # ---- Work done on the BACKGROUND thread ----
238 # Only file I/O here — zipfile/zlib/shutil all release the GIL.
239 def _background_save():
240 """Background thread: write ZIP file from pre-serialized data."""
241 temp_dir = None
242 try:
243 temp_dir = tempfile.mkdtemp(prefix="pyPhotoAlbum_save_")
244 temp_zip_path = os.path.join(temp_dir, "project.ppz")
246 if on_progress:
247 on_progress(25, f"Creating ZIP archive ({total_files} files)...")
249 with zipfile.ZipFile(temp_zip_path, "w", zipfile.ZIP_DEFLATED) as zipf:
250 zipf.writestr("project.json", project_json_str)
252 if asset_files:
253 progress_range = 90 - 25
254 for idx, (file_path, arcname) in enumerate(asset_files):
255 zipf.write(file_path, arcname)
256 if idx % 10 == 0 or idx == len(asset_files) - 1:
257 progress = 25 + int((idx + 1) / len(asset_files) * progress_range)
258 if on_progress:
259 on_progress(progress, f"Adding assets... ({idx + 1}/{len(asset_files)})")
261 if on_progress:
262 on_progress(95, "Finalizing save...")
264 os.makedirs(os.path.dirname(os.path.abspath(final_zip_path)), exist_ok=True)
265 if os.path.exists(final_zip_path):
266 os.remove(final_zip_path)
267 shutil.move(temp_zip_path, final_zip_path)
269 if on_progress:
270 on_progress(100, "Save complete!")
272 print(f"Project saved to {final_zip_path}")
274 if on_complete:
275 on_complete(True, None)
277 except Exception as e:
278 error_msg = f"Error saving project: {str(e)}"
279 print(error_msg)
280 if on_complete:
281 on_complete(False, error_msg)
283 finally:
284 if temp_dir and os.path.exists(temp_dir):
285 try:
286 shutil.rmtree(temp_dir)
287 except Exception:
288 pass
290 save_thread = threading.Thread(target=_background_save, daemon=True)
291 save_thread.start()
292 return save_thread
295def load_from_zip(zip_path: str, extract_to: Optional[str] = None) -> Project:
296 """
297 Load a project from a ZIP file.
299 Args:
300 zip_path: Path to the ZIP file to load
301 extract_to: Optional directory to extract to. If None, uses a temporary
302 directory that will be cleaned up when the project is closed.
304 Returns:
305 Project instance (raises exception on error)
306 """
307 if not os.path.exists(zip_path):
308 raise FileNotFoundError(f"ZIP file not found: {zip_path}")
310 # Track if we created a temp directory
311 temp_dir_obj = None
313 # Determine extraction directory
314 if extract_to is None:
315 # Create a temporary directory using TemporaryDirectory
316 # This will be attached to the Project and auto-cleaned on deletion
317 zip_basename = os.path.splitext(os.path.basename(zip_path))[0]
318 temp_dir_obj = tempfile.TemporaryDirectory(prefix=f"pyPhotoAlbum_{zip_basename}_")
319 extract_to = temp_dir_obj.name
320 else:
321 # Create extraction directory if it doesn't exist
322 os.makedirs(extract_to, exist_ok=True)
324 # Extract ZIP contents
325 with zipfile.ZipFile(zip_path, "r") as zipf:
326 zipf.extractall(extract_to)
328 # Load project.json
329 project_json_path = os.path.join(extract_to, "project.json")
330 if not os.path.exists(project_json_path):
331 raise ValueError("Invalid project file: project.json not found")
333 with open(project_json_path, "r") as f:
334 project_data = json.load(f)
336 # Check version compatibility
337 # Try new version field first, fall back to legacy field
338 file_version = project_data.get("data_version", project_data.get("serialization_version", "1.0"))
340 # Check if version is compatible
341 is_compatible, error_msg = check_version_compatibility(file_version, zip_path)
342 if not is_compatible:
343 raise ValueError(error_msg)
345 # Apply migrations if needed
346 if VersionCompatibility.needs_migration(file_version):
347 print(f"Migrating project from version {file_version} to {CURRENT_DATA_VERSION}...")
348 project_data = DataMigration.migrate(project_data, file_version, CURRENT_DATA_VERSION)
349 print(f"Migration completed successfully")
350 elif file_version != CURRENT_DATA_VERSION:
351 print(f"Note: Loading project with version {file_version}, current version is {CURRENT_DATA_VERSION}")
353 # Create new project
354 project_name = project_data.get("name", "Untitled Project")
355 project = Project(name=project_name, folder_path=extract_to)
357 # Deserialize project data
358 project.deserialize(project_data)
360 # Update folder path to extraction location
361 project.folder_path = extract_to
362 project.asset_manager.project_folder = extract_to
363 project.asset_manager.assets_folder = os.path.join(extract_to, "assets")
365 # Attach temporary directory to project (if we created one)
366 # The TemporaryDirectory will auto-cleanup when the project is deleted
367 if temp_dir_obj is not None:
368 project._temp_dir = temp_dir_obj
369 print(f"Project loaded to temporary directory: {extract_to}")
371 # Normalize asset paths in all ImageData elements
372 # This fixes old projects that have absolute or wrong relative paths
373 _normalize_asset_paths(project, extract_to)
375 # Set asset resolution context for ImageData rendering
376 # Only set project folder - search paths are reserved for healing functionality
377 from pyPhotoAlbum.models import set_asset_resolution_context
379 set_asset_resolution_context(extract_to)
381 print(f"Project loaded from {zip_path} to {extract_to}")
382 return project
385def get_project_info(zip_path: str) -> Optional[dict]:
386 """
387 Get basic information about a project without fully loading it.
389 Args:
390 zip_path: Path to the ZIP file
392 Returns:
393 Dictionary with project info, or None if error
394 """
395 try:
396 with zipfile.ZipFile(zip_path, "r") as zipf:
397 # Read project.json
398 project_json = zipf.read("project.json").decode("utf-8")
399 project_data = json.loads(project_json)
401 return {
402 "name": project_data.get("name", "Unknown"),
403 "version": project_data.get("serialization_version", "Unknown"),
404 "page_count": len(project_data.get("pages", [])),
405 "page_size_mm": project_data.get("page_size_mm", (0, 0)),
406 "working_dpi": project_data.get("working_dpi", 300),
407 }
408 except Exception as e:
409 print(f"Error reading project info: {e}")
410 return None