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.
334 lines
12 KiB
Python
334 lines
12 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 Department, Hospital, Section, Staff
|
|
|
|
|
|
CATEGORY_NORMALIZE = {
|
|
"Medical": "medical",
|
|
"Medical ": "medical",
|
|
"Admintrative": "administrative",
|
|
"Adminstrative": "administrative",
|
|
"Administrative": "administrative",
|
|
"Nursing": "nursing",
|
|
"Support Services": "support_services",
|
|
"Non-Medical": "non_medical",
|
|
}
|
|
|
|
LOCATION_TYPE_MAP = {
|
|
"OP": "OP",
|
|
"IP": "IP",
|
|
"ER": "ER",
|
|
"GO": "GENERAL",
|
|
}
|
|
|
|
ROLE_COLUMNS_DEPT = [
|
|
("manager_3rd", "3rd Manager"),
|
|
("manager_2nd", "2nd Manager"),
|
|
("deputy_manager", "Deputy Manger"),
|
|
("champion", "Champion"),
|
|
]
|
|
|
|
ROLE_COLUMNS_SECTION = [
|
|
("supervisor", "Supervisor\nHead Nurse\nHOD In Charge"),
|
|
("deputy_supervisor", "Deputy Supervisor\nDeputy Head Nurse"),
|
|
]
|
|
|
|
HOSPITAL_LOCATION_MAP = {
|
|
"HH-S": "السويدي",
|
|
"HH-N": "النزهة",
|
|
"HH-A": "العليا",
|
|
}
|
|
|
|
|
|
def _make_dept_code(hospital_code, dept_name):
|
|
prefix = hospital_code.lower().replace("-", "_")
|
|
slug = re.sub(r"[^a-zA-Z0-9]", "_", dept_name).strip("_").lower()[:80]
|
|
return f"{prefix}_{slug}"
|
|
|
|
|
|
def _make_sec_code(hospital_code, dept_name, sec_name):
|
|
prefix = hospital_code.lower().replace("-", "_")
|
|
dept_slug = re.sub(r"[^a-zA-Z0-9]", "_", dept_name).strip("_").lower()[:40]
|
|
sec_slug = re.sub(r"[^a-zA-Z0-9]", "_", sec_name).strip("_").lower()[:40]
|
|
return f"{prefix}_{dept_slug}__{sec_slug}"
|
|
|
|
|
|
def parse_employee_id(val):
|
|
if not val or not isinstance(val, str):
|
|
return None
|
|
val = val.strip()
|
|
if val == "?" or val == "":
|
|
return None
|
|
m = re.match(r"^(\d+)\s*-", val)
|
|
if m:
|
|
return m.group(1)
|
|
return None
|
|
|
|
|
|
def _build_eid_location_map():
|
|
from apps.organizations.models import Staff as S
|
|
|
|
return {s.employee_id: s.hospital.code for s in S.objects.select_related("hospital").all()}
|
|
|
|
|
|
def _find_staff_at_hospital(eid, hospital_code):
|
|
if not eid:
|
|
return None
|
|
try:
|
|
return Staff.objects.select_related("hospital").get(employee_id=eid, hospital__code=hospital_code)
|
|
except Staff.DoesNotExist:
|
|
return None
|
|
|
|
|
|
def _find_staff_at_hospital_fuzzy(name, hospital_code):
|
|
if not name:
|
|
return None
|
|
clean = re.sub(r"\s+", " ", name.strip()).upper()
|
|
for staff in Staff.objects.filter(hospital__code=hospital_code):
|
|
full = re.sub(r"\s+", " ", staff.name.strip()).upper() if staff.name else ""
|
|
if clean in full or full in clean:
|
|
return staff
|
|
return None
|
|
|
|
|
|
class Command(BaseCommand):
|
|
help = "Import departments and sections from PX360 Department Breakdown List Excel (multi-hospital)"
|
|
|
|
def add_arguments(self, parser):
|
|
parser.add_argument("--file", default="Documents/PX360 - Department Breakdown List.xlsx")
|
|
parser.add_argument("--roles", action="store_true", help="Backfill role FKs (requires staff already imported)")
|
|
parser.add_argument("--hospital-code", required=True, help="Hospital code (HH-S, HH-N, HH-A)")
|
|
parser.add_argument("--all-hospitals", action="store_true", help="Import for all hospitals from breakdown list")
|
|
|
|
def handle(self, *args, **options):
|
|
filepath = options["file"]
|
|
do_roles = options["roles"]
|
|
hospital_code = options["hospital_code"]
|
|
|
|
df = pd.read_excel(filepath, header=0)
|
|
df["Main Section"] = df["Main Section"].fillna("").str.strip()
|
|
|
|
if options["all_hospitals"]:
|
|
for code in ["HH-S", "HH-N"]:
|
|
try:
|
|
hospital = Hospital.objects.get(code=code)
|
|
except Hospital.DoesNotExist:
|
|
self.stderr.write(f"Hospital {code} not found, skipping")
|
|
continue
|
|
if do_roles:
|
|
self._backfill_roles(df, code)
|
|
else:
|
|
self._import_departments_and_sections(df, hospital, code)
|
|
else:
|
|
try:
|
|
hospital = Hospital.objects.get(code=hospital_code)
|
|
except Hospital.DoesNotExist:
|
|
self.stderr.write(f"Hospital {hospital_code} not found. Create it first.")
|
|
return
|
|
|
|
if do_roles:
|
|
self._backfill_roles(df, hospital_code)
|
|
else:
|
|
self._import_departments_and_sections(df, hospital, hospital_code)
|
|
|
|
@transaction.atomic
|
|
def _import_departments_and_sections(self, df, hospital, hospital_code):
|
|
df["Main Section"] = df["Main Section"].fillna("").str.strip()
|
|
|
|
dept_names = df["Department Name"].dropna().unique()
|
|
self.stdout.write(f"[{hospital_code}] Found {len(dept_names)} departments in Excel")
|
|
|
|
dept_count = 0
|
|
sec_count = 0
|
|
|
|
for dept_name in sorted(dept_names):
|
|
dept_rows = df[df["Department Name"] == dept_name]
|
|
first_row = dept_rows.iloc[0]
|
|
|
|
cat_raw = first_row["Main Section"]
|
|
category = CATEGORY_NORMALIZE.get(cat_raw, "")
|
|
|
|
code = _make_dept_code(hospital_code, dept_name)
|
|
|
|
dept, created = Department.objects.update_or_create(
|
|
hospital=hospital,
|
|
code=code,
|
|
defaults={
|
|
"name": dept_name,
|
|
"name_en": dept_name,
|
|
"name_ar": "",
|
|
"category": category,
|
|
"location_type": "",
|
|
"status": "active",
|
|
},
|
|
)
|
|
dept_count += 1
|
|
action = "Created" if created else "Updated"
|
|
self.stdout.write(f" {action} dept: {dept_name} [{category}]")
|
|
|
|
unique_sections = dept_rows.dropna(subset=["Section"])
|
|
seen = set()
|
|
for _, sec_row in unique_sections.iterrows():
|
|
sec_name = str(sec_row["Section"]).strip()
|
|
if not sec_name or sec_name in seen:
|
|
continue
|
|
seen.add(sec_name)
|
|
|
|
loc_type_raw = str(sec_row.get("LOCATION TYPE", "")).strip()
|
|
loc_type = LOCATION_TYPE_MAP.get(loc_type_raw, "")
|
|
|
|
area = str(sec_row.get("AREA", "")).strip() if pd.notna(sec_row.get("AREA")) else ""
|
|
zone = str(sec_row.get("ZONE", "")).strip() if pd.notna(sec_row.get("ZONE")) else ""
|
|
floor = str(sec_row.get("FLOOR", "")).strip() if pd.notna(sec_row.get("FLOOR")) else ""
|
|
|
|
sub_location = f"{area}/{zone}".strip("/") if area or zone else ""
|
|
|
|
display_en = str(sec_row.get("Name (EN)", "")).strip() if pd.notna(sec_row.get("Name (EN)")) else ""
|
|
display_ar = str(sec_row.get("Name (AR)", "")).strip() if pd.notna(sec_row.get("Name (AR)")) else ""
|
|
if display_en == "nan":
|
|
display_en = ""
|
|
if display_ar == "nan":
|
|
display_ar = ""
|
|
|
|
sec_code = _make_sec_code(hospital_code, dept_name, sec_name)
|
|
|
|
Section.objects.update_or_create(
|
|
department=dept,
|
|
code=sec_code,
|
|
defaults={
|
|
"name_en": sec_name,
|
|
"name_ar": "",
|
|
"location_type": loc_type,
|
|
"sub_location": sub_location,
|
|
"floor": floor,
|
|
"display_name_en": display_en,
|
|
"display_name_ar": display_ar,
|
|
"status": "active",
|
|
},
|
|
)
|
|
sec_count += 1
|
|
|
|
self.stdout.write(self.style.SUCCESS(f"[{hospital_code}] Done: {dept_count} departments, {sec_count} sections"))
|
|
|
|
@transaction.atomic
|
|
def _backfill_roles(self, df, hospital_code):
|
|
self.stdout.write(f"[{hospital_code}] Backfilling roles...")
|
|
staff_count = Staff.objects.filter(hospital__code=hospital_code).count()
|
|
self.stdout.write(f" Staff at {hospital_code}: {staff_count}")
|
|
if staff_count == 0:
|
|
self.stderr.write(f" No staff at {hospital_code}, skipping")
|
|
return
|
|
|
|
target_location = HOSPITAL_LOCATION_MAP.get(hospital_code)
|
|
|
|
eid_loc_map = _build_eid_location_map()
|
|
|
|
dept_names = df["Department Name"].dropna().unique()
|
|
dept_updated = 0
|
|
sec_updated = 0
|
|
unmatched = []
|
|
|
|
for dept_name in sorted(dept_names):
|
|
dept_rows = df[df["Department Name"] == dept_name]
|
|
|
|
code = _make_dept_code(hospital_code, dept_name)
|
|
try:
|
|
dept = Department.objects.get(code=code)
|
|
except Department.DoesNotExist:
|
|
continue
|
|
|
|
# Collect all unique role values across all rows for this dept
|
|
for field_name, col_name in ROLE_COLUMNS_DEPT:
|
|
all_vals = dept_rows[col_name].dropna().unique()
|
|
staff = None
|
|
for raw_val in all_vals:
|
|
raw_val = str(raw_val).strip()
|
|
if raw_val == "?" or not raw_val:
|
|
continue
|
|
eid = parse_employee_id(raw_val)
|
|
if eid:
|
|
loc = eid_loc_map.get(eid)
|
|
if loc == target_location:
|
|
staff = _find_staff_at_hospital(eid, hospital_code)
|
|
if staff:
|
|
break
|
|
if not staff:
|
|
for raw_val in all_vals:
|
|
raw_val = str(raw_val).strip()
|
|
if raw_val == "?" or not raw_val:
|
|
continue
|
|
eid = parse_employee_id(raw_val)
|
|
if eid:
|
|
staff = _find_staff_at_hospital(eid, hospital_code)
|
|
if staff:
|
|
break
|
|
if not staff and len(all_vals) > 0:
|
|
staff = _find_staff_at_hospital_fuzzy(str(all_vals[0]), hospital_code)
|
|
|
|
if staff:
|
|
setattr(dept, field_name, staff)
|
|
else:
|
|
for raw_val in all_vals[:1]:
|
|
unmatched.append(f"DEPT {hospital_code}/{dept_name} {field_name}: {raw_val}")
|
|
|
|
dept.save()
|
|
dept_updated += 1
|
|
|
|
# Section-level roles
|
|
unique_sections = dept_rows.dropna(subset=["Section"])
|
|
seen = set()
|
|
for _, sec_row in unique_sections.iterrows():
|
|
sec_name = str(sec_row["Section"]).strip()
|
|
if not sec_name or sec_name in seen:
|
|
continue
|
|
seen.add(sec_name)
|
|
|
|
sec_code = _make_sec_code(hospital_code, dept_name, sec_name)
|
|
try:
|
|
section = Section.objects.get(code=sec_code)
|
|
except Section.DoesNotExist:
|
|
continue
|
|
|
|
changed = False
|
|
for field_name, col_name in ROLE_COLUMNS_SECTION:
|
|
raw_val = sec_row.get(col_name)
|
|
if pd.isna(raw_val):
|
|
continue
|
|
raw_val = str(raw_val).strip()
|
|
if raw_val == "?" or not raw_val:
|
|
continue
|
|
|
|
eid = parse_employee_id(raw_val)
|
|
staff = None
|
|
if eid:
|
|
loc = eid_loc_map.get(eid)
|
|
if loc == target_location:
|
|
staff = _find_staff_at_hospital(eid, hospital_code)
|
|
if not staff and eid:
|
|
staff = _find_staff_at_hospital(eid, hospital_code)
|
|
if not staff:
|
|
staff = _find_staff_at_hospital_fuzzy(raw_val, hospital_code)
|
|
|
|
if staff:
|
|
setattr(section, field_name, staff)
|
|
changed = True
|
|
else:
|
|
unmatched.append(f"SEC {hospital_code}/{dept_name}/{sec_name} {field_name}: {raw_val}")
|
|
|
|
if changed:
|
|
section.save()
|
|
sec_updated += 1
|
|
|
|
self.stdout.write(
|
|
self.style.SUCCESS(f"[{hospital_code}] Done: {dept_updated} depts, {sec_updated} sections updated with roles")
|
|
)
|
|
|
|
if unmatched:
|
|
self.stdout.write(self.style.WARNING(f"\n[{hospital_code}] {len(unmatched)} unmatched roles:"))
|
|
for u in unmatched[:50]:
|
|
self.stdout.write(f" {u}")
|