dtourolle/pyPhotoAlbum
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Added autosave

Browse Source 
Fixed unusual bug where scrolling while draging content results in desyned movements
This commit is contained in:
Duncan Tourolle 2025-11-22 22:29:05 +01:00
parent 3bfe2fa654
commit e93a789524
6 changed files with 459 additions and 32 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

247
pyPhotoAlbum/autosave_manager.py Normal file
View File

@ -0,0 +1,247 @@
"""
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
from typing import 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:
success, result = load_from_zip(str(checkpoint_path))
return success, result
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
"""
from datetime import timedelta
now = datetime.now()
checkpoints_by_project = {}
# 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

33
pyPhotoAlbum/commands.py
View File

@ -705,24 +705,29 @@ class StateChangeCommand(Command):
class CommandHistory:
"""Manages undo/redo command history"""
def __init__(self, max_history: int = 100, asset_manager=None):
def __init__(self, max_history: int = 100, asset_manager=None, project=None):
self.undo_stack: List[Command] = []
self.redo_stack: List[Command] = []
self.max_history = max_history
self.asset_manager = asset_manager
self.project = project # Reference to project for dirty flag tracking
def execute(self, command: Command):
"""Execute a command and add it to history"""
command.execute()
# When clearing redo stack, release asset references
for cmd in self.redo_stack:
self._release_command_assets(cmd)
self.redo_stack.clear()
self.undo_stack.append(command)
# Mark project as dirty
if self.project:
self.project.mark_dirty()
# Limit history size - release assets from old commands
if len(self.undo_stack) > self.max_history:
old_cmd = self.undo_stack.pop(0)
@ -747,20 +752,30 @@ class CommandHistory:
"""Undo the last command"""
if not self.can_undo():
return False
command = self.undo_stack.pop()
command.undo()
self.redo_stack.append(command)
# Mark project as dirty
if self.project:
self.project.mark_dirty()
return True
def redo(self) -> bool:
"""Redo the last undone command"""
if not self.can_redo():
return False
command = self.redo_stack.pop()
command.redo()
self.undo_stack.append(command)
# Mark project as dirty
if self.project:
self.project.mark_dirty()
return True
def can_undo(self) -> bool:

127
pyPhotoAlbum/main.py
View File

