322 lines
11 KiB
Python
322 lines
11 KiB
Python
"""
|
|
Version management and migration system for pyPhotoAlbum projects
|
|
"""
|
|
|
|
import os
|
|
import uuid
|
|
from datetime import datetime, timezone
|
|
from typing import Dict, Any, Optional, Callable, List
|
|
|
|
|
|
# Current data version - increment when making breaking changes to data format
|
|
CURRENT_DATA_VERSION = "3.0"
|
|
|
|
# Version history and compatibility information
|
|
VERSION_HISTORY = {
|
|
"1.0": {
|
|
"description": "Initial format with basic serialization",
|
|
"released": "2024-01-01",
|
|
"breaking_changes": [],
|
|
"compatible_with": ["1.0"],
|
|
},
|
|
"2.0": {
|
|
"description": "Fixed asset path handling - paths now stored relative to project folder",
|
|
"released": "2025-01-11",
|
|
"breaking_changes": [
|
|
"Asset paths changed from absolute/full-project-relative to project-relative",
|
|
"Added automatic path normalization for legacy projects"
|
|
],
|
|
"compatible_with": ["1.0", "2.0"], # 2.0 can read 1.0 with migration
|
|
},
|
|
"3.0": {
|
|
"description": "Added merge conflict resolution support with UUIDs, timestamps, and project IDs",
|
|
"released": "2025-01-22",
|
|
"breaking_changes": [
|
|
"Added required UUID fields to all pages and elements",
|
|
"Added created/last_modified timestamps to projects, pages, and elements",
|
|
"Added project_id for merge detection (same ID = merge, different ID = concatenate)",
|
|
"Added deletion tracking (deleted flag and deleted_at timestamp)",
|
|
],
|
|
"compatible_with": ["1.0", "2.0", "3.0"], # 3.0 can read older versions with migration
|
|
}
|
|
}
|
|
|
|
|
|
class VersionCompatibility:
|
|
"""Handles version compatibility checks and migrations"""
|
|
|
|
@staticmethod
|
|
def is_compatible(file_version: str) -> bool:
|
|
"""
|
|
Check if a file version is compatible with the current version.
|
|
|
|
Args:
|
|
file_version: Version string from the file
|
|
|
|
Returns:
|
|
True if compatible, False otherwise
|
|
"""
|
|
current_info = VERSION_HISTORY.get(CURRENT_DATA_VERSION, {})
|
|
compatible_versions = current_info.get("compatible_with", [])
|
|
return file_version in compatible_versions
|
|
|
|
@staticmethod
|
|
def needs_migration(file_version: str) -> bool:
|
|
"""
|
|
Check if a file needs migration to work with current version.
|
|
|
|
Args:
|
|
file_version: Version string from the file
|
|
|
|
Returns:
|
|
True if migration is needed, False otherwise
|
|
"""
|
|
# If versions don't match but are compatible, migration may be needed
|
|
return file_version != CURRENT_DATA_VERSION and VersionCompatibility.is_compatible(file_version)
|
|
|
|
@staticmethod
|
|
def get_version_info(version: str) -> Optional[Dict[str, Any]]:
|
|
"""Get information about a specific version."""
|
|
return VERSION_HISTORY.get(version)
|
|
|
|
@staticmethod
|
|
def get_migration_path(from_version: str, to_version: str) -> Optional[List[str]]:
|
|
"""
|
|
Get the migration path from one version to another.
|
|
|
|
Args:
|
|
from_version: Starting version
|
|
to_version: Target version
|
|
|
|
Returns:
|
|
List of version steps needed, or None if no path exists
|
|
"""
|
|
# For now, we only support direct migration paths
|
|
# In the future, we could implement multi-step migrations
|
|
|
|
if from_version == to_version:
|
|
return []
|
|
|
|
from_info = VERSION_HISTORY.get(from_version)
|
|
to_info = VERSION_HISTORY.get(to_version)
|
|
|
|
if not from_info or not to_info:
|
|
return None
|
|
|
|
# Check if direct migration is possible
|
|
compatible_versions = to_info.get("compatible_with", [])
|
|
if from_version in compatible_versions:
|
|
return [from_version, to_version]
|
|
|
|
return None
|
|
|
|
|
|
class DataMigration:
|
|
"""Handles data migrations between versions"""
|
|
|
|
# Registry of migration functions
|
|
_migrations: Dict[tuple, Callable] = {}
|
|
|
|
@classmethod
|
|
def register_migration(cls, from_version: str, to_version: str):
|
|
"""Decorator to register a migration function"""
|
|
def decorator(func):
|
|
cls._migrations[(from_version, to_version)] = func
|
|
return func
|
|
return decorator
|
|
|
|
@classmethod
|
|
def migrate(cls, data: Dict[str, Any], from_version: str, to_version: str) -> Dict[str, Any]:
|
|
"""
|
|
Migrate data from one version to another.
|
|
|
|
Args:
|
|
data: Project data dictionary
|
|
from_version: Current version of the data
|
|
to_version: Target version
|
|
|
|
Returns:
|
|
Migrated data dictionary
|
|
"""
|
|
if from_version == to_version:
|
|
return data
|
|
|
|
# Get migration path
|
|
migration_path = VersionCompatibility.get_migration_path(from_version, to_version)
|
|
if not migration_path:
|
|
raise ValueError(f"No migration path from {from_version} to {to_version}")
|
|
|
|
# Apply migrations in sequence
|
|
current_data = data
|
|
for i in range(len(migration_path) - 1):
|
|
step_from = migration_path[i]
|
|
step_to = migration_path[i + 1]
|
|
migration_key = (step_from, step_to)
|
|
|
|
if migration_key in cls._migrations:
|
|
print(f"Applying migration: {step_from} → {step_to}")
|
|
current_data = cls._migrations[migration_key](current_data)
|
|
else:
|
|
print(f"Warning: No explicit migration for {step_from} → {step_to}, using as-is")
|
|
|
|
return current_data
|
|
|
|
|
|
# Register migrations
|
|
|
|
@DataMigration.register_migration("1.0", "2.0")
|
|
def migrate_1_0_to_2_0(data: Dict[str, Any]) -> Dict[str, Any]:
|
|
"""
|
|
Migrate from version 1.0 to 2.0.
|
|
|
|
Main changes:
|
|
- Asset paths are normalized to be relative to project folder
|
|
- This is now handled automatically in load_from_zip via _normalize_asset_paths
|
|
"""
|
|
print("Migration 1.0 → 2.0: Asset paths will be normalized during load")
|
|
|
|
# Update version in data
|
|
data['data_version'] = "2.0"
|
|
|
|
# Note: Actual path normalization is handled in load_from_zip
|
|
# This migration mainly updates the version number
|
|
|
|
return data
|
|
|
|
|
|
@DataMigration.register_migration("2.0", "3.0")
|
|
def migrate_2_0_to_3_0(data: Dict[str, Any]) -> Dict[str, Any]:
|
|
"""
|
|
Migrate from version 2.0 to 3.0.
|
|
|
|
Main changes:
|
|
- Add UUIDs to all pages and elements
|
|
- Add timestamps (created, last_modified) to project, pages, and elements
|
|
- Add project_id to project
|
|
- Add deletion tracking (deleted, deleted_at) to pages and elements
|
|
"""
|
|
print("Migration 2.0 → 3.0: Adding UUIDs, timestamps, and project_id")
|
|
|
|
# Get current timestamp for migration
|
|
now = datetime.now(timezone.utc).isoformat()
|
|
|
|
# Add project-level fields
|
|
if "project_id" not in data:
|
|
data["project_id"] = str(uuid.uuid4())
|
|
print(f" Generated project_id: {data['project_id']}")
|
|
|
|
if "created" not in data:
|
|
data["created"] = now
|
|
|
|
if "last_modified" not in data:
|
|
data["last_modified"] = now
|
|
|
|
# Migrate pages
|
|
for page_data in data.get("pages", []):
|
|
# Add UUID
|
|
if "uuid" not in page_data:
|
|
page_data["uuid"] = str(uuid.uuid4())
|
|
|
|
# Add timestamps
|
|
if "created" not in page_data:
|
|
page_data["created"] = now
|
|
if "last_modified" not in page_data:
|
|
page_data["last_modified"] = now
|
|
|
|
# Add deletion tracking
|
|
if "deleted" not in page_data:
|
|
page_data["deleted"] = False
|
|
if "deleted_at" not in page_data:
|
|
page_data["deleted_at"] = None
|
|
|
|
# Migrate elements in page layout
|
|
layout_data = page_data.get("layout", {})
|
|
for element_data in layout_data.get("elements", []):
|
|
# Add UUID
|
|
if "uuid" not in element_data:
|
|
element_data["uuid"] = str(uuid.uuid4())
|
|
|
|
# Add timestamps
|
|
if "created" not in element_data:
|
|
element_data["created"] = now
|
|
if "last_modified" not in element_data:
|
|
element_data["last_modified"] = now
|
|
|
|
# Add deletion tracking
|
|
if "deleted" not in element_data:
|
|
element_data["deleted"] = False
|
|
if "deleted_at" not in element_data:
|
|
element_data["deleted_at"] = None
|
|
|
|
# Update version
|
|
data['data_version'] = "3.0"
|
|
|
|
print(f" Migrated {len(data.get('pages', []))} pages to v3.0")
|
|
|
|
return data
|
|
|
|
|
|
def check_version_compatibility(file_version: str, file_path: str = "") -> tuple[bool, Optional[str]]:
|
|
"""
|
|
Check version compatibility and provide user-friendly messages.
|
|
|
|
Args:
|
|
file_version: Version from the file
|
|
file_path: Optional path to the file (for error messages)
|
|
|
|
Returns:
|
|
Tuple of (is_compatible, error_message)
|
|
"""
|
|
if not file_version:
|
|
return True, None # No version specified, assume compatible
|
|
|
|
if VersionCompatibility.is_compatible(file_version):
|
|
if VersionCompatibility.needs_migration(file_version):
|
|
print(f"File version {file_version} is compatible but needs migration to {CURRENT_DATA_VERSION}")
|
|
return True, None
|
|
|
|
# Not compatible
|
|
file_info = VersionCompatibility.get_version_info(file_version)
|
|
current_info = VersionCompatibility.get_version_info(CURRENT_DATA_VERSION)
|
|
|
|
error_msg = f"Incompatible file version: {file_version}\n\n"
|
|
error_msg += f"This file was created with version {file_version}, "
|
|
error_msg += f"but this application uses version {CURRENT_DATA_VERSION}.\n\n"
|
|
|
|
if file_info:
|
|
error_msg += f"File version info:\n"
|
|
error_msg += f" Description: {file_info.get('description', 'Unknown')}\n"
|
|
error_msg += f" Released: {file_info.get('released', 'Unknown')}\n"
|
|
breaking_changes = file_info.get('breaking_changes', [])
|
|
if breaking_changes:
|
|
error_msg += f" Breaking changes:\n"
|
|
for change in breaking_changes:
|
|
error_msg += f" - {change}\n"
|
|
|
|
error_msg += f"\nPlease use a compatible version of pyPhotoAlbum to open this file."
|
|
|
|
return False, error_msg
|
|
|
|
|
|
def format_version_info() -> str:
|
|
"""Format version information for display"""
|
|
info = [
|
|
f"pyPhotoAlbum Data Format Version: {CURRENT_DATA_VERSION}",
|
|
"",
|
|
"Version History:",
|
|
]
|
|
|
|
for version in sorted(VERSION_HISTORY.keys(), reverse=True):
|
|
version_info = VERSION_HISTORY[version]
|
|
info.append(f"\n Version {version}")
|
|
info.append(f" Description: {version_info.get('description', 'Unknown')}")
|
|
info.append(f" Released: {version_info.get('released', 'Unknown')}")
|
|
|
|
breaking_changes = version_info.get('breaking_changes', [])
|
|
if breaking_changes:
|
|
info.append(f" Breaking changes:")
|
|
for change in breaking_changes:
|
|
info.append(f" - {change}")
|
|
|
|
return "\n".join(info)
|