J621/backend/apps/accounts/views.py

import logging

from rest_framework import mixins, status, viewsets
from rest_framework.authtoken.models import Token
from rest_framework.authtoken.views import ObtainAuthToken
from rest_framework.permissions import AllowAny, IsAuthenticated
from rest_framework.response import Response
from rest_framework.throttling import ScopedRateThrottle
from rest_framework.views import APIView
from django.db.models import Count, Q

from apps.core.permissions import IsAppStaff
from apps.library.models import MediaItem

from .crypto import encrypt_secret
from .models import User
from .serializers import (
    E621CredentialsSerializer,
    PreferencesSerializer,
    RegisterSerializer,
    UserListSerializer,
    UserSerializer,
    UserUpdateSerializer,
)

logger = logging.getLogger(__name__)


class LoginView(ObtainAuthToken):
    """Token login, rate limited per IP to slow down credential stuffing."""

    throttle_classes = [ScopedRateThrottle]
    throttle_scope = "login"


class RegisterView(APIView):
    permission_classes = [AllowAny]
    throttle_classes = [ScopedRateThrottle]
    throttle_scope = "register"

    def post(self, request):
        serializer = RegisterSerializer(data=request.data)
        serializer.is_valid(raise_exception=True)
        user = serializer.save()
        token, _ = Token.objects.get_or_create(user=user)
        return Response(
            {
                "token": token.key,
                "user": UserSerializer(user, context={"request": request}).data,
            },
            status=status.HTTP_201_CREATED,
        )


class MeView(APIView):
    permission_classes = [IsAuthenticated]

    def get(self, request):
        return Response(UserSerializer(request.user, context={"request": request}).data)


class LogoutView(APIView):
    permission_classes = [IsAuthenticated]

    def post(self, request):
        if request.auth:
            request.auth.delete()
        return Response(status=status.HTTP_204_NO_CONTENT)


class E621CredentialsView(APIView):
    """Store the e621 credentials used by the user's browser to call e621 directly."""

    permission_classes = [IsAuthenticated]

    def _payload(self, user):
        return {
            "username": user.e621_username,
            "api_key": user.e621_api_key_plain,
            "base_url": user.e621_base_url,
            "configured": user.e621_configured,
        }

    def get(self, request):
        return Response(self._payload(request.user))

    def put(self, request):
        serializer = E621CredentialsSerializer(data=request.data)
        serializer.is_valid(raise_exception=True)
        data = serializer.validated_data
        user = request.user
        user.e621_username = data["username"].strip()
        user.e621_api_key = encrypt_secret(data["api_key"].strip())
        user.e621_base_url = (
            (data.get("base_url") or "https://e621.net").strip().rstrip("/")
        )
        user.save(
            update_fields=["e621_username", "e621_api_key", "e621_base_url"]
        )
        return Response(self._payload(user))


def resolve_avatar_item(value):
    """Turn a "J-42" / "42" string into a MediaItem.

    Returns ``(item, error)``: an empty value clears the avatar (``None``,
    ``None``), an unknown item returns an error message.
    """
    value = str(value or "").strip()
    if not value:
        return None, None
    numeric = value[2:] if value.upper().startswith("J-") else value
    item = (
        MediaItem.objects.filter(pk=int(numeric)).first()
        if numeric.isdigit()
        else None
    )
    if item is None:
        return None, f"No library item {value}."
    return item, None


class AvatarView(APIView):
    """Self-service profile picture chosen from the library."""

    permission_classes = [IsAuthenticated]

    def post(self, request):
        serializer = UserUpdateSerializer(data=request.data)
        serializer.is_valid(raise_exception=True)
        item, error = resolve_avatar_item(
            serializer.validated_data.get("avatar_j_id")
        )
        if error:
            return Response({"detail": error}, status=status.HTTP_400_BAD_REQUEST)
        user = request.user
        user.avatar = item
        user.save(update_fields=["avatar"])
        return Response(UserSerializer(user, context={"request": request}).data)


class PreferencesView(APIView):
    """Per-user browse preferences, merged so callers can send single keys."""

    permission_classes = [IsAuthenticated]

    def get(self, request):
        return Response(request.user.preferences or {})

    def post(self, request):
        serializer = PreferencesSerializer(data=request.data)
        serializer.is_valid(raise_exception=True)
        preferences = dict(request.user.preferences or {})
        preferences.update(serializer.validated_data)
        request.user.preferences = preferences
        request.user.save(update_fields=["preferences"])
        return Response(preferences)


class UserViewSet(
    mixins.ListModelMixin,
    mixins.RetrieveModelMixin,
    mixins.DestroyModelMixin,
    viewsets.GenericViewSet,
):
    """Staff user directory: roles, J-ID avatars, and account deletion."""

    permission_classes = [IsAppStaff]
    http_method_names = ["get", "patch", "delete", "head", "options"]

    def get_queryset(self):
        queryset = User.objects.annotate(uploads_count=Count("uploads")).order_by("id")
        search = self.request.query_params.get("search", "").strip()
        if search:
            queryset = queryset.filter(
                Q(username__icontains=search) | Q(email__icontains=search)
            )
        return queryset

    def get_serializer_class(self):
        if self.request.method in {"PATCH", "PUT"}:
            return UserUpdateSerializer
        return UserListSerializer

    def update(self, request, *args, **kwargs):
        user = self.get_object()
        actor = request.user
        serializer = UserUpdateSerializer(data=request.data)
        serializer.is_valid(raise_exception=True)
        data = serializer.validated_data

        update_fields = []
        if "role" in data:
            new_role = data["role"]
            # Same rule as account deletion: only admins may move accounts
            # across the staff boundary (granting or revoking staff).
            if (
                not actor.is_superuser
                and new_role != user.role
                and (user.is_app_staff or new_role == User.ROLE_STAFF)
            ):
                return Response(
                    {
                        "detail": "Only an admin can change staff roles.",
                    },
                    status=status.HTTP_403_FORBIDDEN,
                )
            user.role = new_role
            update_fields.append("role")
        if "avatar_j_id" in data:
            item, error = resolve_avatar_item(data.get("avatar_j_id"))
            if error:
                return Response(
                    {"detail": error},
                    status=status.HTTP_400_BAD_REQUEST,
                )
            user.avatar = item
            update_fields.append("avatar")
        if update_fields:
            user.save(update_fields=update_fields)

        return Response(
            UserListSerializer(user, context=self.get_serializer_context()).data
        )

    partial_update = update

    def destroy(self, request, *args, **kwargs):
        target = self.get_object()
        actor = request.user
        if target.pk == actor.pk:
            return Response(
                {"detail": "You cannot delete the account you are signed in as."},
                status=status.HTTP_400_BAD_REQUEST,
            )
        if not actor.is_superuser and target.is_app_staff:
            return Response(
                {"detail": "Only an admin can delete staff or admin accounts."},
                status=status.HTTP_403_FORBIDDEN,
            )
        if (
            target.is_superuser
            and User.objects.filter(is_superuser=True).count() <= 1
        ):
            return Response(
                {"detail": "This is the last admin account and cannot be deleted."},
                status=status.HTTP_400_BAD_REQUEST,
            )

        username = target.username
        # Staged upload files live on disk; the row cascade would orphan them.
        for temp in target.temp_uploads.all():
            if temp.file:
                temp.file.delete(save=False)
        target.delete()
        logger.info("Account %s deleted by %s", username, actor.username)
        return Response(status=status.HTTP_204_NO_CONTENT)