File size: 5,800 Bytes
f201243 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 |
"""
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()
|