Source code for nlpmed_portal.management.api.views

# SPDX-FileCopyrightText: Copyright (C) 2026 Omid Jafari <omidjafari.com>
# SPDX-License-Identifier: AGPL-3.0-or-later
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU Affero General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
# GNU Affero General Public License for more details.
#
# You should have received a copy of the GNU Affero General Public License
# along with this program.  If not, see <http://www.gnu.org/licenses/>.

import contextlib
from typing import Any

import httpx
from allauth.account.forms import SetPasswordForm
from allauth.usersessions.models import UserSession
from django.conf import settings
from django.contrib.auth.models import Group
from django.contrib.auth.models import Permission
from django.db.models import Prefetch
from rest_framework import status
from rest_framework.decorators import action
from rest_framework.mixins import ListModelMixin
from rest_framework.mixins import RetrieveModelMixin
from rest_framework.mixins import UpdateModelMixin
from rest_framework.permissions import IsAuthenticated
from rest_framework.request import Request
from rest_framework.response import Response
from rest_framework.viewsets import GenericViewSet
from rest_framework.viewsets import ModelViewSet
from rest_framework.viewsets import ViewSet

from nlpmed_portal.management.api.serializers import FeedbackSerializer
from nlpmed_portal.management.api.serializers import GroupDetailSerializer
from nlpmed_portal.management.api.serializers import GroupListSerializer
from nlpmed_portal.management.api.serializers import UserDetailSerializer
from nlpmed_portal.management.api.serializers import UserListSerializer
from nlpmed_portal.management.api.serializers import UserSessionSerializer
from nlpmed_portal.management.constants import PERMISSION_CODENAMES
from nlpmed_portal.management.models import Feedback
from nlpmed_portal.management.permissions import MethodBasePermission
from nlpmed_portal.users.models import User




class UserViewSet(RetrieveModelMixin, ListModelMixin, UpdateModelMixin, GenericViewSet):
    queryset = User.objects.filter(is_staff=False).prefetch_related("groups")
    permission_classes = [IsAuthenticated, MethodBasePermission]
    required_perms = {
        "GET": "management.view_user",
        "HEAD": "management.view_user",
        "OPTIONS": "management.view_user",
        "POST": "management.change_user",
        "PUT": "management.change_user",
        "PATCH": "management.change_user",
    }



    def get_serializer_class(self):
        if self.action == "list":
            return UserListSerializer

        return UserDetailSerializer




    @action(detail=False, methods=["GET"])
    def meta(self, request):
        data = {
            "all_groups": list(Group.objects.values("id", "name")),
        }
        return Response(data)




    @action(detail=True, methods=["POST"])
    def set_password(self, request, pk=None):
        user = self.get_object()
        form = SetPasswordForm(data=request.data, user=user)

        if not form.is_valid():
            return Response(form.errors, status=status.HTTP_400_BAD_REQUEST)

        form.save()

        return Response({"detail": "Password set."}, status=status.HTTP_200_OK)






class UserSessionViewSet(RetrieveModelMixin, ListModelMixin, GenericViewSet):
    queryset = User.objects.all()
    permission_classes = [IsAuthenticated, MethodBasePermission]
    required_perms = {
        "GET": "management.view_user",
        "HEAD": "management.view_user",
        "OPTIONS": "management.view_user",
    }



    def retrieve(self, request: Request, *args: Any, **kwargs: Any) -> Response:
        user = self.get_object()
        sessions = UserSession.objects.purge_and_list(user)
        sessions = sorted(sessions, key=lambda s: s.created_at)
        serializer = UserSessionSerializer(sessions, many=True)
        return Response(serializer.data)