@ -9,9 +9,9 @@ import sys
from pathlib import Path
from PyQt6.QtWidgets import (
QApplication, QMainWindow, QVBoxLayout, QWidget,
QStatusBar, QScrollBar, QHBoxLayout
QStatusBar, QScrollBar, QHBoxLayout, QMessageBox
)
from PyQt6.QtCore import Qt, QSize
from PyQt6.QtCore import Qt, QSize, QTimer
from PyQt6.QtGui import QIcon
from pyPhotoAlbum.project import Project
@ -19,6 +19,7 @@ from pyPhotoAlbum.template_manager import TemplateManager
from pyPhotoAlbum.ribbon_widget import RibbonWidget
from pyPhotoAlbum.ribbon_builder import build_ribbon_config, print_ribbon_summary
from pyPhotoAlbum.gl_widget import GLWidget
from pyPhotoAlbum.autosave_manager import AutosaveManager
# Import mixins
from pyPhotoAlbum.mixins.base import ApplicationStateMixin
@ -60,13 +61,24 @@ class MainWindow(
def __init__(self):
super().__init__()
# Initialize autosave manager
self._autosave_manager = AutosaveManager()
# Initialize shared state first
self._init_state()
# Initialize UI
self._init_ui()
# Check for checkpoint recovery
self._check_checkpoint_recovery()
# Setup autosave timer (every 5 minutes)
self._autosave_timer = QTimer(self)
self._autosave_timer.timeout.connect(self._perform_autosave)
self._autosave_timer.start(5 * 60 * 1000) # 5 minutes in milliseconds
# Add a sample page for demonstration
#self._add_sample_page()
@ -264,7 +276,110 @@ class MainWindow(
page = Page(layout=page_layout, page_number=1)
page.manually_sized = False # Not manually sized, uses defaults
self.project.add_page(page)
def _perform_autosave(self):
"""Perform automatic checkpoint save"""
if self.project and self.project.is_dirty():
success, message = self._autosave_manager.create_checkpoint(self.project)
if success:
print(f"Autosave: {message}")
else:
print(f"Autosave failed: {message}")
def _check_checkpoint_recovery(self):
"""Check for available checkpoints on startup and offer recovery"""
if not self._autosave_manager.has_checkpoints():
return
# Get the latest checkpoint
checkpoint_info = self._autosave_manager.get_latest_checkpoint()
if not checkpoint_info:
return
checkpoint_path, metadata = checkpoint_info
project_name = metadata.get('project_name', 'Unknown')
timestamp_str = metadata.get('timestamp', 'Unknown time')
# Parse timestamp for better display
try:
from datetime import datetime
timestamp = datetime.fromisoformat(timestamp_str)
time_display = timestamp.strftime("%Y-%m-%d %H:%M:%S")
except:
time_display = timestamp_str
# Ask user if they want to recover
reply = QMessageBox.question(
self,
"Checkpoint Recovery",
f"A checkpoint was found:\n\n"
f"Project: {project_name}\n"
f"Time: {time_display}\n\n"
f"Would you like to recover this checkpoint?",
QMessageBox.StandardButton.Yes | QMessageBox.StandardButton.No,
QMessageBox.StandardButton.Yes
)
if reply == QMessageBox.StandardButton.Yes:
# Load the checkpoint
success, result = self._autosave_manager.load_checkpoint(checkpoint_path)
if success:
# Replace current project with recovered one
if hasattr(self, '_project') and self._project:
self._project.cleanup()
self._project = result
self.gl_widget.current_page_index = 0
self.update_view()
self.show_status(f"Recovered checkpoint: {project_name}")
print(f"Successfully recovered checkpoint: {project_name}")
else:
error_msg = f"Failed to recover checkpoint: {result}"
self.show_error("Recovery Failed", error_msg)
print(error_msg)
def closeEvent(self, event):
"""Handle window close event"""
# Check if project has unsaved changes
if self.project and self.project.is_dirty():
reply = QMessageBox.question(
self,
"Unsaved Changes",
"You have unsaved changes. Would you like to save before exiting?",
QMessageBox.StandardButton.Save | QMessageBox.StandardButton.Discard | QMessageBox.StandardButton.Cancel,
QMessageBox.StandardButton.Save
)
if reply == QMessageBox.StandardButton.Save:
# Trigger save
self.save_project()
# Check if save was successful (project should be clean now)
if self.project.is_dirty():
# User cancelled save dialog or save failed
event.ignore()
return
elif reply == QMessageBox.StandardButton.Cancel:
# User cancelled exit
event.ignore()
return
# If Discard, continue with exit
# Clean up checkpoints on successful exit
if self.project:
self._autosave_manager.delete_all_checkpoints(self.project.name)
self.project.cleanup()
# Stop autosave timer
if hasattr(self, '_autosave_timer'):
self._autosave_timer.stop()
# Cleanup old checkpoints
self._autosave_manager.cleanup_old_checkpoints()
event.accept()
def main():

21
pyPhotoAlbum/mixins/mouse_interaction.py
View File

@ -280,11 +280,23 @@ class MouseInteractionMixin:
new_zoom = self.zoom_level * zoom_factor
if 0.1 <= new_zoom <= 5.0:
old_pan_x = self.pan_offset[0]
old_pan_y = self.pan_offset[1]
self.zoom_level = new_zoom
self.pan_offset[0] = mouse_x - world_x * self.zoom_level
self.pan_offset[1] = mouse_y - world_y * self.zoom_level
# If dragging, adjust drag_start_pos to account for pan_offset change
if self.is_dragging and self.drag_start_pos:
pan_delta_x = self.pan_offset[0] - old_pan_x
pan_delta_y = self.pan_offset[1] - old_pan_y
self.drag_start_pos = (
self.drag_start_pos[0] + pan_delta_x,
self.drag_start_pos[1] + pan_delta_y
)
# Clamp pan offset to content bounds
if hasattr(self, 'clamp_pan_offset'):
self.clamp_pan_offset()
@ -301,12 +313,21 @@ class MouseInteractionMixin:
else:
# Regular wheel: Vertical scroll
scroll_amount = delta * 0.5
old_pan_y = self.pan_offset[1]
self.pan_offset[1] += scroll_amount
# Clamp pan offset to content bounds
if hasattr(self, 'clamp_pan_offset'):
self.clamp_pan_offset()
# If dragging, adjust drag_start_pos to account for pan_offset change
if self.is_dragging and self.drag_start_pos:
pan_delta_y = self.pan_offset[1] - old_pan_y
self.drag_start_pos = (
self.drag_start_pos[0],
self.drag_start_pos[1] + pan_delta_y
)
self.update()
# Update scrollbars if available

32
pyPhotoAlbum/mixins/operations/file_ops.py
View File

@ -200,6 +200,7 @@ class FileOperationsMixin:
# Create and configure async loader
self._project_loader = AsyncProjectLoader(file_path)
self._opening_file_path = file_path # Store for later
# Connect signals
self._project_loader.progress_updated.connect(self._on_load_progress)
@ -223,6 +224,13 @@ class FileOperationsMixin:
# Set new project
self.project = project
# Set file path and mark as clean
if hasattr(self, '_opening_file_path'):
self.project.file_path = self._opening_file_path
delattr(self, '_opening_file_path')
self.project.mark_clean()
self.gl_widget.current_page_index = 0 # Reset to first page
# Hide loading widget
@ -255,20 +263,26 @@ class FileOperationsMixin:
)
def save_project(self):
"""Save the current project"""
file_path, _ = QFileDialog.getSaveFileName(
self,
"Save Project",
"",
"pyPhotoAlbum Projects (*.ppz);;All Files (*)"
)
# If project has a file path, use it; otherwise prompt for location
file_path = self.project.file_path if hasattr(self.project, 'file_path') and self.project.file_path else None
if not file_path:
file_path, _ = QFileDialog.getSaveFileName(
self,
"Save Project",
"",
"pyPhotoAlbum Projects (*.ppz);;All Files (*)"
)
if file_path:
print(f"Saving project to: {file_path}")
# Save project to ZIP
success, error = save_to_zip(self.project, file_path)
if success:
self.project.file_path = file_path
self.project.mark_clean()
self.show_status(f"Project saved: {file_path}")
print(f"Successfully saved project to: {file_path}")
else:

31
pyPhotoAlbum/project.py
View File

@ -135,9 +135,24 @@ class Project:
# Initialize asset manager
self.asset_manager = AssetManager(self.folder_path)
# Initialize command history with asset manager
self.history = CommandHistory(max_history=100)
self.history.asset_manager = self.asset_manager
# Initialize command history with asset manager and project reference
self.history = CommandHistory(max_history=100, asset_manager=self.asset_manager, project=self)
# Track unsaved changes
self._dirty = False
self.file_path = None # Path to the saved .ppz file
def mark_dirty(self):
"""Mark the project as having unsaved changes."""
self._dirty = True
def mark_clean(self):
"""Mark the project as saved (no unsaved changes)."""
self._dirty = False
def is_dirty(self) -> bool:
"""Check if the project has unsaved changes."""
return self._dirty
def add_page(self, page: Page):
"""Add a page to the project"""
@ -145,6 +160,7 @@ class Project:
# Update cover dimensions if we have a cover
if self.has_cover and self.pages:
self.update_cover_dimensions()
self.mark_dirty()
def remove_page(self, page: Page):
"""Remove a page from the project"""
@ -152,6 +168,7 @@ class Project:
# Update cover dimensions if we have a cover
if self.has_cover and self.pages:
self.update_cover_dimensions()
self.mark_dirty()
def calculate_spine_width(self) -> float:
"""
@ -380,15 +397,13 @@ class Project:
page.deserialize(page_data)
self.pages.append(page)
# Deserialize command history with asset manager
# Deserialize command history with asset manager and project reference
history_data = data.get("history")
if history_data:
self.history = CommandHistory(max_history=100)
self.history.asset_manager = self.asset_manager
self.history = CommandHistory(max_history=100, asset_manager=self.asset_manager, project=self)
self.history.deserialize(history_data, self)
else:
self.history = CommandHistory(max_history=100)
self.history.asset_manager = self.asset_manager
self.history = CommandHistory(max_history=100, asset_manager=self.asset_manager, project=self)
def cleanup(self):
"""