Initial commit of the Ad Generator Lite project, including backend services, frontend components, and configuration files. Added core functionalities for ad generation, user management, and image processing, along with a structured matrix system for ad testing.
f201243
| """ | |
| 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() | |