ismail
7369d08012
All checks were successful
Build and Push Docker Image / build (push) Successful in 4m14s
Reference numbers (unified scheme PREFIX-YYYYMM-HOSP-NNNN, e.g. CMP-202606-HHN-0001): - new ReferenceSequence model + generate_reference() helper (apps/core) - Complaint/Inquiry/Observation/Appreciation/Suggestion emit unified refs via save() - prefix-based auto-routing in public track API (CMP/INQ/OBS trackable; APR/SGT internal-only) - removed legacy CMP-/INQ- generators in ui_views, integrations, px_sources - migrations: core.0003_referencesequence, appreciation.0006, feedback.0008, observations.0012 - unit tests (format, sanitization, monthly reset, 40-thread concurrency) QA audit: - isolated E2E hospital sandbox mirroring HH-N + 10 role users (create_e2e_isolated_env) - feedback-modules-audit.spec.ts + audit helper (headed, run-to-completion) - reports/feedback-modules-qa-report.md Also bundles accumulated in-progress work across complaints, observations, organizations, templates, and other modules.
120 lines
3.8 KiB
Python
120 lines
3.8 KiB
Python
import re
|
|
|
|
import pandas as pd
|
|
from django.core.management.base import BaseCommand
|
|
from django.db import transaction
|
|
|
|
from apps.organizations.models import Area, Hospital
|
|
|
|
|
|
LOCATION_TYPE_MAP = {
|
|
"Inpatient": "IP",
|
|
"Outpatient": "OP",
|
|
"Emergency": "ER",
|
|
}
|
|
|
|
AREAS_DATA = [
|
|
("Labor & Delivery", "IP"),
|
|
("NICU - Neonatal Intensive Care Unit", "IP"),
|
|
("Nursery", "IP"),
|
|
("OB Wards", "IP"),
|
|
("OPD1", "OP"),
|
|
("OPD2", "OP"),
|
|
("OPD3", "OP"),
|
|
("OPD4", "OP"),
|
|
("OPD5", "OP"),
|
|
("OPD6", "OP"),
|
|
("OPD7", "OP"),
|
|
("Main OR", "IP"),
|
|
("Recovery", "IP"),
|
|
("Anesthesia", "GENERAL"),
|
|
("Dialysis", "OP"),
|
|
("Endoscopy", "GENERAL"),
|
|
("ICCU", "IP"),
|
|
("ICU", "IP"),
|
|
("PICU", "IP"),
|
|
("ICU Stepdown", "IP"),
|
|
("PICU Stepdown", "IP"),
|
|
("Stepdown 3-4", "IP"),
|
|
("LTACU", "IP"),
|
|
("Surgical Wards", "IP"),
|
|
("Medical Ward", "IP"),
|
|
("Pediatric Ward", "IP"),
|
|
("Emergency", "ER"),
|
|
]
|
|
|
|
|
|
def _slugify(name):
|
|
slug = re.sub(r"[^a-zA-Z0-9]+", "_", name).strip("_").lower()
|
|
return slug
|
|
|
|
|
|
class Command(BaseCommand):
|
|
help = "Import Areas from the LOCATIONS1 sheet for all 3 hospitals"
|
|
|
|
def add_arguments(self, parser):
|
|
parser.add_argument(
|
|
"--file",
|
|
default="data/Final List of Departments - 6th Version.xlsx",
|
|
help="Path to the departments Excel file",
|
|
)
|
|
parser.add_argument(
|
|
"--dry-run",
|
|
action="store_true",
|
|
help="Show what would be created without writing to DB",
|
|
)
|
|
|
|
def handle(self, *args, **options):
|
|
filepath = options["file"]
|
|
dry_run = options["dry_run"]
|
|
|
|
hospitals = Hospital.objects.filter(status="active").order_by("code")
|
|
if not hospitals.exists():
|
|
self.stderr.write(self.style.ERROR("No active hospitals found"))
|
|
return
|
|
|
|
self.stdout.write(f"Found {hospitals.count()} hospitals: {list(hospitals.values_list('code', flat=True))}")
|
|
|
|
created_count = 0
|
|
skipped_count = 0
|
|
|
|
with transaction.atomic():
|
|
for hospital in hospitals:
|
|
self.stdout.write(f"\nHospital: {hospital.code} ({hospital.name})")
|
|
|
|
for name_en, location_type in AREAS_DATA:
|
|
code = _slugify(name_en)
|
|
|
|
if dry_run:
|
|
self.stdout.write(f" [DRY RUN] Would create: {name_en} | code={code} | loc={location_type}")
|
|
created_count += 1
|
|
continue
|
|
|
|
area, created = Area.objects.get_or_create(
|
|
hospital=hospital,
|
|
code=code,
|
|
defaults={
|
|
"name_en": name_en,
|
|
"location_type": location_type,
|
|
"status": "active",
|
|
},
|
|
)
|
|
|
|
if created:
|
|
self.stdout.write(self.style.SUCCESS(f" Created: {name_en} ({location_type})"))
|
|
created_count += 1
|
|
else:
|
|
if area.location_type != location_type:
|
|
area.location_type = location_type
|
|
area.save(update_fields=["location_type"])
|
|
self.stdout.write(f" Updated location_type: {name_en} → {location_type}")
|
|
else:
|
|
skipped_count += 1
|
|
|
|
action = "Would create" if dry_run else "Created"
|
|
self.stdout.write(self.style.SUCCESS(f"\n{action}: {created_count} | Skipped (existing): {skipped_count}"))
|
|
|
|
if not dry_run:
|
|
total = Area.objects.filter(status="active").count()
|
|
self.stdout.write(f"Total active Areas in DB: {total}")
|