""" Image cleanup service for removing old temporary images. In production, images are saved temporarily and cleaned up after a retention period. """ import os import sys import time from pathlib import Path from typing import List, Optional from datetime import datetime, timedelta # Add parent directory to path for imports sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) from config import settings class ImageCleanupService: """Service for cleaning up old temporary images.""" def __init__(self): """Initialize cleanup service.""" self.output_dir = Path(settings.output_dir) self.retention_hours = settings.local_image_retention_hours def get_image_age_hours(self, filepath: Path) -> Optional[float]: """ Get the age of an image file in hours. Args: filepath: Path to the image file Returns: Age in hours, or None if file doesn't exist or can't determine age """ try: if not filepath.exists(): return None # Get file modification time mtime = filepath.stat().st_mtime age_seconds = time.time() - mtime age_hours = age_seconds / 3600 return age_hours except Exception as e: print(f"Error getting age for {filepath}: {e}") return None def cleanup_old_images(self, dry_run: bool = False) -> dict: """ Clean up images older than the retention period. Args: dry_run: If True, only report what would be deleted without actually deleting Returns: Dictionary with cleanup statistics """ if not self.output_dir.exists(): return { "deleted": 0, "failed": 0, "skipped": 0, "total_size_mb": 0.0, "errors": [] } deleted = 0 failed = 0 skipped = 0 total_size_mb = 0.0 errors = [] # Find all image files in the output directory image_extensions = {'.png', '.jpg', '.jpeg', '.gif', '.webp'} for filepath in self.output_dir.rglob('*'): if not filepath.is_file(): continue # Check if it's an image file if filepath.suffix.lower() not in image_extensions: continue # Get file age age_hours = self.get_image_age_hours(filepath) if age_hours is None: skipped += 1 continue # Check if file is older than retention period if age_hours > self.retention_hours: try: # Get file size before deletion file_size_mb = filepath.stat().st_size / (1024 * 1024) total_size_mb += file_size_mb if not dry_run: filepath.unlink() print(f"Deleted old image: {filepath} (age: {age_hours:.2f} hours)") deleted += 1 except Exception as e: failed += 1 error_msg = f"Failed to delete {filepath}: {e}" errors.append(error_msg) print(error_msg) result = { "deleted": deleted, "failed": failed, "skipped": skipped, "total_size_mb": round(total_size_mb, 2), "errors": errors } if dry_run: print(f"[DRY RUN] Would delete {deleted} images ({total_size_mb:.2f} MB)") else: print(f"Cleanup complete: Deleted {deleted} images, freed {total_size_mb:.2f} MB") return result def get_storage_stats(self) -> dict: """ Get statistics about image storage. Returns: Dictionary with storage statistics """ if not self.output_dir.exists(): return { "total_files": 0, "total_size_mb": 0.0, "oldest_file_hours": None, "newest_file_hours": None, } image_extensions = {'.png', '.jpg', '.jpeg', '.gif', '.webp'} total_files = 0 total_size = 0 oldest_hours = None newest_hours = None for filepath in self.output_dir.rglob('*'): if not filepath.is_file(): continue if filepath.suffix.lower() not in image_extensions: continue total_files += 1 try: file_size = filepath.stat().st_size total_size += file_size age_hours = self.get_image_age_hours(filepath) if age_hours is not None: if oldest_hours is None or age_hours > oldest_hours: oldest_hours = age_hours if newest_hours is None or age_hours < newest_hours: newest_hours = age_hours except Exception: pass return { "total_files": total_files, "total_size_mb": round(total_size / (1024 * 1024), 2), "oldest_file_hours": round(oldest_hours, 2) if oldest_hours is not None else None, "newest_file_hours": round(newest_hours, 2) if newest_hours is not None else None, } # Global instance cleanup_service = ImageCleanupService()