204 lines
8.3 KiB
Python
204 lines
8.3 KiB
Python
"""
|
|
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("")
|