J621/backend/apps/library/tools.py

"""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 IsAuthenticated
from rest_framework.response import Response
from rest_framework.views import APIView

from .models import MediaItem, MediaLocation
from .permissions import CanUpload

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):
    locations = list(item.locations.all())
    location = locations[0] if locations else 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,
    }


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)
            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)
            brief["similarity"] = round(similarity * 100, 1)
            matches.append(brief)
        matches.sort(key=lambda entry: entry["similarity"], reverse=True)

        return Response(
            {
                "target": item_brief(target),
                "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) for item in members]}
                    for members in page_groups
                ],
            }
        )


def can_delete(user, item):
    if user.is_superuser or user.role == user.ROLE_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)

        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 StorageView(APIView):
    """Disk usage for the watched folder, media root and temp uploads."""

    permission_classes = [IsAuthenticated]

    def get(self, request):
        cached = cache.get(STORAGE_CACHE_KEY)
        if cached is not None:
            return Response(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 Response(payload)