# 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 re
from typing import Any
from auditlog.models import LogEntry
from django.contrib.contenttypes.models import ContentType
from rest_framework import status
from rest_framework.exceptions import ValidationError
from rest_framework.permissions import IsAuthenticated
from rest_framework.response import Response
from rest_framework.viewsets import ViewSet
from nlpmed_portal.annotations.api.serializers import PatientHistorySerializer
from nlpmed_portal.annotations.models import Patient
from nlpmed_portal.annotations.models import PatientAnnotation
from nlpmed_portal.annotations.permissions import MethodBasePermission
from nlpmed_portal.annotations.permissions import has_permission
[docs]
class PatientHistoryViewSet(ViewSet):
serializer_class = PatientHistorySerializer
permission_classes = [IsAuthenticated, MethodBasePermission]
required_perms = {
"GET": "view_patientannotation_history",
"HEAD": "view_patientannotation_history",
"OPTIONS": "view_patientannotation_history",
}
@staticmethod
def _format_auditlog_object_string(field: str, value: str) -> str:
if not value:
return value
if field == "notes":
lines = []
for line in value.split("\n"):
stripped_line = line.strip()
m = re.match(r"^Note\s+(\d+)\b", stripped_line)
lines.append(m.group(1) if m else stripped_line)
return ", ".join(lines)
if field == "event_label_values":
lines = []
for line in value.split("\n"):
stripped_line = line.strip()
m = re.match(r"^(.*?)\s+\((.*?):\s+(.*?)\)$", stripped_line)
if m:
label_value_name = m.group(1).strip()
category_name = m.group(3).strip()
lines.append(f"{category_name}: {label_value_name}")
else:
lines.append(stripped_line)
return ", ".join(lines)
return value
def _parse_log_entry(self, log: LogEntry) -> dict[str, Any]:
actor = getattr(log.actor, "username", None)
parsed_changes: list[dict] = []
changes = log.changes_dict or {}
for field, value in changes.items():
# m2m fields
if isinstance(value, dict) and value.get("type") == "m2m":
objects = value.get("objects", []) or []
operation = value.get("operation", "")
joined_objects = "\n".join(objects) if objects else None
formatted_objects = (
self._format_auditlog_object_string(field, joined_objects)
if joined_objects
else None
)
if operation == "add":
old_value = None
new_value = formatted_objects
elif operation == "delete":
old_value = formatted_objects
new_value = None
else:
old_value = None
new_value = formatted_objects
parsed_changes.append(
{
"field": field,
"operation": operation,
"old_value": old_value,
"new_value": new_value,
},
)
continue
# regular fields
if isinstance(value, (list, tuple)) and len(value) == 2: # ruff: ignore[magic-value-comparison]
old_value, new_value = value
else:
old_value, new_value = None, str(value)
parsed_changes.append(
{
"field": field,
"operation": "update",
"old_value": None if old_value is None else str(old_value),
"new_value": None if new_value is None else str(new_value),
},
)
# fallback
if not parsed_changes:
parsed_changes.append(
{
"field": "",
"operation": str(log.action),
"old_value": None,
"new_value": None,
},
)
return {
"log_id": log.id,
"model_type": "annotation",
"object_id": int(log.object_id),
"timestamp": log.timestamp,
"actor": actor,
"message": f"{actor or 'System'} updated annotation {log.object_id}",
"changes": parsed_changes,
}
[docs]
def list(self, request):
patient_id = request.query_params.get("patient_id")
if not patient_id:
raise ValidationError(
{"patient_id": "patient_id query parameter is required."},
)
if not patient_id.isdigit():
raise ValidationError(
{"patient_id": "Invalid patient_id. Must be an integer."},
)
try:
patient = Patient.objects.select_related("project").get(pk=patient_id)
except Patient.DoesNotExist as exc:
raise ValidationError(
{"patient_id": "Patient not found."},
) from exc
if not has_permission(
user=request.user,
codename="view_patientannotation_history",
project=patient.project,
):
return Response(
{"error": "No permission to view annotation history."},
status=status.HTTP_403_FORBIDDEN,
)
annotation_ids = list(
PatientAnnotation.objects.filter(patient_id=patient.id).values_list(
"id",
flat=True,
),
)
if not annotation_ids:
return Response([], status=status.HTTP_200_OK)
annotation_ct = ContentType.objects.get_for_model(PatientAnnotation)
logs = (
LogEntry.objects
.filter(
content_type=annotation_ct,
object_id__in=annotation_ids,
)
.select_related("actor")
.order_by("-timestamp", "-id")
)
rows = [self._parse_log_entry(log) for log in logs]
serializer = self.serializer_class(rows, many=True)
return Response(serializer.data, status=status.HTTP_200_OK)