# 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 httpx
from django.conf import settings
from django.db.models import Count
from django.db.models import IntegerField
from django.db.models import OuterRef
from django.db.models import QuerySet
from django.db.models import Subquery
from django.db.models.functions import Coalesce
from nlpmed_portal.annotations.models import Note
from nlpmed_portal.annotations.models import Section
from nlpmed_portal.annotations.models import Sentence
from nlpmed_portal.nlp.api.serializers import NLPSettingSerializer
from nlpmed_portal.nlp.models import NLPSetting
[docs]
def process_patient_api_call(
nlp_setting: NLPSetting,
patient_pk: int,
notes: QuerySet[Note],
) -> dict:
transport = httpx.HTTPTransport(retries=3)
patient_json = {
"patient_id": str(patient_pk),
"notes": [{"text": n.note_text} for n in notes],
}
with httpx.Client(transport=transport, verify=False) as client: # ruff: ignore[request-with-no-cert-validation]
try:
response = client.post(
f"{settings.NLP_API_URL}/process_patient",
json={
"patient": patient_json,
"config": NLPSettingSerializer(nlp_setting).data,
},
timeout=300,
)
response.raise_for_status()
return response.json()
except Exception as e:
msg = f"Error processing patient {patient_pk}: {e}"
raise RuntimeError(msg) from e
[docs]
def process_patient_write_db(
nlp_setting: NLPSetting,
response_json: dict,
notes: QuerySet[Note],
) -> None:
processed_notes = response_json.get("notes", [])
sections_to_create = []
sentences_to_create = []
touched_note_ids: set[int] = set()
for processed_note, original_note in zip(processed_notes, notes, strict=False):
touched_note_ids.add(original_note.pk)
original_note.note_sections.all().delete()
original_note.preprocessed_text = processed_note.get("preprocessed_text") or ""
if nlp_setting.ml_inference.status.lower() == "enabled":
original_note.predicted_label = processed_note.get("predicted_label") or "none"
else:
original_note.predicted_label = processed_note.get("predicted_label") or ""
original_note.predicted_score = processed_note.get("predicted_score")
original_note.note_text = processed_note.get("text", original_note.note_text)
original_note.save(
update_fields=[
"preprocessed_text",
"predicted_label",
"predicted_score",
"note_text",
],
)
for section in processed_note.get("sections", []):
if not section["important_indices"] and not section["duplicate_indices"]:
continue
new_sec = Section(
text=section["text"],
start_index=section["start_index"],
end_index=section["end_index"],
note=original_note,
)
sections_to_create.append(new_sec)
sentences_to_create.extend(
[
Sentence(
text=sentence["text"],
start_index=sentence["start_index"],
end_index=sentence["end_index"],
is_important=sentence["is_important"],
is_duplicate=sentence["is_duplicate"],
section=new_sec,
)
for sentence in section.get("sentences", [])
if sentence["is_important"] or sentence["is_duplicate"]
],
)
if sections_to_create:
Section.objects.bulk_create(sections_to_create)
if sentences_to_create:
Sentence.objects.bulk_create(sentences_to_create)
if touched_note_ids:
imp_sq = (
Sentence.objects
.filter(section__note_id=OuterRef("pk"), is_important=True)
.values("section__note_id")
.annotate(c=Count("*"))
.values("c")[:1]
)
dup_sq = (
Sentence.objects
.filter(section__note_id=OuterRef("pk"), is_duplicate=True)
.values("section__note_id")
.annotate(c=Count("*"))
.values("c")[:1]
)
Note.objects.filter(pk__in=touched_note_ids).update(
important_sentences_count=Coalesce(
Subquery(imp_sq, output_field=IntegerField()),
0,
),
duplicate_sentences_count=Coalesce(
Subquery(dup_sq, output_field=IntegerField()),
0,
),
)