""" Management command to simulate HIS data fetch for testing. Replica of the fetch_his_surveys periodic task, but fetches fake data from the generate-visit simulator endpoint instead of the real HIS API. Usage: python manage.py simulate_his_fetch python manage.py simulate_his_fetch --count 10 python manage.py simulate_his_fetch --count 5 --patient-type ED python manage.py simulate_his_fetch --hospital-id 2 python manage.py simulate_his_fetch --dry-run """ import json import requests from django.conf import settings from django.core.management.base import BaseCommand logger = __import__("logging").getLogger(__name__) class Command(BaseCommand): help = "Simulate HIS data fetch using the generate-visit endpoint (replica of fetch_his_surveys task)" def add_arguments(self, parser): parser.add_argument( "--count", type=int, default=1, help="Number of patients to generate and process (default: 1)", ) parser.add_argument( "--patient-type", type=str, default="", choices=["", "ED", "IP", "OP"], help="Patient type to generate (default: random mix)", ) parser.add_argument( "--hospital-id", type=str, default="", help="Hospital ID: '2' (SUWAIDI) or '3' (NUZHA) (default: random)", ) parser.add_argument( "--url", type=str, default="", help="Override the simulator URL (default: http://localhost:8000/api/simulator/generate-visit/)", ) parser.add_argument( "--dry-run", action="store_true", help="Generate data but skip processing through HISAdapter", ) def handle(self, *args, **options): count = options["count"] patient_type = options["patient_type"] hospital_id = options["hospital_id"] dry_run = options["dry_run"] base_url = options["url"] or "http://localhost:8000/api/simulator/generate-visit/" if not base_url.endswith("/"): base_url += "/" self.stdout.write("") self.stdout.write(self.style.WARNING("=" * 70)) self.stdout.write(self.style.WARNING(" HIS FETCH SIMULATOR")) self.stdout.write(self.style.WARNING("=" * 70)) self.stdout.write(f" URL: {base_url}") self.stdout.write(f" Count: {count}") if patient_type: self.stdout.write(f" Patient Type: {patient_type}") if hospital_id: self.stdout.write(f" Hospital ID: {hospital_id}") if dry_run: self.stdout.write(f" Mode: DRY RUN (no processing)") self.stdout.write(self.style.WARNING("=" * 70)) self.stdout.write("") stats = { "total": 0, "success": 0, "failed": 0, "visits_saved": 0, "surveys_created": 0, "surveys_skipped": 0, "errors": [], } for i in range(count): self.stdout.write(f"\n[{i + 1}/{count}] Fetching patient data...") try: params = {} if patient_type: params["patient_type"] = patient_type if hospital_id: params["hospital_id"] = hospital_id response = requests.get(base_url, params=params, timeout=15) response.raise_for_status() his_data = response.json() patient_list = his_data.get("FetchPatientDataTimeStampList", []) if not patient_list: self.stdout.write(self.style.WARNING(" No patient data returned")) continue patient = patient_list[0] patient_name = patient.get("PatientName", "Unknown") pt_type = patient.get("PatientType", "Unknown") hospital_name = patient.get("HospitalName", "Unknown") admit = patient.get("AdmitDate", "") discharge = patient.get("DischargeDate") or "N/A" ed_visits = his_data.get("FetchPatientDataTimeStampVisitEDDataList", []) ip_visits = his_data.get("FetchPatientDataTimeStampVisitIPDataList", []) op_visits = his_data.get("FetchPatientDataTimeStampVisitOPDataList", []) total_visits = len(ed_visits) + len(ip_visits) + len(op_visits) self.stdout.write(f" Patient: {self.style.SQL_FIELD(patient_name)}") self.stdout.write(f" Hospital: {hospital_name} | Type: {pt_type}") self.stdout.write(f" Admit: {admit} | Discharge: {discharge}") self.stdout.write(f" Visit Events: {total_visits}") if dry_run: self.stdout.write(self.style.WARNING(" [DRY RUN] Skipping HISAdapter processing")) stats["total"] += 1 stats["success"] += 1 continue from apps.integrations.services.his_adapter import HISAdapter process_result = HISAdapter.process_his_response(his_data) stats["total"] += 1 if process_result["success"]: stats["success"] += 1 visits = process_result.get("visits_saved", 0) surveys = process_result.get("surveys_created", 0) skipped = process_result.get("surveys_skipped", 0) stats["visits_saved"] += visits stats["surveys_created"] += surveys stats["surveys_skipped"] += skipped self.stdout.write(self.style.SUCCESS(f" Visit Saved: {visits}")) self.stdout.write(self.style.SUCCESS(f" Surveys Created: {surveys}")) if skipped: self.stdout.write(self.style.WARNING(f" Surveys Skipped: {skipped}")) for detail in process_result.get("details", []): reason = detail.get("reason", "") error = detail.get("error", "") if reason: self.stdout.write(f" - {detail['patient_name']}: {reason}") if error: self.stdout.write(self.style.ERROR(f" - {detail['patient_name']}: {error}")) for err in process_result.get("errors", []): stats["errors"].append(err) self.stdout.write(self.style.ERROR(f" Error: {err}")) else: stats["failed"] += 1 for err in process_result.get("errors", []): stats["errors"].append(err) self.stdout.write(self.style.ERROR(f" Error: {err}")) except requests.exceptions.RequestException as e: stats["total"] += 1 stats["failed"] += 1 err = f"HTTP request failed: {e}" stats["errors"].append(err) self.stdout.write(self.style.ERROR(f" {err}")) except Exception as e: stats["total"] += 1 stats["failed"] += 1 err = f"Processing error: {e}" stats["errors"].append(err) self.stdout.write(self.style.ERROR(f" {err}")) self.stdout.write("") self.stdout.write(self.style.WARNING("=" * 70)) self.stdout.write(self.style.WARNING(" SUMMARY")) self.stdout.write(self.style.WARNING("=" * 70)) self.stdout.write(f" Total: {stats['total']}") self.stdout.write(f" Success: {self.style.SUCCESS(str(stats['success']))}") self.stdout.write(f" Failed: {self.style.ERROR(str(stats['failed']))}") self.stdout.write(f" Visits: {stats['visits_saved']}") self.stdout.write(f" Surveys: {self.style.SUCCESS(str(stats['surveys_created']))}") if stats["surveys_skipped"]: self.stdout.write(f" Skipped: {self.style.WARNING(str(stats['surveys_skipped']))}") if stats["errors"]: self.stdout.write("") self.stdout.write(self.style.ERROR(" Errors:")) for err in stats["errors"]: self.stdout.write(self.style.ERROR(f" - {err}")) self.stdout.write(self.style.WARNING("=" * 70)) self.stdout.write("")