import re from pathlib import Path from urllib.parse import urlparse from django.conf import settings from django.db.models import Min from django.http import Http404 from django.shortcuts import get_object_or_404 from rest_framework import mixins, status, viewsets from rest_framework.decorators import action from rest_framework.permissions import AllowAny, IsAuthenticatedOrReadOnly from rest_framework.response import Response from rest_framework.views import APIView from . import services from .models import MediaItem from .permissions import CanUpload, IsUploaderOrStaffOrReadOnly from .serializers import MediaItemSerializer LIST_ORDERINGS = {"name", "-name", "size", "-size", "created_at", "-created_at"} MD5_RE = re.compile(r"[0-9a-fA-F]{32}") class MediaItemViewSet( mixins.ListModelMixin, mixins.RetrieveModelMixin, mixins.UpdateModelMixin, viewsets.GenericViewSet, ): serializer_class = MediaItemSerializer permission_classes = [IsAuthenticatedOrReadOnly, IsUploaderOrStaffOrReadOnly] http_method_names = ["get", "post", "patch", "head", "options"] def get_queryset(self): queryset = MediaItem.objects.prefetch_related("locations").annotate( name=Min("locations__rel_path") ) if not self.request.user.is_authenticated: queryset = queryset.filter(hidden_from_guests=False) search = self.request.query_params.get("search", "").strip() if search: queryset = queryset.filter(locations__rel_path__icontains=search) ratings = [ value for value in self.request.query_params.get("rating", "").split(",") if value in {"s", "q", "e"} ] if ratings: queryset = queryset.filter(rating__in=ratings) ordering = self.request.query_params.get("ordering", "").strip() queryset = queryset.order_by( ordering if ordering in LIST_ORDERINGS else "-created_at" ) return queryset.distinct() def get_object(self): """Resolve J-, a bare numeric id, or an MD5 fingerprint.""" value = str(self.kwargs.get("pk", "")).strip() queryset = self.get_queryset() if MD5_RE.fullmatch(value): obj = get_object_or_404(queryset, md5=value.lower()) else: numeric = value[2:] if value.upper().startswith("J-") else value if not numeric.isdigit(): raise Http404 obj = get_object_or_404(queryset, pk=int(numeric)) self.check_object_permissions(self.request, obj) return obj @action(detail=True, methods=["get"]) def raw(self, request, pk=None): item = self.get_object() location = item.locations.first() if location is None: return Response( {"detail": "No file for this item."}, status=status.HTTP_404_NOT_FOUND, ) return services.serve_file( request, location.path, download=request.query_params.get("download") == "1" ) @action(detail=True, methods=["get"]) def thumbnail(self, request, pk=None): item = self.get_object() location = item.locations.first() if location is None: return Response( {"detail": "No file for this item."}, status=status.HTTP_404_NOT_FOUND, ) path = Path(location.path) if path.suffix.lower() in services.VIDEO_EXTENSIONS: thumbnail = services.generate_video_thumbnail(item.md5, path) if thumbnail is None: return Response( {"detail": "Thumbnail unavailable."}, status=status.HTTP_404_NOT_FOUND, ) return services.serve_file(request, thumbnail) return services.serve_file(request, path) @action(detail=False, methods=["post"], permission_classes=[AllowAny]) def lookup(self, request): """Report which of the given MD5s exist in the library.""" md5s = request.data.get("md5s") if not isinstance(md5s, list): return Response( {"detail": "md5s must be a list."}, status=status.HTTP_400_BAD_REQUEST, ) wanted = {str(value).strip().lower()[:32] for value in md5s if value} queryset = MediaItem.objects.filter(md5__in=wanted) if not request.user.is_authenticated: queryset = queryset.filter(hidden_from_guests=False) rows = list(queryset.values_list("md5", "id")) return Response( { "found": sorted(md5 for md5, _ in rows), "j_ids": {md5: f"J-{item_id}" for md5, item_id in rows}, } ) @action(detail=True, methods=["get"]) def neighbors(self, request, pk=None): """Previous/next items in the ordered list, for keyboard navigation.""" item = self.get_object() ordering = request.query_params.get("ordering", "-created_at").strip() if ordering not in LIST_ORDERINGS: ordering = "-created_at" field = ordering.lstrip("-") descending = ordering.startswith("-") value = { "size": item.size, "created_at": item.created_at, "name": item.name, }[field] queryset = self.get_queryset().exclude(pk=item.pk) if descending: previous_query = queryset.filter(**{f"{field}__gt": value}).order_by(field) next_query = queryset.filter(**{f"{field}__lt": value}).order_by(f"-{field}") else: previous_query = queryset.filter(**{f"{field}__lt": value}).order_by( f"-{field}" ) next_query = queryset.filter(**{f"{field}__gt": value}).order_by(field) def brief(obj): location = obj.locations.first() return { "j_id": f"J-{obj.id}", "md5": obj.md5, "filename": Path(location.rel_path).name if location else obj.md5, } previous = previous_query.first() next_item = next_query.first() return Response( { "previous": brief(previous) if previous else None, "next": brief(next_item) if next_item else None, } ) class DownloadToLibraryView(APIView): """Fetch an e621 file into the watched folder and index it.""" permission_classes = [CanUpload] def post(self, request): url = str(request.data.get("url") or "").strip() post_id = request.data.get("post_id") filename = str(request.data.get("filename") or "").strip() post_payload = request.data.get("post") if not url.startswith(("http://", "https://")): return Response( {"detail": "A valid file URL is required."}, status=status.HTTP_400_BAD_REQUEST, ) folder = Path(settings.WATCHED_FOLDER) name = ( filename or Path(urlparse(url).path).name or f"post-{post_id or 'download'}" ) destination = services.unique_destination(folder, name) try: services.download_file(url, destination) item, _, _ = services.index_file(destination, folder) except Exception as exc: # noqa: BLE001 - report download failures destination.unlink(missing_ok=True) return Response( {"detail": f"Download failed: {exc}"}, status=status.HTTP_400_BAD_REQUEST, ) update_fields = [] if item.uploaded_by_id is None: item.uploaded_by = request.user update_fields.append("uploaded_by") trimmed = services.trim_e621_post(post_payload) if trimmed is not None: if trimmed.get("id") is not None: item.e621_post_id = int(trimmed["id"]) elif post_id and str(post_id).isdigit(): item.e621_post_id = int(post_id) item.e621_data = trimmed update_fields += ["e621_post_id", "e621_data"] if not item.rating and trimmed.get("rating") in {"s", "q", "e"}: item.rating = trimmed["rating"] update_fields.append("rating") if update_fields: item.save(update_fields=update_fields + ["updated_at"]) return Response( MediaItemSerializer(item, context={"request": request}).data, status=status.HTTP_201_CREATED, )