# 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/>.
from django.db import transaction
from django.db.models import Prefetch
from rest_framework import status
from rest_framework.decorators import action
from rest_framework.exceptions import ValidationError
from rest_framework.permissions import IsAuthenticated
from rest_framework.response import Response
from nlpmed_portal.annotations.api.serializers import LabelCategorySerializer
from nlpmed_portal.annotations.api.serializers import NERAnnotationSerializer
from nlpmed_portal.annotations.api.serializers import PatientAnnotationDetailSerializer
from nlpmed_portal.annotations.api.serializers import PatientAnnotationListSerializer
from nlpmed_portal.annotations.api.viewsets.base import BaseViewSet
from nlpmed_portal.annotations.models import Adjudication
from nlpmed_portal.annotations.models import LabelValue
from nlpmed_portal.annotations.models import NERAnnotation
from nlpmed_portal.annotations.models import Note
from nlpmed_portal.annotations.models import Patient
from nlpmed_portal.annotations.models import PatientAnnotation
from nlpmed_portal.annotations.models import Project
from nlpmed_portal.annotations.permissions import MethodBasePermission
from nlpmed_portal.annotations.permissions import has_global_permission
from nlpmed_portal.annotations.permissions import has_permission
[docs]
class PatientAnnotationViewSet(BaseViewSet):
queryset = PatientAnnotation.objects.select_related(
"patient",
"annotator",
).prefetch_related(
Prefetch(
"event_label_values",
queryset=LabelValue.objects.select_related("category"),
),
Prefetch(
"notes",
queryset=Note.objects.all(),
),
)
permission_classes = [IsAuthenticated, MethodBasePermission]
required_perms = {
"GET": "view_patientannotation",
"HEAD": "view_patientannotation",
"OPTIONS": "view_patientannotation",
"POST": "add_patientannotation",
"PUT": "change_patientannotation",
"PATCH": "change_patientannotation",
"DELETE": "delete_patientannotation",
}
[docs]
def get_serializer_class(self):
if self.action == "list":
return PatientAnnotationListSerializer
return PatientAnnotationDetailSerializer
[docs]
def get_queryset(self):
required_perm = self.required_perms.get("GET")
queryset = super().get_queryset()
patient_id = self.request.query_params.get("patient_id")
if patient_id:
if not patient_id.isdigit():
raise ValidationError(
{"patient_id": "Invalid patient_id. Must be an integer."},
)
queryset = queryset.filter(patient_id=patient_id)
user = self.request.user
can_view_all = False
if patient_id:
project_id = (
Patient.objects.filter(pk=patient_id).values_list("project_id", flat=True).first()
)
if project_id and has_permission(
user,
"view_all_patientannotations",
project=project_id,
):
can_view_all = True
elif has_global_permission(user, "view_all_patientannotations"):
can_view_all = True
if not can_view_all:
queryset = queryset.filter(annotator=user)
return self.filter_queryset_by_scope(
queryset,
required_perm,
project_field="patient__project",
)
[docs]
def destroy(self, request, *args, **kwargs):
inst = self.get_object()
if Adjudication.objects.filter(source_annotations=inst).exists():
raise ValidationError(
{
"detail": "This annotation is referenced by an \
adjudication. Remove the adjudication first.",
},
)
return super().destroy(request, *args, **kwargs)
[docs]
@action(detail=False, methods=["DELETE"])
def delete_all(self, request, *args, **kwargs):
patient_id = request.query_params.get("patient_id")
project_id = request.query_params.get("project_id")
if not patient_id and not project_id:
return Response(
{
"error": "Either patient_id or project_id must be \
provided for delete all operation.",
},
status=status.HTTP_400_BAD_REQUEST,
)
queryset = super().get_queryset()
project = None
if patient_id:
if not patient_id.isdigit():
raise ValidationError(
{"patient_id": "Invalid patient_id. Must be an integer."},
)
queryset = queryset.filter(patient_id=patient_id)
project = Patient.objects.get(id=patient_id).project
elif project_id:
if not project_id.isdigit():
raise ValidationError(
{"project_id": "Invalid project_id. Must be an integer."},
)
queryset = queryset.filter(patient__project_id=project_id)
project = Project.objects.get(id=project_id)
if not has_permission(
user=request.user,
codename="delete_all_patientannotations",
project=project,
):
return Response(
{"error": "No permission to delete all patient annotations."},
status=status.HTTP_403_FORBIDDEN,
)
referenced = Adjudication.objects.filter(
source_annotations__in=queryset,
).values_list("id", flat=True)[:1]
if referenced:
return Response(
{
"error": "At least one annotation is referenced by \
an adjudication. Remove adjudications first.",
},
status=status.HTTP_409_CONFLICT,
)
with transaction.atomic():
_, deleted_details = queryset.delete()
model_label = PatientAnnotation._meta.label # ruff: ignore[private-member-access]
model_deleted_count = deleted_details.get(model_label, 0)
return Response(
{"deleteCount": model_deleted_count},
status=status.HTTP_200_OK,
)
[docs]
class NERAnnotationViewSet(BaseViewSet):
queryset = NERAnnotation.objects.select_related("patient").all()
serializer_class = NERAnnotationSerializer
permission_classes = [IsAuthenticated, MethodBasePermission]
required_perms = {
"GET": "view_nerannotation",
"HEAD": "view_nerannotation",
"OPTIONS": "view_nerannotation",
"POST": "add_nerannotation",
"PUT": "change_nerannotation",
"PATCH": "change_nerannotation",
"DELETE": "delete_nerannotation",
}
[docs]
def get_queryset(self):
required_perm = self.required_perms.get("GET")
queryset = super().get_queryset()
patient_id = self.request.query_params.get("patient_id")
if patient_id:
if not patient_id.isdigit():
raise ValidationError(
{"patient_id": "Invalid patient_id. Must be an integer."},
)
queryset = queryset.filter(patient_id=patient_id)
queryset = self.filter_queryset_by_scope(
queryset,
required_perm,
project_field="patient__project",
)
user = self.request.user
can_view_all = False
if patient_id:
try:
patient = Patient.objects.get(pk=patient_id)
except Patient.DoesNotExist:
patient = None
if patient:
if has_permission(
user,
"view_all_nerannotations",
project=patient.project,
):
can_view_all = True
elif has_global_permission(user, "view_all_nerannotations"):
can_view_all = True
if not can_view_all:
queryset = queryset.filter(annotator=user)
return queryset
[docs]
@action(detail=False, methods=["DELETE"])
def delete_all(self, request, *args, **kwargs):
patient_id = request.query_params.get("patient_id")
project_id = request.query_params.get("project_id")
if not patient_id and not project_id:
return Response(
{
"error": "Either patient_id or project_id must be \
provided for delete all operation.",
},
status=status.HTTP_400_BAD_REQUEST,
)
queryset = super().get_queryset()
project = None
if patient_id:
if not patient_id.isdigit():
raise ValidationError(
{"patient_id": "Invalid patient_id. Must be an integer."},
)
queryset = queryset.filter(patient_id=patient_id)
project = Patient.objects.get(id=patient_id).project
elif project_id:
if not project_id.isdigit():
raise ValidationError(
{"project_id": "Invalid project_id. Must be an integer."},
)
queryset = queryset.filter(patient__project_id=project_id)
project = Project.objects.get(id=project_id)
if not has_permission(
user=request.user,
codename="delete_all_nerannotations",
project=project,
):
return Response(
{"error": "No permission to delete all ner annotations."},
status=status.HTTP_403_FORBIDDEN,
)
_, deleted_details = queryset.delete()
model_label = NERAnnotation._meta.label # ruff: ignore[private-member-access]
model_deleted_count = deleted_details.get(model_label, 0)
return Response(
{"deleteCount": model_deleted_count},
status=status.HTTP_200_OK,
)