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.
260 lines
10 KiB
Python
260 lines
10 KiB
Python
import re
|
|
import uuid
|
|
|
|
import pandas as pd
|
|
from django.core.management.base import BaseCommand
|
|
from django.db import transaction
|
|
|
|
from apps.organizations.models import (
|
|
Department,
|
|
Hospital,
|
|
LegacyHierarchyMapping,
|
|
Section,
|
|
OrgSubSection,
|
|
Staff,
|
|
)
|
|
|
|
|
|
CATEGORY_MAP = {
|
|
"medical": "medical",
|
|
"adminstrative": "administrative",
|
|
"admintrative": "administrative",
|
|
"nursing": "nursing",
|
|
"support services": "support_services",
|
|
}
|
|
|
|
|
|
def _clean(val):
|
|
if pd.isna(val):
|
|
return ""
|
|
return str(val).strip()
|
|
|
|
|
|
def _clean_multiline(val):
|
|
text = _clean(val)
|
|
text = text.replace("\n", " ").replace("\r", " ")
|
|
text = re.sub(r"\s+", " ", text)
|
|
return text.strip()
|
|
|
|
|
|
def _extract_employee_id(name_str):
|
|
if not name_str:
|
|
return None
|
|
match = re.match(r"(\d+)\s*-", name_str)
|
|
if match:
|
|
return match.group(1)
|
|
return None
|
|
|
|
|
|
def _resolve_staff(name_str):
|
|
if not name_str:
|
|
return None
|
|
emp_id = _extract_employee_id(name_str)
|
|
if emp_id:
|
|
staff = Staff.objects.filter(employee_id=emp_id).first()
|
|
if staff:
|
|
return staff
|
|
return None
|
|
|
|
|
|
def _normalize_category(raw):
|
|
if not raw:
|
|
return ""
|
|
return CATEGORY_MAP.get(raw.lower(), raw.lower())
|
|
|
|
|
|
class Command(BaseCommand):
|
|
help = "Import departments, sections, and sub-sections from the Final List of Departments Excel file (4th Version)"
|
|
|
|
def add_arguments(self, parser):
|
|
parser.add_argument(
|
|
"--file",
|
|
type=str,
|
|
default="data/Final List of Departments - 4th Version.xlsx",
|
|
help="Path to the Excel file",
|
|
)
|
|
parser.add_argument(
|
|
"--hospital-code",
|
|
type=str,
|
|
default="HH-NZ",
|
|
help="Hospital code to assign departments to",
|
|
)
|
|
parser.add_argument(
|
|
"--dry-run",
|
|
action="store_true",
|
|
help="Show what would be created without creating it",
|
|
)
|
|
|
|
def handle(self, *args, **options):
|
|
file_path = options["file"]
|
|
hospital_code = options["hospital_code"]
|
|
dry_run = options["dry_run"]
|
|
|
|
hospital = Hospital.objects.filter(code=hospital_code).first()
|
|
if not hospital:
|
|
self.stderr.write(self.style.ERROR(f"Hospital with code '{hospital_code}' not found"))
|
|
return
|
|
|
|
self.stdout.write(f"Importing from: {file_path}")
|
|
self.stdout.write(f"Hospital: {hospital.name} ({hospital.code})")
|
|
|
|
df = pd.read_excel(file_path, sheet_name="ALL LIST")
|
|
|
|
stats = {"departments": 0, "sections": 0, "sub_sections": 0, "mappings": 0, "skipped": 0}
|
|
|
|
with transaction.atomic():
|
|
if dry_run:
|
|
self.stdout.write(self.style.WARNING("DRY RUN - no changes will be saved"))
|
|
|
|
dept_cache = {}
|
|
|
|
for idx, row in df.iterrows():
|
|
dept_name = _clean(row.get("Department"))
|
|
if not dept_name:
|
|
stats["skipped"] += 1
|
|
continue
|
|
|
|
section_name = _clean(row.get("Section"))
|
|
subsection_name = _clean(row.get("Sub-Section"))
|
|
area_raw = _clean(row.get("Main Section"))
|
|
|
|
dept_key = dept_name
|
|
if dept_key not in dept_cache:
|
|
code = re.sub(r"[^a-zA-Z0-9]", "_", dept_name.lower())[:100]
|
|
category = _normalize_category(area_raw) if area_raw else ""
|
|
|
|
defaults = {
|
|
"name": dept_name,
|
|
"name_en": dept_name,
|
|
"category": category,
|
|
"main_section": area_raw,
|
|
"location_type": self._normalize_location_type(_clean(row.get("Department\nLocation"))),
|
|
"sub_location": _clean(row.get("Department\nSub-Location")),
|
|
"floor": _clean(row.get("Department\nFloor")),
|
|
"manager_3rd": _resolve_staff(_clean(row.get("3rd Manager"))),
|
|
"manager_2nd": _resolve_staff(_clean_multiline(row.get("2nd Manager"))),
|
|
"deputy_manager": _resolve_staff(_clean(row.get("Deputy Manger"))),
|
|
"supervisor": _resolve_staff(_clean(row.get("Supervisor\nHead Nurse\nHOD In Charge"))),
|
|
"deputy_supervisor": _resolve_staff(
|
|
_clean(row.get("Deputy Supervisor\nDeputy Head Nurse"))
|
|
),
|
|
"champion_email": _clean(row.get("Champion Email")),
|
|
"old_name_en": _clean(row.get("Unnamed: 22")),
|
|
"old_name_ar": _clean(row.get("Unnamed: 23")),
|
|
}
|
|
|
|
if not dry_run:
|
|
existing = Department.objects.filter(
|
|
hospital=hospital, name_en=dept_name
|
|
).first()
|
|
if existing:
|
|
for k, v in defaults.items():
|
|
setattr(existing, k, v)
|
|
existing.save()
|
|
dept_cache[dept_key] = existing
|
|
self.stdout.write(f" ~ Department: {dept_name} (updated)")
|
|
else:
|
|
dept = Department.objects.create(
|
|
hospital=hospital,
|
|
code=code,
|
|
**defaults,
|
|
)
|
|
dept_cache[dept_key] = dept
|
|
stats["departments"] += 1
|
|
self.stdout.write(f" + Department: {dept_name} (code={code})")
|
|
else:
|
|
stats["departments"] += 1
|
|
dept_cache[dept_key] = dept_name
|
|
self.stdout.write(f" [DRY] Department: {dept_name} (code={code}, area={area})")
|
|
|
|
department = dept_cache.get(dept_key)
|
|
if not department:
|
|
continue
|
|
|
|
if section_name and not subsection_name:
|
|
if not dry_run:
|
|
sec_code = f"{department.code}__{re.sub(r'[^a-zA-Z0-9]', '_', section_name.lower())[:100]}"
|
|
defaults = {
|
|
"name_en": section_name,
|
|
"location_type": self._normalize_location_type(
|
|
_clean(row.get("Department\nLocation"))
|
|
),
|
|
"sub_location": _clean(row.get("Department\nSub-Location")),
|
|
"floor": _clean(row.get("Department\nFloor")),
|
|
"status": "active",
|
|
}
|
|
sec, created = Section.objects.update_or_create(
|
|
department=department,
|
|
code=sec_code,
|
|
defaults=defaults,
|
|
)
|
|
if created:
|
|
stats["sections"] += 1
|
|
self.stdout.write(f" + Section: {section_name}")
|
|
else:
|
|
stats["sections"] += 1
|
|
self.stdout.write(f" [DRY] Section: {section_name}")
|
|
|
|
|
|
old_name_en_raw = _clean_multiline(row.get("Unnamed: 22"))
|
|
old_name_ar_raw = _clean_multiline(row.get("Unnamed: 23"))
|
|
old_loc_en = _clean_multiline(row.get("Unnamed: 24"))
|
|
old_loc_ar = _clean_multiline(row.get("Unnamed: 25"))
|
|
old_area_en = _clean_multiline(row.get("Unnamed: 26"))
|
|
old_area_ar = _clean_multiline(row.get("Unnamed: 27"))
|
|
|
|
if old_name_en_raw or old_name_ar_raw:
|
|
old_names = [n.strip() for n in re.split(r'[&\n]', old_name_en_raw) if n.strip()]
|
|
old_locs = [n.strip() for n in re.split(r'[&\n]', old_loc_en) if n.strip()]
|
|
old_areas = [n.strip() for n in re.split(r'[&\n]', old_area_en) if n.strip()]
|
|
|
|
if not old_names:
|
|
old_names = [""]
|
|
|
|
if not dry_run:
|
|
for i, old_sub_name in enumerate(old_names):
|
|
loc = old_locs[i] if i < len(old_locs) else (old_locs[0] if old_locs else "")
|
|
area = old_areas[i] if i < len(old_areas) else (old_areas[0] if old_areas else "")
|
|
LegacyHierarchyMapping.objects.update_or_create(
|
|
old_location_ar="",
|
|
old_main_section_ar=area,
|
|
old_subsection_ar=old_sub_name,
|
|
defaults={
|
|
"old_location_en": loc,
|
|
"old_main_section_en": area,
|
|
"old_subsection_en": old_sub_name,
|
|
"main_section": department,
|
|
},
|
|
)
|
|
stats["mappings"] += 1
|
|
|
|
if dry_run:
|
|
transaction.set_rollback(True)
|
|
self.stdout.write(self.style.WARNING("DRY RUN - all changes rolled back"))
|
|
|
|
self.stdout.write(self.style.SUCCESS(
|
|
f"\nImport complete: {stats['departments']} departments, "
|
|
f"{stats['sections']} sections, {stats['sub_sections']} sub-sections, "
|
|
f"{stats['mappings']} legacy mappings, {stats['skipped']} skipped"
|
|
))
|
|
|
|
def _normalize_location_type(self, raw):
|
|
if not raw:
|
|
return ""
|
|
raw_lower = raw.lower().replace("\n", " ").replace("\r", " ")
|
|
if "op" in raw_lower and "ip" in raw_lower and "er" in raw_lower:
|
|
return "OP,IP,ER"
|
|
if "op" in raw_lower and "ip" in raw_lower:
|
|
return "OP,IP"
|
|
if "op" in raw_lower and "er" in raw_lower:
|
|
return "OP,ER"
|
|
if "general" in raw_lower or "offices" in raw_lower:
|
|
return "GENERAL"
|
|
if "ip" in raw_lower or "inpatient" in raw_lower:
|
|
return "IP"
|
|
if "er" in raw_lower or "emergency" in raw_lower:
|
|
return "ER"
|
|
if "op" in raw_lower or "outpatient" in raw_lower:
|
|
return "OP"
|
|
return raw[:20]
|