246 lines
8.3 KiB
Python
246 lines
8.3 KiB
Python
"""
|
|
Autosave and checkpoint management for pyPhotoAlbum.
|
|
|
|
This module provides automatic checkpoint creation and recovery functionality
|
|
to prevent data loss from crashes or unexpected exits.
|
|
"""
|
|
|
|
import os
|
|
import json
|
|
import shutil
|
|
from pathlib import Path
|
|
from datetime import datetime, timedelta
|
|
from typing import Dict, Optional, List, Tuple
|
|
from pyPhotoAlbum.project_serializer import save_to_zip, load_from_zip
|
|
|
|
|
|
class AutosaveManager:
|
|
"""Manages autosave checkpoints for projects."""
|
|
|
|
CHECKPOINT_DIR = Path.home() / ".pyphotoalbum" / "checkpoints"
|
|
CHECKPOINT_PREFIX = "checkpoint_"
|
|
CHECKPOINT_EXTENSION = ".ppz"
|
|
|
|
def __init__(self):
|
|
"""Initialize the autosave manager."""
|
|
self._ensure_checkpoint_directory()
|
|
|
|
def _ensure_checkpoint_directory(self):
|
|
"""Ensure the checkpoint directory exists."""
|
|
self.CHECKPOINT_DIR.mkdir(parents=True, exist_ok=True)
|
|
|
|
def _get_checkpoint_path(self, project_name: str, timestamp: Optional[datetime] = None) -> Path:
|
|
"""
|
|
Get the path for a checkpoint file.
|
|
|
|
Args:
|
|
project_name: Name of the project
|
|
timestamp: Optional timestamp, defaults to current time
|
|
|
|
Returns:
|
|
Path to the checkpoint file
|
|
"""
|
|
if timestamp is None:
|
|
timestamp = datetime.now()
|
|
|
|
# Sanitize project name for filename
|
|
safe_name = "".join(c if c.isalnum() or c in "-_" else "_" for c in project_name)
|
|
timestamp_str = timestamp.strftime("%Y%m%d_%H%M%S")
|
|
filename = f"{self.CHECKPOINT_PREFIX}{safe_name}_{timestamp_str}{self.CHECKPOINT_EXTENSION}"
|
|
|
|
return self.CHECKPOINT_DIR / filename
|
|
|
|
def create_checkpoint(self, project) -> Tuple[bool, str]:
|
|
"""
|
|
Create a checkpoint for the given project.
|
|
|
|
Args:
|
|
project: Project instance to checkpoint
|
|
|
|
Returns:
|
|
Tuple of (success: bool, message: str)
|
|
"""
|
|
try:
|
|
checkpoint_path = self._get_checkpoint_path(project.name)
|
|
success, message = save_to_zip(project, str(checkpoint_path))
|
|
|
|
if success:
|
|
# Also save metadata about this checkpoint
|
|
self._save_checkpoint_metadata(project, checkpoint_path)
|
|
return True, f"Checkpoint created: {checkpoint_path.name}"
|
|
else:
|
|
return False, f"Checkpoint failed: {message}"
|
|
|
|
except Exception as e:
|
|
return False, f"Checkpoint error: {str(e)}"
|
|
|
|
def _save_checkpoint_metadata(self, project, checkpoint_path: Path):
|
|
"""
|
|
Save metadata about a checkpoint.
|
|
|
|
Args:
|
|
project: Project instance
|
|
checkpoint_path: Path to the checkpoint file
|
|
"""
|
|
metadata = {
|
|
"project_name": project.name,
|
|
"timestamp": datetime.now().isoformat(),
|
|
"checkpoint_path": str(checkpoint_path),
|
|
"original_path": getattr(project, "file_path", None),
|
|
}
|
|
|
|
metadata_path = checkpoint_path.with_suffix(".json")
|
|
with open(metadata_path, "w") as f:
|
|
json.dump(metadata, f, indent=2)
|
|
|
|
def list_checkpoints(self, project_name: Optional[str] = None) -> List[Tuple[Path, dict]]:
|
|
"""
|
|
List available checkpoints.
|
|
|
|
Args:
|
|
project_name: Optional filter by project name
|
|
|
|
Returns:
|
|
List of tuples (checkpoint_path, metadata)
|
|
"""
|
|
checkpoints = []
|
|
|
|
for checkpoint_file in self.CHECKPOINT_DIR.glob(f"{self.CHECKPOINT_PREFIX}*{self.CHECKPOINT_EXTENSION}"):
|
|
metadata_file = checkpoint_file.with_suffix(".json")
|
|
|
|
# Try to load metadata
|
|
metadata = {}
|
|
if metadata_file.exists():
|
|
try:
|
|
with open(metadata_file, "r") as f:
|
|
metadata = json.load(f)
|
|
except:
|
|
pass
|
|
|
|
# Filter by project name if specified
|
|
if project_name is None or metadata.get("project_name") == project_name:
|
|
checkpoints.append((checkpoint_file, metadata))
|
|
|
|
# Sort by timestamp (newest first)
|
|
checkpoints.sort(key=lambda x: x[1].get("timestamp", ""), reverse=True)
|
|
return checkpoints
|
|
|
|
def load_checkpoint(self, checkpoint_path: Path):
|
|
"""
|
|
Load a project from a checkpoint.
|
|
|
|
Args:
|
|
checkpoint_path: Path to the checkpoint file
|
|
|
|
Returns:
|
|
Tuple of (success: bool, project or error_message)
|
|
"""
|
|
try:
|
|
project = load_from_zip(str(checkpoint_path))
|
|
return True, project
|
|
except Exception as e:
|
|
return False, f"Failed to load checkpoint: {str(e)}"
|
|
|
|
def delete_checkpoint(self, checkpoint_path: Path) -> bool:
|
|
"""
|
|
Delete a checkpoint file and its metadata.
|
|
|
|
Args:
|
|
checkpoint_path: Path to the checkpoint file
|
|
|
|
Returns:
|
|
True if successful
|
|
"""
|
|
try:
|
|
# Delete checkpoint file
|
|
if checkpoint_path.exists():
|
|
checkpoint_path.unlink()
|
|
|
|
# Delete metadata file
|
|
metadata_path = checkpoint_path.with_suffix(".json")
|
|
if metadata_path.exists():
|
|
metadata_path.unlink()
|
|
|
|
return True
|
|
except Exception as e:
|
|
print(f"Error deleting checkpoint: {e}")
|
|
return False
|
|
|
|
def delete_all_checkpoints(self, project_name: Optional[str] = None):
|
|
"""
|
|
Delete all checkpoints, optionally filtered by project name.
|
|
|
|
Args:
|
|
project_name: Optional filter by project name
|
|
"""
|
|
checkpoints = self.list_checkpoints(project_name)
|
|
for checkpoint_path, _ in checkpoints:
|
|
self.delete_checkpoint(checkpoint_path)
|
|
|
|
def cleanup_old_checkpoints(self, max_age_hours: int = 24 * 7, max_count: int = 50):
|
|
"""
|
|
Clean up old checkpoints to prevent unlimited growth.
|
|
|
|
Args:
|
|
max_age_hours: Maximum age in hours (default: 7 days)
|
|
max_count: Maximum number of checkpoints to keep per project
|
|
"""
|
|
now = datetime.now()
|
|
checkpoints_by_project: Dict[str, List[Tuple[Path, dict]]] = {}
|
|
|
|
# Group checkpoints by project
|
|
for checkpoint_path, metadata in self.list_checkpoints():
|
|
project_name = metadata.get("project_name", "unknown")
|
|
if project_name not in checkpoints_by_project:
|
|
checkpoints_by_project[project_name] = []
|
|
checkpoints_by_project[project_name].append((checkpoint_path, metadata))
|
|
|
|
# Clean up each project's checkpoints
|
|
for project_name, checkpoints in checkpoints_by_project.items():
|
|
# Sort by timestamp (newest first)
|
|
checkpoints.sort(key=lambda x: x[1].get("timestamp", ""), reverse=True)
|
|
|
|
for idx, (checkpoint_path, metadata) in enumerate(checkpoints):
|
|
# Delete if too old
|
|
timestamp_str = metadata.get("timestamp")
|
|
if timestamp_str:
|
|
try:
|
|
timestamp = datetime.fromisoformat(timestamp_str)
|
|
age = now - timestamp
|
|
if age > timedelta(hours=max_age_hours):
|
|
self.delete_checkpoint(checkpoint_path)
|
|
continue
|
|
except:
|
|
pass
|
|
|
|
# Delete if beyond max count
|
|
if idx >= max_count:
|
|
self.delete_checkpoint(checkpoint_path)
|
|
|
|
def has_checkpoints(self, project_name: Optional[str] = None) -> bool:
|
|
"""
|
|
Check if there are any checkpoints available.
|
|
|
|
Args:
|
|
project_name: Optional filter by project name
|
|
|
|
Returns:
|
|
True if checkpoints exist
|
|
"""
|
|
return len(self.list_checkpoints(project_name)) > 0
|
|
|
|
def get_latest_checkpoint(self, project_name: Optional[str] = None) -> Optional[Tuple[Path, dict]]:
|
|
"""
|
|
Get the most recent checkpoint.
|
|
|
|
Args:
|
|
project_name: Optional filter by project name
|
|
|
|
Returns:
|
|
Tuple of (checkpoint_path, metadata) or None
|
|
"""
|
|
checkpoints = self.list_checkpoints(project_name)
|
|
if checkpoints:
|
|
return checkpoints[0] # Already sorted newest first
|
|
return None
|