class GroupViewSet(ModelViewSet):
    allowed_perms_qs = Permission.objects.filter(
        content_type__app_label__in=["management", "nlp", "annotations"],
        codename__in=PERMISSION_CODENAMES,
    ).select_related("content_type")
    queryset = Group.objects.all().prefetch_related(
        Prefetch("permissions", queryset=allowed_perms_qs),
    )
    permission_classes = [IsAuthenticated, MethodBasePermission]
    required_perms = {
        "GET": "management.view_group",
        "HEAD": "management.view_group",
        "OPTIONS": "management.view_group",
        "POST": "management.add_group",
        "PUT": "management.change_group",
        "PATCH": "management.change_group",
        "DELETE": "management.delete_group",
    }



    def get_serializer_class(self):
        if self.action == "list":
            return GroupListSerializer

        return GroupDetailSerializer




    def get_serializer_context(self):
        ctx = dict(super().get_serializer_context())
        if self.action != "list":
            ctx["all_perms"] = [{"pk": p.pk, "name": str(p)} for p in self.allowed_perms_qs]
        return ctx




    def destroy(self, *args, **kwargs):
        serializer = self.get_serializer(self.get_object())
        super().destroy(*args, **kwargs)
        return Response(serializer.data, status=status.HTTP_200_OK)




    @action(detail=False, methods=["GET"])
    def meta(self, request):
        return Response(
            {
                "all_permissions": self.get_serializer_context().get("all_perms")
                or [{"pk": p.pk, "name": str(p)} for p in self.allowed_perms_qs],
            },
        )






class BackendViewSet(ViewSet):
    permission_classes = [IsAuthenticated, MethodBasePermission]
    required_perms = {
        "GET": "management.view_backend",
        "HEAD": "management.view_backend",
        "OPTIONS": "management.view_backend",
    }



    @staticmethod
    def check_api_state(api_url: str) -> str:
        try:
            resp = httpx.get(api_url, timeout=3, verify=False)  # ruff: ignore[request-with-no-cert-validation]

        except httpx.RequestError:
            return "unreachable"

        if resp.status_code == status.HTTP_200_OK:
            return "reachable"
        return "unreachable"




    @staticmethod
    def fetch_nlp_models_meta(api_url: str) -> dict[str, Any] | None:
        with contextlib.suppress(httpx.RequestError):
            resp = httpx.get(
                f"{api_url.rstrip('/')}/ml_models",
                timeout=3,
                verify=False,  # ruff: ignore[request-with-no-cert-validation]
            )

            if resp.status_code == status.HTTP_200_OK:
                data = resp.json()

                if isinstance(data, dict) and "models" in data:
                    models = [
                        {
                            "name": m.get("name"),
                            "device": m.get("device"),
                            "max_length": m.get("max_length"),
                            "loaded": m.get("loaded"),
                            "loaded_at": m.get("loaded_at"),
                        }
                        for m in data.get("models", [])
                    ]

                    return {"default_name": data.get("default_name"), "models": models}

        return None




    def list(self, request):
        nlp_url = settings.NLP_API_URL

        nlp_status = self.check_api_state(nlp_url)
        models_meta = self.fetch_nlp_models_meta(nlp_url) if nlp_status == "reachable" else None

        backends_data = {
            "nlp": {
                "url": nlp_url,
                "status": nlp_status,
                "models": models_meta,
            },
        }

        return Response(backends_data, status=status.HTTP_200_OK)






class FeedbackViewSet(ModelViewSet):
    serializer_class = FeedbackSerializer
    queryset = Feedback.objects.select_related("user").all()
    permission_classes = [IsAuthenticated, MethodBasePermission]
    required_perms = {
        "GET": "management.view_feedback",
        "HEAD": "management.view_feedback",
        "OPTIONS": "management.view_feedback",
        "POST": "management.add_feedback",
        "PUT": "management.change_feedback",
        "PATCH": "management.change_feedback",
        "DELETE": "management.delete_feedback",
    }



    def destroy(self, *args, **kwargs):
        serializer = self.get_serializer(self.get_object())
        super().destroy(*args, **kwargs)
        return Response(serializer.data, status=status.HTTP_200_OK)