"""Library tooling: duplicates, deletion and storage overview.""" import logging import shutil from pathlib import Path from django.conf import settings from django.core.cache import cache from django.db.models import Count, Q from rest_framework import status from rest_framework.permissions import AllowAny, IsAuthenticated from rest_framework.response import Response from rest_framework.views import APIView from .models import TAG_CLOUD_CACHE_KEYS, MediaItem, MediaLocation from .permissions import CanUpload from .services import signed_media_url logger = logging.getLogger(__name__) HASH_FIELDS = ("ahash", "dhash", "phash", "whash") STORAGE_CACHE_KEY = "j621.library.storage" STORAGE_CACHE_TTL = 60 GROUPS_PER_PAGE = 20 def parse_threshold(value, default=0.8): try: parsed = float(value) except (TypeError, ValueError): return default if parsed > 1: parsed = parsed / 100.0 return min(max(parsed, 0.0), 1.0) def parse_algorithms(value): if not isinstance(value, list): return list(HASH_FIELDS) selected = [field for field in value if field in HASH_FIELDS] return selected or list(HASH_FIELDS) def hashes_similarity(first_hashes, second_hashes, algorithms, threshold): """Best similarity between two hash mappings, or None below threshold.""" best = None for field in algorithms: first = first_hashes.get(field) or "" second = second_hashes.get(field) or "" if not first or not second: continue try: distance = bin(int(first, 16) ^ int(second, 16)).count("1") except ValueError: continue value = 1.0 - distance / 64.0 if best is None or value > best: best = value if best is None or best < threshold: return None return best def similarity_between(item_a, item_b, algorithms, threshold): """Best similarity across the selected algorithms, or None below threshold.""" return hashes_similarity( {field: getattr(item_a, field, "") for field in algorithms}, {field: getattr(item_b, field, "") for field in algorithms}, algorithms, threshold, ) def hashed_items(algorithms): query = Q() for field in algorithms: query |= ~Q(**{field: ""}) return list(MediaItem.objects.filter(query).prefetch_related("locations")) def display_rating(item): if item.rating: return item.rating data = item.e621_data or {} rating = data.get("rating") if isinstance(data, dict) else None return rating if rating in {"s", "q", "e"} else "" def item_brief(item, request=None): locations = list(item.locations.all()) location = locations[0] if locations else None user = getattr(request, "user", None) return { "j_id": f"J-{item.id}", "md5": item.md5, "filename": Path(location.rel_path).name if location else item.md5, "size": item.size, "rating": display_rating(item), "location_count": len(locations), "uploaded_by": item.uploaded_by.username if item.uploaded_by else None, "e621_post_id": item.e621_post_id, "thumbnail_url": signed_media_url(item, user, "thumbnail", request=request), } def resolve_item(data): j_id = str(data.get("j_id") or "").strip() md5 = str(data.get("md5") or "").strip().lower() if j_id: numeric = j_id[2:] if j_id.upper().startswith("J-") else j_id if numeric.isdigit(): item = MediaItem.objects.filter(pk=int(numeric)).first() if item is not None: return item if md5: return MediaItem.objects.filter(md5=md5).first() return None class ExactDuplicatesView(APIView): """Items whose content exists at more than one path.""" permission_classes = [IsAuthenticated] def get(self, request): items = ( MediaItem.objects.annotate(location_count=Count("locations")) .filter(location_count__gt=1) .prefetch_related("locations") .order_by("-location_count", "id") ) groups = [] for item in items: brief = item_brief(item, request) brief["locations"] = [ {"id": location.id, "rel_path": location.rel_path} for location in item.locations.all() ] groups.append(brief) return Response({"count": len(groups), "groups": groups}) class VisualMatchesView(APIView): """Items visually similar to one library item.""" permission_classes = [IsAuthenticated] def post(self, request): threshold = parse_threshold(request.data.get("threshold")) algorithms = parse_algorithms(request.data.get("algorithms")) target = resolve_item(request.data) if target is None: return Response( {"detail": "A j_id or md5 is required."}, status=status.HTTP_400_BAD_REQUEST, ) matches = [] for item in hashed_items(algorithms): if item.pk == target.pk: continue similarity = similarity_between(target, item, algorithms, threshold) if similarity is None: continue brief = item_brief(item, request) brief["similarity"] = round(similarity * 100, 1) matches.append(brief) matches.sort(key=lambda entry: entry["similarity"], reverse=True) return Response( { "target": item_brief(target, request), "threshold": round(threshold * 100, 1), "algorithms": algorithms, "count": len(matches), "matches": matches[:200], } ) class VisualGroupsView(APIView): """Groups of visually similar items (union-find over perceptual hashes). Pairwise comparison is O(n^2) with fast bit operations, which is fine for a personal library. Revisit with a bucketed index if libraries grow huge. """ permission_classes = [IsAuthenticated] def post(self, request): threshold = parse_threshold(request.data.get("threshold")) algorithms = parse_algorithms(request.data.get("algorithms")) page = max(1, int(request.data.get("page") or 1)) items = hashed_items(algorithms) parent = list(range(len(items))) def find(index): while parent[index] != index: parent[index] = parent[parent[index]] index = parent[index] return index def union(first, second): root_a = find(first) root_b = find(second) if root_a != root_b: parent[root_b] = root_a for first in range(len(items)): for second in range(first + 1, len(items)): if find(first) == find(second): continue if ( similarity_between( items[first], items[second], algorithms, threshold ) is not None ): union(first, second) grouped = {} for index, item in enumerate(items): grouped.setdefault(find(index), []).append(item) groups = [members for members in grouped.values() if len(members) >= 2] groups.sort(key=len, reverse=True) total = len(groups) start = (page - 1) * GROUPS_PER_PAGE page_groups = groups[start : start + GROUPS_PER_PAGE] return Response( { "count": total, "page": page, "per_page": GROUPS_PER_PAGE, "has_next": start + GROUPS_PER_PAGE < total, "threshold": round(threshold * 100, 1), "algorithms": algorithms, "groups": [ { "size": len(members), "members": [item_brief(item, request) for item in members], } for members in page_groups ], } ) def can_delete(user, item): if user.is_app_staff: return True return item.uploaded_by_id == user.id def remove_watched_file(path): """Delete a file only when it lives inside the watched folder.""" watched = Path(settings.WATCHED_FOLDER).resolve() try: resolved = Path(path).resolve() resolved.relative_to(watched) except (ValueError, OSError): return False if resolved.is_file(): resolved.unlink() return True return False class DeleteFilesView(APIView): """Delete items (with every copy) or individual duplicate locations.""" permission_classes = [CanUpload] def post(self, request): j_ids = request.data.get("j_ids") or [] location_ids = request.data.get("location_ids") or [] if not isinstance(j_ids, list) or not isinstance(location_ids, list): return Response( {"detail": "j_ids and location_ids must be lists."}, status=status.HTTP_400_BAD_REQUEST, ) deleted = [] errors = [] numeric_ids = [] for value in j_ids: text = str(value).strip() numeric = text[2:] if text.upper().startswith("J-") else text if numeric.isdigit(): numeric_ids.append(int(numeric)) for item in MediaItem.objects.filter(pk__in=numeric_ids).prefetch_related( "locations" ): if not can_delete(request.user, item): errors.append({"j_id": f"J-{item.id}", "error": "permission denied"}) continue for location in item.locations.all(): remove_watched_file(location.path) deleted.append(f"J-{item.id}") item.delete() cache.delete(STORAGE_CACHE_KEY) cache.delete_many(TAG_CLOUD_CACHE_KEYS) for location in MediaLocation.objects.filter( id__in=location_ids ).select_related("item"): if not can_delete(request.user, location.item): errors.append( {"location": location.id, "error": "permission denied"} ) continue location_id = location.id item = location.item remove_watched_file(location.path) location.delete() if item.locations.exists(): deleted.append(f"location {location_id}") else: deleted.append(f"J-{item.id}") item.delete() return Response({"deleted": deleted, "errors": errors}) class ClearTempView(APIView): """Remove staged files from the temp upload folder.""" permission_classes = [CanUpload] def post(self, request): temp_dir = Path(settings.MEDIA_ROOT) / "uploads" / "temp" removed = 0 if temp_dir.exists(): for entry in temp_dir.iterdir(): if entry.is_file(): entry.unlink() removed += 1 cache.delete(STORAGE_CACHE_KEY) return Response({"removed": removed}) class TagCloudView(APIView): """Most-used tags across the library (custom + e621 tags).""" permission_classes = [AllowAny] def get(self, request): cache_key = ( "j621.library.tag_cloud.auth" if request.user.is_authenticated else "j621.library.tag_cloud.guest" ) cached = cache.get(cache_key) if cached is not None: return Response(cached) queryset = MediaItem.objects.all() if not request.user.is_authenticated: queryset = queryset.filter(hidden_from_guests=False) counts = {} for item in queryset.only("tags", "e621_data").iterator(chunk_size=500): names = set() for tag in item.tags or []: name = str(tag).strip().lower() if name: names.add(name) data = item.e621_data or {} categories = data.get("tags") if isinstance(data, dict) else None if isinstance(categories, dict): for values in categories.values(): if isinstance(values, list): names.update( str(tag).strip().lower() for tag in values if str(tag).strip() ) for name in names: counts[name] = counts.get(name, 0) + 1 ranked = sorted(counts.items(), key=lambda entry: (-entry[1], entry[0])) payload = { "count": len(counts), "tags": [ {"tag": tag, "count": count} for tag, count in ranked[:120] ], } cache.set(cache_key, payload, 120) return Response(payload) def storage_info(): """Disk usage payload, cached briefly (also used by the stats dashboard).""" cached = cache.get(STORAGE_CACHE_KEY) if cached is not None: return cached watched = Path(settings.WATCHED_FOLDER) media_root = Path(settings.MEDIA_ROOT) temp_dir = media_root / "uploads" / "temp" usage = shutil.disk_usage( str(watched) if watched.exists() else str(Path(settings.BASE_DIR)) ) library_size = ( sum(file.stat().st_size for file in watched.rglob("*") if file.is_file()) if watched.exists() else 0 ) media_size = ( sum(file.stat().st_size for file in media_root.rglob("*") if file.is_file()) if media_root.exists() else 0 ) temp_files = ( [file for file in temp_dir.rglob("*") if file.is_file()] if temp_dir.exists() else [] ) payload = { "watched_folder": { "path": str(watched), "total": usage.total, "used": usage.used, "free": usage.free, "library_size": library_size, "percent_used": ( round(usage.used / usage.total * 100, 1) if usage.total else 0 ), }, "media": {"path": str(media_root), "size": media_size}, "temp": { "path": str(temp_dir), "size": sum(file.stat().st_size for file in temp_files), "files": len(temp_files), }, "library_items": MediaItem.objects.count(), } cache.set(STORAGE_CACHE_KEY, payload, STORAGE_CACHE_TTL) return payload class StorageView(APIView): """Disk usage for the watched folder, media root and temp uploads.""" permission_classes = [IsAuthenticated] def get(self, request): return Response(storage_info())