Duncan Tourolle
f96200c799
Some checks failed
Lint / lint (push) Successful in 1m47s
Tests / test (3.11) (push) Successful in 57s
Tests / test (3.12) (push) Successful in 55s
Tests / test (3.13) (push) Successful in 56s
Tests / test (3.14) (push) Successful in 1m3s
Python CI / test (push) Failing after 6m1s
415 lines
16 KiB
Python
415 lines
16 KiB
Python
"""
|
|
Project serialization to/from ZIP files for pyPhotoAlbum
|
|
"""
|
|
|
|
import os
|
|
import json
|
|
import zipfile
|
|
import shutil
|
|
import tempfile
|
|
import threading
|
|
from typing import Optional, Tuple, Callable
|
|
from pathlib import Path
|
|
from pyPhotoAlbum.project import Project
|
|
from pyPhotoAlbum.version_manager import (
|
|
CURRENT_DATA_VERSION,
|
|
check_version_compatibility,
|
|
VersionCompatibility,
|
|
DataMigration,
|
|
)
|
|
|
|
|
|
# Legacy constant for backward compatibility
|
|
SERIALIZATION_VERSION = CURRENT_DATA_VERSION
|
|
|
|
|
|
def _import_external_images(project: Project):
|
|
"""
|
|
Find and import any images that have external (absolute or non-assets) paths.
|
|
This ensures all images are in the assets folder before saving.
|
|
|
|
Args:
|
|
project: The Project instance to check
|
|
"""
|
|
from pyPhotoAlbum.models import ImageData
|
|
|
|
imported_count = 0
|
|
|
|
for page in project.pages:
|
|
for element in page.layout.elements:
|
|
if isinstance(element, ImageData) and element.image_path:
|
|
# Check if this is an external path (absolute or not in assets/)
|
|
is_external = False
|
|
|
|
if os.path.isabs(element.image_path):
|
|
# Absolute path - definitely external
|
|
is_external = True
|
|
external_path = element.image_path
|
|
elif not element.image_path.startswith("assets/"):
|
|
# Relative path but not in assets folder
|
|
# Check if it exists relative to project folder
|
|
full_path = os.path.join(project.folder_path, element.image_path)
|
|
if os.path.exists(full_path) and not full_path.startswith(project.asset_manager.assets_folder):
|
|
is_external = True
|
|
external_path = full_path
|
|
else:
|
|
# Path doesn't exist - skip it (will be caught as missing asset)
|
|
continue
|
|
else:
|
|
# Already in assets/ folder
|
|
continue
|
|
|
|
# Import the external image
|
|
if is_external and os.path.exists(external_path):
|
|
try:
|
|
new_asset_path = project.asset_manager.import_asset(external_path)
|
|
element.image_path = new_asset_path
|
|
imported_count += 1
|
|
print(f"Auto-imported external image: {external_path} → {new_asset_path}")
|
|
except Exception as e:
|
|
print(f"Warning: Failed to import external image {external_path}: {e}")
|
|
|
|
if imported_count > 0:
|
|
print(f"Auto-imported {imported_count} external image(s) to assets folder")
|
|
|
|
|
|
def _normalize_asset_paths(project: Project, project_folder: str):
|
|
"""
|
|
Normalize asset paths in a loaded project to be relative to the project folder.
|
|
This fixes legacy projects that may have absolute paths or paths relative to old locations.
|
|
|
|
Args:
|
|
project: The Project instance to normalize
|
|
project_folder: The current project folder path
|
|
"""
|
|
from pyPhotoAlbum.models import ImageData
|
|
|
|
normalized_count = 0
|
|
|
|
for page in project.pages:
|
|
for element in page.layout.elements:
|
|
if isinstance(element, ImageData) and element.image_path:
|
|
original_path = element.image_path
|
|
|
|
# Skip if already a simple relative path (assets/...)
|
|
if not os.path.isabs(original_path) and not original_path.startswith("./projects/"):
|
|
continue
|
|
|
|
# Try to extract just the filename or relative path from assets folder
|
|
# Pattern 1: "./projects/XXX/assets/filename.jpg" -> "assets/filename.jpg"
|
|
if "/assets/" in original_path:
|
|
parts = original_path.split("/assets/")
|
|
if len(parts) == 2:
|
|
new_path = os.path.join("assets", parts[1])
|
|
element.image_path = new_path
|
|
normalized_count += 1
|
|
print(f"Normalized path: {original_path} -> {new_path}")
|
|
continue
|
|
|
|
# Pattern 2: Absolute path - try to make it relative if it's in the extraction folder
|
|
if os.path.isabs(original_path):
|
|
try:
|
|
new_path = os.path.relpath(original_path, project_folder)
|
|
element.image_path = new_path
|
|
normalized_count += 1
|
|
print(f"Normalized absolute path: {original_path} -> {new_path}")
|
|
except ValueError:
|
|
# Can't make relative (different drives on Windows, etc.)
|
|
pass
|
|
|
|
if normalized_count > 0:
|
|
print(f"Normalized {normalized_count} asset paths")
|
|
|
|
|
|
def save_to_zip(project: Project, zip_path: str) -> Tuple[bool, Optional[str]]:
|
|
"""
|
|
Save a project to a ZIP file, including all assets.
|
|
|
|
Args:
|
|
project: The Project instance to save
|
|
zip_path: Path where the ZIP file should be created
|
|
|
|
Returns:
|
|
Tuple of (success: bool, error_message: Optional[str])
|
|
"""
|
|
try:
|
|
# Ensure .ppz extension
|
|
if not zip_path.lower().endswith(".ppz"):
|
|
zip_path += ".ppz"
|
|
|
|
# Check for and import any external images before saving
|
|
_import_external_images(project)
|
|
|
|
# Serialize project to dictionary
|
|
project_data = project.serialize()
|
|
|
|
# Add version information
|
|
project_data["serialization_version"] = SERIALIZATION_VERSION # Legacy field
|
|
project_data["data_version"] = CURRENT_DATA_VERSION # New versioning system
|
|
|
|
# Create ZIP file
|
|
with zipfile.ZipFile(zip_path, "w", zipfile.ZIP_DEFLATED) as zipf:
|
|
# Write project.json with stable sorting for git-friendly diffs
|
|
project_json = json.dumps(project_data, indent=2, sort_keys=True)
|
|
zipf.writestr("project.json", project_json)
|
|
|
|
# Add all files from the assets folder
|
|
assets_folder = project.asset_manager.assets_folder
|
|
if os.path.exists(assets_folder):
|
|
for root, dirs, files in os.walk(assets_folder):
|
|
for file in files:
|
|
file_path = os.path.join(root, file)
|
|
# Store with relative path from project folder
|
|
arcname = os.path.relpath(file_path, project.folder_path)
|
|
zipf.write(file_path, arcname)
|
|
|
|
print(f"Project saved to {zip_path}")
|
|
return True, None
|
|
|
|
except Exception as e:
|
|
error_msg = f"Error saving project: {str(e)}"
|
|
print(error_msg)
|
|
return False, error_msg
|
|
|
|
|
|
def save_to_zip_async(
|
|
project: Project,
|
|
zip_path: str,
|
|
on_complete: Optional[Callable[[bool, Optional[str]], None]] = None,
|
|
on_progress: Optional[Callable[[int, str], None]] = None,
|
|
) -> threading.Thread:
|
|
"""
|
|
Save a project to a ZIP file asynchronously in a background thread.
|
|
|
|
This provides instant UI responsiveness by:
|
|
1. Immediately serializing project.json to a temp folder (fast)
|
|
2. Creating the ZIP file in a background thread (slow)
|
|
3. Calling on_complete when done
|
|
|
|
Args:
|
|
project: The Project instance to save
|
|
zip_path: Path where the ZIP file should be created
|
|
on_complete: Optional callback(success: bool, error_msg: Optional[str])
|
|
called when save completes
|
|
on_progress: Optional callback(progress: int, message: str) where
|
|
progress is 0-100 and message describes current step
|
|
|
|
Returns:
|
|
The background thread (already started)
|
|
"""
|
|
# Ensure .ppz extension
|
|
final_zip_path = zip_path
|
|
if not final_zip_path.lower().endswith(".ppz"):
|
|
final_zip_path += ".ppz"
|
|
|
|
# ---- Work done on the CALLING (main) thread ----
|
|
# Serialization is pure Python and holds the GIL, but it's fast.
|
|
# Doing it here keeps the background thread to file-I/O only, which
|
|
# releases the GIL and keeps the UI responsive.
|
|
if on_progress:
|
|
on_progress(0, "Preparing to save...")
|
|
|
|
_import_external_images(project)
|
|
|
|
if on_progress:
|
|
on_progress(10, "Serializing project data...")
|
|
project_data = project.serialize()
|
|
project_data["serialization_version"] = SERIALIZATION_VERSION
|
|
project_data["data_version"] = CURRENT_DATA_VERSION
|
|
project_json_str = json.dumps(project_data, indent=2, sort_keys=True)
|
|
|
|
# Collect the asset file list now so the background thread doesn't
|
|
# need to touch the (potentially temporary) project folder.
|
|
assets_folder = project.asset_manager.assets_folder
|
|
folder_path = project.folder_path
|
|
asset_files: list[tuple[str, str]] = []
|
|
if os.path.exists(assets_folder):
|
|
for root, _dirs, files in os.walk(assets_folder):
|
|
for file in files:
|
|
file_path = os.path.join(root, file)
|
|
arcname = os.path.relpath(file_path, folder_path)
|
|
asset_files.append((file_path, arcname))
|
|
|
|
total_files = 1 + len(asset_files) # project.json + assets
|
|
|
|
if on_progress:
|
|
on_progress(20, f"Starting background write ({total_files} files)...")
|
|
|
|
# ---- Work done on the BACKGROUND thread ----
|
|
# Only file I/O here — zipfile/zlib/shutil all release the GIL.
|
|
def _background_save():
|
|
"""Background thread: write ZIP file from pre-serialized data."""
|
|
temp_dir = None
|
|
try:
|
|
temp_dir = tempfile.mkdtemp(prefix="pyPhotoAlbum_save_")
|
|
temp_zip_path = os.path.join(temp_dir, "project.ppz")
|
|
|
|
if on_progress:
|
|
on_progress(25, f"Creating ZIP archive ({total_files} files)...")
|
|
|
|
with zipfile.ZipFile(temp_zip_path, "w", zipfile.ZIP_DEFLATED) as zipf:
|
|
zipf.writestr("project.json", project_json_str)
|
|
|
|
if asset_files:
|
|
progress_range = 90 - 25
|
|
for idx, (file_path, arcname) in enumerate(asset_files):
|
|
zipf.write(file_path, arcname)
|
|
if idx % 10 == 0 or idx == len(asset_files) - 1:
|
|
progress = 25 + int((idx + 1) / len(asset_files) * progress_range)
|
|
if on_progress:
|
|
on_progress(
|
|
progress,
|
|
f"Adding assets... ({idx + 1}/{len(asset_files)})"
|
|
)
|
|
|
|
if on_progress:
|
|
on_progress(95, "Finalizing save...")
|
|
|
|
os.makedirs(os.path.dirname(os.path.abspath(final_zip_path)), exist_ok=True)
|
|
if os.path.exists(final_zip_path):
|
|
os.remove(final_zip_path)
|
|
shutil.move(temp_zip_path, final_zip_path)
|
|
|
|
if on_progress:
|
|
on_progress(100, "Save complete!")
|
|
|
|
print(f"Project saved to {final_zip_path}")
|
|
|
|
if on_complete:
|
|
on_complete(True, None)
|
|
|
|
except Exception as e:
|
|
error_msg = f"Error saving project: {str(e)}"
|
|
print(error_msg)
|
|
if on_complete:
|
|
on_complete(False, error_msg)
|
|
|
|
finally:
|
|
if temp_dir and os.path.exists(temp_dir):
|
|
try:
|
|
shutil.rmtree(temp_dir)
|
|
except Exception:
|
|
pass
|
|
|
|
save_thread = threading.Thread(target=_background_save, daemon=True)
|
|
save_thread.start()
|
|
return save_thread
|
|
|
|
|
|
def load_from_zip(zip_path: str, extract_to: Optional[str] = None) -> Project:
|
|
"""
|
|
Load a project from a ZIP file.
|
|
|
|
Args:
|
|
zip_path: Path to the ZIP file to load
|
|
extract_to: Optional directory to extract to. If None, uses a temporary
|
|
directory that will be cleaned up when the project is closed.
|
|
|
|
Returns:
|
|
Project instance (raises exception on error)
|
|
"""
|
|
if not os.path.exists(zip_path):
|
|
raise FileNotFoundError(f"ZIP file not found: {zip_path}")
|
|
|
|
# Track if we created a temp directory
|
|
temp_dir_obj = None
|
|
|
|
# Determine extraction directory
|
|
if extract_to is None:
|
|
# Create a temporary directory using TemporaryDirectory
|
|
# This will be attached to the Project and auto-cleaned on deletion
|
|
zip_basename = os.path.splitext(os.path.basename(zip_path))[0]
|
|
temp_dir_obj = tempfile.TemporaryDirectory(prefix=f"pyPhotoAlbum_{zip_basename}_")
|
|
extract_to = temp_dir_obj.name
|
|
else:
|
|
# Create extraction directory if it doesn't exist
|
|
os.makedirs(extract_to, exist_ok=True)
|
|
|
|
# Extract ZIP contents
|
|
with zipfile.ZipFile(zip_path, "r") as zipf:
|
|
zipf.extractall(extract_to)
|
|
|
|
# Load project.json
|
|
project_json_path = os.path.join(extract_to, "project.json")
|
|
if not os.path.exists(project_json_path):
|
|
raise ValueError("Invalid project file: project.json not found")
|
|
|
|
with open(project_json_path, "r") as f:
|
|
project_data = json.load(f)
|
|
|
|
# Check version compatibility
|
|
# Try new version field first, fall back to legacy field
|
|
file_version = project_data.get("data_version", project_data.get("serialization_version", "1.0"))
|
|
|
|
# Check if version is compatible
|
|
is_compatible, error_msg = check_version_compatibility(file_version, zip_path)
|
|
if not is_compatible:
|
|
raise ValueError(error_msg)
|
|
|
|
# Apply migrations if needed
|
|
if VersionCompatibility.needs_migration(file_version):
|
|
print(f"Migrating project from version {file_version} to {CURRENT_DATA_VERSION}...")
|
|
project_data = DataMigration.migrate(project_data, file_version, CURRENT_DATA_VERSION)
|
|
print(f"Migration completed successfully")
|
|
elif file_version != CURRENT_DATA_VERSION:
|
|
print(f"Note: Loading project with version {file_version}, current version is {CURRENT_DATA_VERSION}")
|
|
|
|
# Create new project
|
|
project_name = project_data.get("name", "Untitled Project")
|
|
project = Project(name=project_name, folder_path=extract_to)
|
|
|
|
# Deserialize project data
|
|
project.deserialize(project_data)
|
|
|
|
# Update folder path to extraction location
|
|
project.folder_path = extract_to
|
|
project.asset_manager.project_folder = extract_to
|
|
project.asset_manager.assets_folder = os.path.join(extract_to, "assets")
|
|
|
|
# Attach temporary directory to project (if we created one)
|
|
# The TemporaryDirectory will auto-cleanup when the project is deleted
|
|
if temp_dir_obj is not None:
|
|
project._temp_dir = temp_dir_obj
|
|
print(f"Project loaded to temporary directory: {extract_to}")
|
|
|
|
# Normalize asset paths in all ImageData elements
|
|
# This fixes old projects that have absolute or wrong relative paths
|
|
_normalize_asset_paths(project, extract_to)
|
|
|
|
# Set asset resolution context for ImageData rendering
|
|
# Only set project folder - search paths are reserved for healing functionality
|
|
from pyPhotoAlbum.models import set_asset_resolution_context
|
|
|
|
set_asset_resolution_context(extract_to)
|
|
|
|
print(f"Project loaded from {zip_path} to {extract_to}")
|
|
return project
|
|
|
|
|
|
def get_project_info(zip_path: str) -> Optional[dict]:
|
|
"""
|
|
Get basic information about a project without fully loading it.
|
|
|
|
Args:
|
|
zip_path: Path to the ZIP file
|
|
|
|
Returns:
|
|
Dictionary with project info, or None if error
|
|
"""
|
|
try:
|
|
with zipfile.ZipFile(zip_path, "r") as zipf:
|
|
# Read project.json
|
|
project_json = zipf.read("project.json").decode("utf-8")
|
|
project_data = json.loads(project_json)
|
|
|
|
return {
|
|
"name": project_data.get("name", "Unknown"),
|
|
"version": project_data.get("serialization_version", "Unknown"),
|
|
"page_count": len(project_data.get("pages", [])),
|
|
"page_size_mm": project_data.get("page_size_mm", (0, 0)),
|
|
"working_dpi": project_data.get("working_dpi", 300),
|
|
}
|
|
except Exception as e:
|
|
print(f"Error reading project info: {e}")
|
|
return None
|