Added command to remove orphaned images and thumbnails.

boxes/management/commands/clean_orphaned_images.py

@@ -0,0 +1,147 @@
+import json
+import os
+from django.conf import settings
+from django.core.management.base import BaseCommand
+from django.db.models import F
+from sorl.thumbnail.models import KVStore
+from boxes.models import Thing
+
+
+class Command(BaseCommand):
+    help = 'Clean up orphaned images and thumbnails from deleted things'
+
+    def add_arguments(self, parser):
+        parser.add_argument(
+            '--dry-run',
+            action='store_true',
+            dest='dry_run',
+            help='Show what would be deleted without actually deleting',
+        )
+
+    def handle(self, *args, **options):
+        dry_run = options.get('dry_run', False)
+        
+        if dry_run:
+            self.stdout.write(self.style.WARNING('DRY RUN - No files will be deleted'))
+        
+        self.stdout.write('Finding orphaned images and thumbnails...')
+        
+        media_root = settings.MEDIA_ROOT
+        cache_root = os.path.join(media_root, 'cache')
+        things_root = os.path.join(media_root, 'things')
+        
+        if not os.path.exists(things_root):
+            self.stdout.write(self.style.WARNING('No things directory found'))
+            return
+        
+        valid_paths = set()
+        for thing in Thing.objects.exclude(picture__exact='').exclude(picture__isnull=True):
+            if thing.picture:
+                valid_paths.add(os.path.basename(thing.picture.name))
+        
+        self.stdout.write(f'Found {len(valid_paths)} valid images in database')
+        
+        orphaned_thumbnail_paths = set()
+        db_cache_paths = set()
+        
+        for kvstore in KVStore.objects.filter(key__startswith='sorl-thumbnail||image||'):
+            try:
+                data = json.loads(kvstore.value)
+                name = data.get('name', '')
+                if name.startswith('things/'):
+                    filename = os.path.basename(name)
+                    if filename not in valid_paths:
+                        image_hash = kvstore.key.split('||')[-1]
+                        thumbnail_kvstore = KVStore.objects.filter(key=f'sorl-thumbnail||thumbnails||{image_hash}').first()
+                        if thumbnail_kvstore:
+                            thumbnail_list = json.loads(thumbnail_kvstore.value)
+                            for thumbnail_hash in thumbnail_list:
+                                thumbnail_image_kvstore = KVStore.objects.filter(key=f'sorl-thumbnail||image||{thumbnail_hash}').first()
+                                if thumbnail_image_kvstore:
+                                    thumbnail_data = json.loads(thumbnail_image_kvstore.value)
+                                    thumbnail_path = thumbnail_data.get('name', '')
+                                    if thumbnail_path.startswith('cache/'):
+                                        orphaned_thumbnail_paths.add(thumbnail_path)
+                elif name.startswith('cache/'):
+                    db_cache_paths.add(name)
+            except (json.JSONDecodeError, KeyError, AttributeError):
+                pass
+        
+        deleted_count = 0
+        thumbnail_deleted_count = 0
+        empty_dirs_removed = 0
+        
+        for root, dirs, files in os.walk(things_root, topdown=False):
+            for filename in files:
+                file_path = os.path.join(root, filename)
+                relative_path = os.path.relpath(file_path, things_root)
+                
+                if relative_path not in valid_paths:
+                    deleted_count += 1
+                    if dry_run:
+                        self.stdout.write(f'Would delete: {file_path}')
+                    else:
+                        try:
+                            os.remove(file_path)
+                            self.stdout.write(f'Deleted: {file_path}')
+                        except OSError as e:
+                            self.stdout.write(self.style.ERROR(f'Failed to delete {file_path}: {e}'))
+            
+            for dirname in dirs:
+                dir_path = os.path.join(root, dirname)
+                if not os.listdir(dir_path):
+                    if dry_run:
+                        self.stdout.write(f'Would remove empty directory: {dir_path}')
+                    else:
+                        try:
+                            os.rmdir(dir_path)
+                            self.stdout.write(f'Removed empty directory: {dir_path}')
+                            empty_dirs_removed += 1
+                        except OSError as e:
+                            self.stdout.write(self.style.ERROR(f'Failed to remove {dir_path}: {e}'))
+        
+        if os.path.exists(cache_root):
+            for root, dirs, files in os.walk(cache_root, topdown=False):
+                for filename in files:
+                    file_path = os.path.join(root, filename)
+                    relative_path = os.path.relpath(file_path, media_root)
+                    
+                    if relative_path in orphaned_thumbnail_paths:
+                        thumbnail_deleted_count += 1
+                        if dry_run:
+                            self.stdout.write(f'Would delete thumbnail (orphaned image): {file_path}')
+                        else:
+                            try:
+                                os.remove(file_path)
+                                self.stdout.write(f'Deleted thumbnail (orphaned image): {file_path}')
+                            except OSError as e:
+                                self.stdout.write(self.style.ERROR(f'Failed to delete {file_path}: {e}'))
+                    elif relative_path not in db_cache_paths:
+                        thumbnail_deleted_count += 1
+                        if dry_run:
+                            self.stdout.write(f'Would delete thumbnail (no db entry): {file_path}')
+                        else:
+                            try:
+                                os.remove(file_path)
+                                self.stdout.write(f'Deleted thumbnail (no db entry): {file_path}')
+                            except OSError as e:
+                                self.stdout.write(self.style.ERROR(f'Failed to delete {file_path}: {e}'))
+                
+                for dirname in dirs:
+                    dir_path = os.path.join(root, dirname)
+                    if not os.listdir(dir_path):
+                        if dry_run:
+                            self.stdout.write(f'Would remove empty cache directory: {dir_path}')
+                        else:
+                            try:
+                                os.rmdir(dir_path)
+                                empty_dirs_removed += 1
+                            except OSError as e:
+                                self.stdout.write(self.style.ERROR(f'Failed to remove {dir_path}: {e}'))
+        
+        if dry_run:
+            self.stdout.write(self.style.WARNING(f'\nDry run complete. Would delete {deleted_count} images and {thumbnail_deleted_count} thumbnails'))
+            self.stdout.write(f'Would remove {empty_dirs_removed} empty directories')
+        else:
+            self.stdout.write(self.style.SUCCESS(f'\nCleanup complete! Deleted {deleted_count} images and {thumbnail_deleted_count} thumbnails'))
+            self.stdout.write(f'Removed {empty_dirs_removed} empty directories')