# 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 pathlib import Path
import pandas as pd
from django.db import transaction
from django.utils.timezone import make_aware
from nlpmed_portal.annotations.models import ImportJob
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.utils import read_dataframe
[docs]
def run_import_job(import_job_id: int) -> None:
job = ImportJob.objects.get(pk=import_job_id)
file_path = Path(job.import_file.path)
try: # ruff: ignore[too-many-statements-in-try-clause]
import_df = read_dataframe(file_path, job.import_format)
# Rename columns
rename_map = job.column_map or {}
import_df = import_df.rename(columns=rename_map)
# Parse dates
date_fmt_map = job.date_format_map or {}
for col_name, date_fmt in date_fmt_map.items():
if col_name in import_df.columns:
import_df[col_name] = pd.to_datetime(
import_df[col_name],
format=date_fmt,
errors="coerce",
).map(make_aware)
job.job_status = "in_progress"
job.job_progress = 0
job.save()
# Import data
if job.import_target == "patient":
_import_patients(import_df, job)
elif job.import_target == "note":
_import_notes(import_df, job)
elif job.import_target == "lab":
_import_labs(import_df, job)
else:
msg = f"Unknown import_target: {job.import_target}"
raise ValueError(msg) # ruff: ignore[raise-within-try]
job.job_status = "completed"
job.save()
except Exception as exc:
job.refresh_from_db()
job.job_status = "failed"
job.error_message = str(exc)
job.save()
raise
finally:
job.delete_import_file()
def _import_patients(patients_df: pd.DataFrame, job: ImportJob) -> None:
# Convert types (dates are already converted)
patients_df["pat_id"] = patients_df["pat_id"].astype("string")
patients_df = patients_df.drop_duplicates(subset=["pat_id"])
total = len(patients_df)
chunk_size = 500
new_count = 0
pat_id_in_file = patients_df["pat_id"].dropna().unique()
existing_set = set(
Patient.objects.filter(
pat_id__in=pat_id_in_file,
project=job.project,
).values_list("pat_id", flat=True),
)
rows_processed = 0
for start in range(0, total, chunk_size):
end = min(start + chunk_size, total)
subset = patients_df.iloc[start:end]
with transaction.atomic():
batch_new = []
for _, row in subset.iterrows():
pat_id = row.get("pat_id")
index_date = row.get("index_date")
if not pat_id or not index_date:
msg = "Missing pat_id or index_date."
raise ValueError(msg)
if pat_id in existing_set:
continue
batch_new.append(
Patient(
pat_id=pat_id,
index_date=index_date,
project=job.project,
),
)
created_objs = Patient.objects.bulk_create(batch_new)
new_count += len(created_objs)
for obj in created_objs:
existing_set.add(obj.pat_id)
rows_processed += len(subset)
job.job_progress = int(rows_processed / total * 100)
job.new_rows_count = new_count
job.save(update_fields=["job_progress", "new_rows_count"])
job.new_rows_count = new_count
job.save(update_fields=["new_rows_count"])
def _import_notes(notes_df: pd.DataFrame, job: ImportJob) -> None: # ruff: ignore[too-many-locals]
# Convert types (dates are already converted)
notes_df["pat_id"] = notes_df["pat_id"].astype("string").str.strip()
notes_df["note_id"] = notes_df["note_id"].astype("string").str.strip()
notes_df["note_text"] = notes_df["note_text"].astype("string").str.strip()
notes_df["note_type"] = notes_df["note_type"].astype("string").str.strip()
notes_df = notes_df.drop_duplicates(subset=["note_id"])
total = len(notes_df)
chunk_size = 500
new_count = 0
note_id_in_file = notes_df["note_id"].dropna().unique()
existing_set = set(
Note.objects.filter(
note_id__in=note_id_in_file,
patient__project=job.project,
).values_list("note_id", flat=True),
)
rows_processed = 0
for start in range(0, total, chunk_size):
end = min(start + chunk_size, total)
subset = notes_df.iloc[start:end]
with transaction.atomic():
batch_new = []
for _, row in subset.iterrows():
note_id = row.get("note_id")
note_text = row.get("note_text")
note_date = row.get("note_date")
note_type = row.get("note_type")
pat_id = row.get("pat_id")
index_date = row.get("index_date")
if not pat_id or not note_id:
msg = "Missing pat_id or note_id."
raise ValueError(msg)
if note_id in existing_set:
continue
patient_obj = Patient.objects.filter(
pat_id=pat_id,
project=job.project,
).first()
if patient_obj is None:
if not job.import_missing_patients:
continue
if not index_date:
msg = "Missing index_date."
raise ValueError(msg)
patient_obj = Patient.objects.create(
pat_id=pat_id,
index_date=index_date,
project=job.project,
)
new_note = Note(
note_id=note_id,
note_text=note_text,
note_date=note_date,
note_type=note_type,
patient=patient_obj,
)
batch_new.append(new_note)
created_notes = Note.objects.bulk_create(batch_new)
new_count += len(created_notes)
for n in created_notes:
existing_set.add(n.note_id)
rows_processed += len(subset)
job.job_progress = int(rows_processed / total * 100)
job.new_rows_count = new_count
job.save(update_fields=["job_progress", "new_rows_count"])
job.new_rows_count = new_count
job.save(update_fields=["new_rows_count"])
def _import_labs(labs_df: pd.DataFrame, job: ImportJob) -> None: # ruff: ignore[too-many-statements, too-many-locals]
# Convert types (dates are already converted)
labs_df["pat_id"] = labs_df["pat_id"].astype("string")
labs_df["lab_id"] = labs_df["lab_id"].astype("string")
labs_df["component"] = labs_df["component"].astype("string")
labs_df["value"] = labs_df["value"].astype(float)
labs_df["unit"] = labs_df["unit"].astype("string")
labs_df["abnormal"] = labs_df["abnormal"].astype(bool)
labs_df = labs_df.drop_duplicates(subset=["lab_id"])
total = len(labs_df)
chunk_size = 500
new_count = 0
lab_id_in_file = labs_df["lab_id"].dropna().unique()
labs_df["abnormal"] = labs_df["abnormal"].astype(str).str.strip().str.lower()
labs_df["abnormal"] = ~labs_df["abnormal"].isin(
[None, "0", "no", "false", ""],
)
existing_set = set(
Lab.objects.filter(
lab_id__in=lab_id_in_file,
patient__project=job.project,
).values_list("lab_id", flat=True),
)
rows_processed = 0
for start in range(0, total, chunk_size):
end = min(start + chunk_size, total)
subset = labs_df.iloc[start:end]
with transaction.atomic():
batch_new = []
for _, row in subset.iterrows():
lab_id = row.get("lab_id")
component = row.get("component")
collection_datetime = row.get("collection_datetime")
value = row.get("value")
unit = row.get("unit")
abnormal = row.get("abnormal", False)
pat_id = row.get("pat_id")
index_date = row.get("index_date")
if not pat_id or not lab_id:
msg = "Missing pat_id or lab_id."
raise ValueError(msg)
if lab_id in existing_set:
continue
patient_obj = Patient.objects.filter(
pat_id=pat_id,
project=job.project,
).first()
if patient_obj is None:
if not job.import_missing_patients:
continue
if not index_date:
msg = "Missing index_date."
raise ValueError(msg)
patient_obj = Patient.objects.create(
pat_id=pat_id,
index_date=index_date,
project=job.project,
)
new_lab = Lab(
lab_id=lab_id,
component=component,
collection_datetime=collection_datetime,
value=value,
unit=unit,
abnormal=abnormal,
patient=patient_obj,
)
batch_new.append(new_lab)
created_labs = Lab.objects.bulk_create(batch_new)
new_count += len(created_labs)
for c_lab in created_labs:
existing_set.add(c_lab.lab_id)
rows_processed += len(subset)
job.job_progress = int(rows_processed / total * 100)
job.new_rows_count = new_count
job.save(update_fields=["job_progress", "new_rows_count"])
job.new_rows_count = new_count
job.save(update_fields=["new_rows_count"])