""" Management command to load visit_data.json into HISTestPatient and HISTestVisit tables. Usage: python manage.py load_his_test_data [--clear] [--file path/to/visit_data.json] Options: --clear Clear existing test data before loading --file Path to JSON file (default: visit_data.json in project root) """ import json from datetime import datetime from pathlib import Path from django.core.management.base import BaseCommand from django.utils import timezone class Command(BaseCommand): help = "Load visit_data.json into test tables for HIS integration testing" def add_arguments(self, parser): parser.add_argument( "--clear", action="store_true", help="Clear existing test data before loading", ) parser.add_argument( "--file", type=str, default="visit_data.json", help="Path to JSON file (default: visit_data.json in project root)", ) def _parse_date(self, date_str): if not date_str: return None try: naive = datetime.strptime(date_str, "%d-%b-%Y %H:%M") return timezone.make_aware(naive) except ValueError: return None def handle(self, *args, **options): clear = options["clear"] file_path = options["file"] json_path = Path(file_path) if not json_path.is_absolute(): json_path = Path.cwd() / json_path if not json_path.exists(): self.stderr.write(self.style.ERROR(f"File not found: {json_path}")) return self.stdout.write(self.style.MIGRATE_HEADING("=" * 60)) self.stdout.write(self.style.MIGRATE_HEADING("Load HIS Test Data")) self.stdout.write(self.style.MIGRATE_HEADING("=" * 60)) self.stdout.write(f"Reading {json_path}...") with open(json_path, "r") as f: data = json.load(f) patient_list = data.get("FetchPatientDataTimeStampList", []) ed_visits = data.get("FetchPatientDataTimeStampVisitEDDataList", []) ip_visits = data.get("FetchPatientDataTimeStampVisitIPDataList", []) op_visits = data.get("FetchPatientDataTimeStampVisitOPDataList", []) self.stdout.write(f" Patients: {len(patient_list)}") self.stdout.write(f" ED visits: {len(ed_visits)}") self.stdout.write(f" IP visits: {len(ip_visits)}") self.stdout.write(f" OP visits: {len(op_visits)}") if clear: from apps.integrations.models import HISTestPatient, HISTestVisit pv, _ = HISTestPatient.objects.all().delete() vv, _ = HISTestVisit.objects.all().delete() self.stdout.write(f"Cleared {pv} patients, {vv} visits") self._load_patients(patient_list) self._load_visits(ed_visits, "ED") self._load_visits(ip_visits, "IP") self._load_visits(op_visits, "OP") self._ensure_test_config() from apps.integrations.models import HISTestPatient, HISTestVisit self.stdout.write(self.style.MIGRATE_HEADING("\nSummary")) self.stdout.write(f" Patients in DB: {HISTestPatient.objects.count()}") self.stdout.write(f" Visits in DB: {HISTestVisit.objects.count()}") self.stdout.write(self.style.SUCCESS("\nDone!")) def _ensure_test_config(self): from apps.integrations.models import IntegrationConfig, SourceSystem name = "HIS Test" config, created = IntegrationConfig.objects.get_or_create( name=name, source_system=SourceSystem.HIS, defaults={ "api_url": "http://127.0.0.1:8000/api/integrations/test-his-data/", "is_active": False, "description": "Local test endpoint for HIS integration testing. Enable before using fetch_his_surveys.", }, ) if created: self.stdout.write(self.style.SUCCESS(f"\nCreated IntegrationConfig '{name}' (disabled by default)")) else: self.stdout.write(f"\nIntegrationConfig '{name}' already exists") self.stdout.write( self.style.WARNING( f"\nTo use: enable '{name}' config, then:\n" f' python manage.py fetch_his_surveys --config "{name}" ' f'--from-date "01-Jan-2026 00:00:00" --to-date "01-Jan-2026 02:00:00"' ) ) def _load_patients(self, patient_list): from apps.integrations.models import HISTestPatient self.stdout.write(f"\nLoading {len(patient_list)} patients...") batch = [] skipped = 0 existing_ids = set(HISTestPatient.objects.values_list("admission_id", flat=True)) for p in patient_list: admission_id = p.get("AdmissionID", "") if not admission_id or admission_id in existing_ids: skipped += 1 continue batch.append( HISTestPatient( admission_id=admission_id, patient_id=str(p.get("PatientID", "")), patient_type=p.get("PatientType", ""), reg_code=p.get("RegCode", ""), ssn=p.get("SSN", ""), mobile_no=p.get("MobileNo", ""), admit_date=self._parse_date(p.get("AdmitDate")), discharge_date=self._parse_date(p.get("DischargeDate")), patient_data=p, hospital_id=str(p.get("HospitalID", "")), hospital_name=p.get("HospitalName", ""), patient_name=p.get("PatientName", ""), ) ) if len(batch) >= 1000: HISTestPatient.objects.bulk_create(batch, batch_size=1000) self.stdout.write(f" ...saved {len(batch)} patients") batch = [] if batch: HISTestPatient.objects.bulk_create(batch, batch_size=1000) self.stdout.write(f" ...saved {len(batch)} patients") if skipped: self.stdout.write(f" Skipped {skipped} duplicate admission_ids") def _load_visits(self, visit_list, category): from apps.integrations.models import HISTestVisit self.stdout.write(f"\nLoading {len(visit_list)} {category} visits...") batch = [] for v in visit_list: batch.append( HISTestVisit( admission_id=str(v.get("AdmissionID", "")), patient_id=str(v.get("PatientID", "")), visit_category=category, event_type=v.get("Type", ""), bill_date=self._parse_date(v.get("BillDate")), reg_code=v.get("RegCode", ""), ssn=v.get("SSN", ""), mobile_no=v.get("MobileNo", ""), visit_data=v, ) ) if len(batch) >= 2000: HISTestVisit.objects.bulk_create(batch, batch_size=2000) self.stdout.write(f" ...saved {len(batch)} {category} visits") batch = [] if batch: HISTestVisit.objects.bulk_create(batch, batch_size=2000) self.stdout.write(f" ...saved {len(batch)} {category} visits")