# 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.core.cache import cache
from django.db.models import F
from django.db.models.functions import Greatest
from django.db.models.signals import post_delete
from django.db.models.signals import post_save
from django.db.models.signals import pre_save
from django.dispatch import receiver
from django.utils import timezone
from nlpmed_portal.annotations.constants import PROJECT_META_CACHE_KEY
from nlpmed_portal.annotations.models import Adjudication
from nlpmed_portal.annotations.models import ImportJob
from nlpmed_portal.annotations.models import Lab
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 Sentence
from nlpmed_portal.annotations.models import Task
def _done_user_ids(patient_id: int, task_type: str) -> set[int]:
if task_type == Task.CLASSIFICATION:
return set(
PatientAnnotation.objects
.filter(patient_id=patient_id)
.values_list("annotator_id", flat=True)
.distinct(),
)
if task_type == Task.NER:
return set(
NERAnnotation.objects
.filter(patient_id=patient_id)
.values_list("annotator_id", flat=True)
.distinct(),
)
if task_type == Task.ADJUDICATION:
return set(
Adjudication.objects
.filter(patient_id=patient_id)
.values_list("adjudicator_id", flat=True)
.distinct(),
)
return set()
def _required_count(patient: Patient, task_type: str) -> int | None:
if task_type in {Task.CLASSIFICATION, Task.NER}:
return int(patient.project.required_patient_annotations or 0)
if task_type == Task.ADJUDICATION:
return int(patient.project.required_patient_adjudications or 0)
return None
[docs]
def recompute_tasks_for_patient(patient_id: int, task_type: str) -> None:
patient = Patient.objects.select_related("project").only("id", "project_id").get(pk=patient_id)
done_users = _done_user_ids(patient_id, task_type)
required = _required_count(patient, task_type)
now = timezone.now()
qs = Task.objects.filter(patient_id=patient_id, task_type=task_type)
# set to completed for users who have done annotation/adjudication
if done_users:
qs.filter(assignee_id__in=done_users).exclude(status="completed").update(
status="completed",
updated_at=now,
)
# set to closed/pending for other users depending on project requirements
remaining = qs.exclude(assignee_id__in=done_users)
requirement_met = (required is not None) and (len(done_users) >= required)
if requirement_met:
remaining.exclude(status="closed").update(status="closed", updated_at=now)
else:
remaining.exclude(status="pending").update(
status="pending",
updated_at=now,
)
@receiver([post_save, post_delete], sender=PatientAnnotation)
def _tasks_on_patient_annotation_change(sender, instance, **kwargs):
recompute_tasks_for_patient(instance.patient_id, Task.CLASSIFICATION)
@receiver([post_save, post_delete], sender=NERAnnotation)
def _tasks_on_ner_annotation_change(sender, instance, **kwargs):
recompute_tasks_for_patient(instance.patient_id, Task.NER)
@receiver([post_save, post_delete], sender=Adjudication)
def _tasks_on_adjudication_change(sender, instance, **kwargs):
recompute_tasks_for_patient(instance.patient_id, Task.ADJUDICATION)
@receiver(post_save, sender=Task)
def _tasks_on_task_assignment(sender, instance, created, **kwargs):
recompute_tasks_for_patient(instance.patient_id, instance.task_type)
[docs]
@receiver(pre_save, sender=Sentence)
def attach_old_flags(sender, instance, **_):
if instance.pk:
old = sender.objects.only("is_important", "is_duplicate").get(pk=instance.pk)
instance._old_is_important = old.is_important # ruff: ignore[private-member-access]
instance._old_is_duplicate = old.is_duplicate # ruff: ignore[private-member-access]
[docs]
@receiver(post_save, sender=Sentence)
def update_cnt_on_save(sender, instance, created, **_):
note_qs = Note.objects.filter(pk=instance.section.note_id)
# New sentence
if created:
if instance.is_important:
note_qs.update(important_sentences_count=F("important_sentences_count") + 1)
if instance.is_duplicate:
note_qs.update(duplicate_sentences_count=F("duplicate_sentences_count") + 1)
# Flags got updated
else:
if getattr(instance, "_old_is_important", False) != instance.is_important:
delta = 1 if instance.is_important else -1
note_qs.update(
important_sentences_count=Greatest(
F("important_sentences_count") + delta,
0,
),
)
if getattr(instance, "_old_is_duplicate", False) != instance.is_duplicate:
delta = 1 if instance.is_duplicate else -1
note_qs.update(
duplicate_sentences_count=Greatest(
F("duplicate_sentences_count") + delta,
0,
),
)
[docs]
@receiver(post_delete, sender=Sentence)
def update_cnt_on_delete(sender, instance, **kwargs):
note_qs = Note.objects.filter(pk=instance.section.note_id)
if instance.is_important:
note_qs.update(
important_sentences_count=Greatest(F("important_sentences_count") - 1, 0),
)
if instance.is_duplicate:
note_qs.update(
duplicate_sentences_count=Greatest(F("duplicate_sentences_count") - 1, 0),
)
[docs]
@receiver(post_delete, sender=ImportJob)
def delete_import_file_on_job_delete(sender, instance, **kwargs):
"""Delete abandoned uploads when their import-job record is deleted."""
instance.delete_import_file(save=False)