# 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.conf import settings
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 LabSerializer
from nlpmed_portal.annotations.api.serializers import NoteSerializer
from nlpmed_portal.annotations.api.viewsets.base import BaseViewSet
from nlpmed_portal.annotations.models import Lab
from nlpmed_portal.annotations.models import Note
from nlpmed_portal.annotations.models import Patient
from nlpmed_portal.annotations.models import Project
from nlpmed_portal.annotations.models import Section
from nlpmed_portal.annotations.models import Sentence
from nlpmed_portal.annotations.permissions import MethodBasePermission
from nlpmed_portal.annotations.permissions import has_permission
from nlpmed_portal.annotations.utils import build_html_hover_text
from nlpmed_portal.annotations.utils import build_html_note_text
from nlpmed_portal.annotations.utils import pivot_labs
[docs]
class NoteViewSet(BaseViewSet):
serializer_class = NoteSerializer
queryset = (
Note.objects
.select_related("patient")
.prefetch_related(
Prefetch(
"note_sections",
queryset=Section.objects.prefetch_related(
Prefetch(
"section_sentences",
queryset=Sentence.objects.all(),
),
),
),
)
.all()
)
permission_classes = [IsAuthenticated, MethodBasePermission]
required_perms = {
"GET": "view_note",
"HEAD": "view_note",
"OPTIONS": "view_note",
"POST": "add_note",
"PUT": "change_note",
"PATCH": "change_note",
"DELETE": "delete_note",
}
[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)
return self.filter_queryset_by_scope(
queryset,
required_perm,
project_field="patient__project",
)
[docs]
def list(self, request, *args, **kwargs):
return_html = request.query_params.get("return_html") == "1"
if not return_html:
return super().list(request, *args, **kwargs)
queryset = self.filter_queryset(self.get_queryset())
patient_id = request.query_params.get("patient_id")
if not patient_id:
raise ValidationError(
{"error": "patient_id is required when return_html=1"},
)
patient = Patient.objects.select_related(
"project__nlp_setting__note_filter",
).get(pk=patient_id)
if patient.project.run_nlp:
keywords = patient.project.nlp_setting.note_filter.words_to_search
else:
keywords = None
data = []
for note in queryset:
note_sections = note.note_sections.all()
html_note_text = build_html_note_text(
note,
note_sections,
keywords,
)
html_hover_text = build_html_hover_text(
note_sections,
keywords,
)
data.append(
{
"id": note.pk,
"note_date": note.note_date.strftime(settings.DATE_FORMAT)
if note.note_date
else None,
"note_type": note.note_type,
"predicted_label": note.predicted_label,
"predicted_score": note.predicted_score,
"important_sentences_count": getattr(
note,
"important_sentences_count",
0,
),
"duplicate_sentences_count": getattr(
note,
"duplicate_sentences_count",
0,
),
"html_note_text": html_note_text,
"html_hover_text": html_hover_text,
},
)
return Response(data, status=status.HTTP_200_OK)
[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_notes",
project=project,
):
return Response(
{"error": "No permission to delete all notes."},
status=status.HTTP_403_FORBIDDEN,
)
_, deleted_details = queryset.delete()
model_label = Note._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 LabViewSet(BaseViewSet):
serializer_class = LabSerializer
queryset = Lab.objects.select_related("patient").all()
permission_classes = [IsAuthenticated, MethodBasePermission]
required_perms = {
"GET": "view_lab",
"HEAD": "view_lab",
"OPTIONS": "view_lab",
"POST": "add_lab",
"PUT": "change_lab",
"PATCH": "change_lab",
"DELETE": "delete_lab",
}
[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)
return self.filter_queryset_by_scope(
queryset,
required_perm,
project_field="patient__project",
)
[docs]
@action(detail=False, methods=["GET"])
def pivot(self, request, *args, **kwargs):
labs = self.get_queryset()
return Response(
pivot_labs(labs),
status=status.HTTP_200_OK,
)
[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_labs",
project=project,
):
return Response(
{"error": "No permission to delete all labs."},
status=status.HTTP_403_FORBIDDEN,
)
_, deleted_details = queryset.delete()
model_label = Lab._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,
)