463 lines
19 KiB
Python
463 lines
19 KiB
Python
"""
|
|
Import 2025 complaints from Excel with basic fields (no AI, skip missing columns).
|
|
|
|
Dynamically detects header row and column positions per sheet because 2025
|
|
monthly sheets have inconsistent layouts.
|
|
|
|
Usage:
|
|
python manage.py import_2025_complaints_basic "Complaints Report - 2025.xlsx" --sheet="JAN"
|
|
"""
|
|
|
|
import logging
|
|
import re
|
|
from datetime import datetime, timedelta
|
|
from typing import Dict, Optional
|
|
|
|
from django.core.management.base import BaseCommand, CommandError
|
|
from django.db import transaction
|
|
from django.utils import timezone
|
|
|
|
from apps.accounts.models import User
|
|
from apps.complaints.models import Complaint
|
|
from apps.organizations.models import Hospital, LegacyLocation, LegacyMainSection, LegacySubSection
|
|
|
|
from .complaint_source_mapping import resolve_px_source, get_complaint_source_type
|
|
from .arabic_dept_mapping import resolve_section as resolve_dept_and_section
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
DEFAULT_HOSPITAL_CODE = "HH-N"
|
|
|
|
# Header aliases: list of possible names in Excel for each field
|
|
HEADER_ALIASES = {
|
|
"complaint_num": ["رقم الشكوى"],
|
|
"mrn": ["رقم الملف"],
|
|
"source": ["جهة الشكوى"],
|
|
"location_name": ["الموقع"],
|
|
"main_dept_name": ["القسم الرئيس"],
|
|
"sub_dept_name": ["القسم الفرعي"],
|
|
"date_received": ["تاريخ إستلام الشكوى"],
|
|
"data_entry_person": ["المدخل"],
|
|
# Timeline columns
|
|
"form_sent_date": ["إرسال نموذج الشكوى"],
|
|
"activated_date": ["تفعيل الشكوى"],
|
|
"sent_date": ["تم ارسال الشكوى"],
|
|
"first_reminder": ["First Reminder Sent"],
|
|
"second_reminder": ["Second Reminder Sent"],
|
|
"escalated_date": ["Escalated"],
|
|
"closed_date": ["Closed"],
|
|
"resolved_date": ["Resolved"],
|
|
"response_date": ["تاريخ الرد"],
|
|
# Complaint details
|
|
"staff_name": ["اسم الشخص المشتكى عليه - ان وجد", "اسم الشخص المشتكى عليه"],
|
|
"complaint_subject": ["موضوع الشكوى الأساسية"],
|
|
"description_ar": ["الشكوى باختصار (عربي)", "محتوى الشكوى (عربي)"],
|
|
"description_en": ["الشكوى باختصار English", "محتوى الشكوى (English)"],
|
|
"satisfaction": ["Satisfied/Dissatisfied"],
|
|
"rightful_side": ["The Rightful Side"],
|
|
# Rich data
|
|
"delay_reason": ["سبب تأخير القسم بالرد"],
|
|
"closure_delay": ["سبب تأخير اغلاق الشكوى خلال 72 ساعه"],
|
|
"action_taken": ["الاجراء المتخذ من قبل القسم المعني"],
|
|
"action_result": ["نتيجة الاجراء المتخذ بعد التحقيق"],
|
|
"recommendation": ["Recommendation/Action plan"],
|
|
"solutions": ["حلول واقتراحات"],
|
|
}
|
|
|
|
SATISFACTION_MAP = {
|
|
"satisfied": "satisfied",
|
|
"dissatisfied": "dissatisfied",
|
|
"no response": "no_response",
|
|
"no_response": "no_response",
|
|
"neutral": "neutral",
|
|
"escalated": "escalated",
|
|
}
|
|
|
|
MONTH_MAP = {
|
|
"JAN": "01",
|
|
"FEB": "02",
|
|
"MAR": "03",
|
|
"APR": "04",
|
|
"MAY": "05",
|
|
"JUN": "06",
|
|
"JUL": "07",
|
|
"AUG": "08",
|
|
"SEP": "09",
|
|
"OCT": "10",
|
|
"NOV": "11",
|
|
"DEC": "12",
|
|
}
|
|
|
|
|
|
class Command(BaseCommand):
|
|
help = "Import 2025 complaints with basic fields (no taxonomy, no staff linking)"
|
|
|
|
def add_arguments(self, parser):
|
|
parser.add_argument("excel_file", type=str)
|
|
parser.add_argument("--sheet", type=str, default="JAN")
|
|
parser.add_argument("--dry-run", action="store_true")
|
|
parser.add_argument("--hospital-code", type=str, default=DEFAULT_HOSPITAL_CODE)
|
|
|
|
def handle(self, *args, **options):
|
|
self.excel_file = options["excel_file"]
|
|
self.sheet_name = options["sheet"]
|
|
self.dry_run = options["dry_run"]
|
|
self.hospital_code = options["hospital_code"]
|
|
|
|
self.hospital = self._load_hospital()
|
|
if not self.hospital:
|
|
raise CommandError(f'Hospital "{self.hospital_code}" not found')
|
|
|
|
self.stdout.write(f"Using hospital: {self.hospital.name}")
|
|
|
|
try:
|
|
import openpyxl
|
|
|
|
self.wb = openpyxl.load_workbook(self.excel_file, read_only=True, data_only=True)
|
|
except ImportError:
|
|
raise CommandError("openpyxl required: pip install openpyxl")
|
|
|
|
if self.sheet_name not in self.wb.sheetnames:
|
|
available = ", ".join(self.wb.sheetnames)
|
|
raise CommandError(f'Sheet "{self.sheet_name}" not found. Available: {available}')
|
|
|
|
self.ws = self.wb[self.sheet_name]
|
|
self.stdout.write(f"Processing sheet: {self.sheet_name}")
|
|
|
|
# Detect header row and build column mapping
|
|
self.column_map = self._detect_columns()
|
|
if not self.column_map.get("complaint_num"):
|
|
raise CommandError("Could not detect 'رقم الشكوى' header in sheet")
|
|
|
|
self.stdout.write(f"Detected columns: {self.column_map}")
|
|
|
|
self.stats = {"processed": 0, "success": 0, "failed": 0}
|
|
self.errors = []
|
|
self.used_refs = set()
|
|
self.unmapped_arabic_depts = {}
|
|
|
|
self._process_sheet()
|
|
self._print_report()
|
|
|
|
def _load_hospital(self) -> Optional[Hospital]:
|
|
try:
|
|
return Hospital.objects.get(code=self.hospital_code)
|
|
except Hospital.DoesNotExist:
|
|
return None
|
|
|
|
def _detect_columns(self) -> Dict[str, int]:
|
|
"""Scan first 10 rows to find header row and map columns."""
|
|
mapping = {}
|
|
for r in range(1, 11):
|
|
row_values = {}
|
|
for c in range(1, 80):
|
|
val = self.ws.cell(r, c).value
|
|
if val:
|
|
# Keep first occurrence only
|
|
if val not in row_values:
|
|
row_values[val] = c
|
|
# Check if this row contains the complaint number header
|
|
if "رقم الشكوى" in row_values:
|
|
for field, aliases in HEADER_ALIASES.items():
|
|
for alias in aliases:
|
|
if alias in row_values:
|
|
mapping[field] = row_values[alias]
|
|
break
|
|
self.header_row = r
|
|
break
|
|
return mapping
|
|
|
|
def _process_sheet(self):
|
|
row_num = self.header_row + 1
|
|
|
|
for row in self.ws.iter_rows(min_row=self.header_row + 1, max_row=5000, values_only=True):
|
|
try:
|
|
row_data = self._extract_row_data_from_values(row)
|
|
|
|
if not row_data.get("complaint_num"):
|
|
row_num += 1
|
|
continue
|
|
|
|
self.stats["processed"] += 1
|
|
|
|
try:
|
|
ref_num = self._get_unique_reference_number(row_data["complaint_num"])
|
|
except (ValueError, TypeError):
|
|
row_num += 1
|
|
continue
|
|
|
|
px_source = resolve_px_source(row_data.get("source"))
|
|
|
|
date_received = self._parse_datetime(row_data.get("date_received"))
|
|
created_at = date_received or timezone.now()
|
|
if created_at and timezone.is_naive(created_at):
|
|
created_at = timezone.make_aware(created_at)
|
|
|
|
# Parse timeline dates
|
|
form_sent_date = self._parse_datetime(row_data.get("form_sent_date"))
|
|
activated_date = self._parse_datetime(row_data.get("activated_date"))
|
|
sent_date = self._parse_datetime(row_data.get("sent_date"))
|
|
first_reminder = self._parse_datetime(row_data.get("first_reminder"))
|
|
second_reminder = self._parse_datetime(row_data.get("second_reminder"))
|
|
escalated_date = self._parse_datetime(row_data.get("escalated_date"))
|
|
closed_date = self._parse_datetime(row_data.get("closed_date"))
|
|
resolved_date = self._parse_datetime(row_data.get("resolved_date"))
|
|
response_date = self._parse_datetime(row_data.get("response_date"))
|
|
|
|
location = self._resolve_location(row_data.get("location_name"))
|
|
main_section = self._resolve_section(row_data.get("main_dept_name"))
|
|
subsection = self._resolve_subsection(row_data.get("sub_dept_name"))
|
|
|
|
sub_dept_raw = row_data.get("sub_dept_name")
|
|
dept, section_obj = resolve_dept_and_section(sub_dept_raw, self.hospital_code)
|
|
if not dept and sub_dept_raw and isinstance(sub_dept_raw, str) and sub_dept_raw.strip():
|
|
name = sub_dept_raw.strip()
|
|
self.unmapped_arabic_depts[name] = self.unmapped_arabic_depts.get(name, 0) + 1
|
|
|
|
assigned_to_user = self._get_or_create_data_entry_user(row_data.get("data_entry_person"))
|
|
|
|
# Determine status from timeline dates
|
|
status = "open"
|
|
if closed_date:
|
|
status = "closed"
|
|
elif resolved_date:
|
|
status = "resolved"
|
|
elif escalated_date:
|
|
status = "in_progress"
|
|
|
|
# Normalize satisfaction
|
|
satisfaction_raw = str(row_data.get("satisfaction") or "").lower().strip()
|
|
satisfaction_val = SATISFACTION_MAP.get(satisfaction_raw, "")
|
|
|
|
# Normalize rightful side
|
|
rightful_side = str(row_data.get("rightful_side") or "").lower().strip()
|
|
resolution_outcome = ""
|
|
if rightful_side in ["patient", "hospital", "other"]:
|
|
resolution_outcome = rightful_side
|
|
|
|
# Rich text fields
|
|
complaint_subject = str(row_data.get("complaint_subject") or "").strip()
|
|
delay_reason = str(row_data.get("delay_reason") or "").strip()
|
|
closure_delay = str(row_data.get("closure_delay") or "").strip()
|
|
action_taken = str(row_data.get("action_taken") or "").strip()
|
|
action_result = str(row_data.get("action_result") or "").strip()
|
|
recommendation_raw = str(row_data.get("recommendation") or "").strip()
|
|
solutions = str(row_data.get("solutions") or "").strip()
|
|
recommendation_combined = recommendation_raw or solutions or ""
|
|
|
|
if not self.dry_run:
|
|
with transaction.atomic():
|
|
complaint = Complaint.objects.create(
|
|
reference_number=ref_num,
|
|
hospital=self.hospital,
|
|
department=dept,
|
|
section=section_obj,
|
|
legacy_location=location,
|
|
legacy_main_section=main_section,
|
|
legacy_subsection=subsection,
|
|
old_location_raw=str(row_data.get("location_name") or "")[:200],
|
|
old_main_section_raw=str(row_data.get("main_dept_name") or "")[:200],
|
|
old_subsection_raw=str(row_data.get("sub_dept_name") or "")[:200],
|
|
title=self._build_title(row_data),
|
|
description=self._build_description(row_data),
|
|
patient_name="Unknown",
|
|
national_id="",
|
|
relation_to_patient="patient",
|
|
staff=None,
|
|
staff_name=row_data.get("staff_name") or "",
|
|
domain=None,
|
|
category=None,
|
|
subcategory_obj=None,
|
|
classification_obj=None,
|
|
status=status,
|
|
assigned_to=assigned_to_user,
|
|
resolved_by=assigned_to_user if resolved_date else None,
|
|
resolution_outcome=resolution_outcome,
|
|
form_sent_at=form_sent_date,
|
|
activated_at=activated_date,
|
|
forwarded_to_dept_at=sent_date,
|
|
reminder_sent_at=first_reminder,
|
|
second_reminder_sent_at=second_reminder,
|
|
escalated_at=escalated_date,
|
|
closed_at=closed_date,
|
|
resolved_at=resolved_date,
|
|
explanation_requested=bool(sent_date),
|
|
explanation_requested_at=sent_date,
|
|
explanation_received_at=response_date,
|
|
due_at=created_at + timedelta(hours=48),
|
|
source=px_source,
|
|
complaint_source_type=get_complaint_source_type(px_source),
|
|
satisfaction=satisfaction_val,
|
|
complaint_subject=complaint_subject,
|
|
explanation_delay_reason=delay_reason,
|
|
delay_reason_closure=closure_delay or "",
|
|
action_taken_by_dept=action_taken,
|
|
action_result=action_result,
|
|
recommendation_action_plan=recommendation_combined,
|
|
metadata={
|
|
"import_source": "2025_excel",
|
|
"original_sheet": self.sheet_name,
|
|
"complaint_num": row_data.get("complaint_num"),
|
|
"source": row_data.get("source"),
|
|
},
|
|
sent_to_department=bool(dept),
|
|
sent_to_department_at=created_at if dept else None,
|
|
)
|
|
Complaint.objects.filter(pk=complaint.pk).update(created_at=created_at)
|
|
|
|
self.stats["success"] += 1
|
|
|
|
except Exception as e:
|
|
self.stats["failed"] += 1
|
|
self.errors.append({"row": row_num, "error": str(e)})
|
|
logger.error(f"Row {row_num}: {e}", exc_info=True)
|
|
|
|
row_num += 1
|
|
|
|
def _extract_row_data(self, row_num: int) -> Dict:
|
|
data = {}
|
|
for field, col in self.column_map.items():
|
|
cell_value = self.ws.cell(row_num, col).value
|
|
data[field] = cell_value
|
|
return data
|
|
|
|
def _extract_row_data_from_values(self, row: tuple) -> Dict:
|
|
data = {}
|
|
for field, col in self.column_map.items():
|
|
cell_value = row[col - 1] if col - 1 < len(row) else None
|
|
data[field] = cell_value
|
|
return data
|
|
|
|
def _build_reference_number(self, complaint_num) -> str:
|
|
sheet_parts = self.sheet_name.strip().split()
|
|
year = "2025"
|
|
month_part = sheet_parts[0].upper()
|
|
month_code = MONTH_MAP.get(month_part, "00")
|
|
return f"CMP-{year}-{month_code}-{int(complaint_num):04d}"
|
|
|
|
def _get_unique_reference_number(self, complaint_num) -> str:
|
|
"""Generate unique reference number with suffix if needed."""
|
|
base_ref = self._build_reference_number(complaint_num)
|
|
|
|
if base_ref not in self.used_refs and not Complaint.objects.filter(reference_number=base_ref).exists():
|
|
self.used_refs.add(base_ref)
|
|
return base_ref
|
|
|
|
suffix = ord("A")
|
|
while suffix <= ord("Z"):
|
|
ref_with_suffix = f"{base_ref}-{chr(suffix)}"
|
|
if (
|
|
ref_with_suffix not in self.used_refs
|
|
and not Complaint.objects.filter(reference_number=ref_with_suffix).exists()
|
|
):
|
|
self.used_refs.add(ref_with_suffix)
|
|
return ref_with_suffix
|
|
suffix += 1
|
|
|
|
raise ValueError(f"Cannot generate unique reference for {base_ref}")
|
|
|
|
def _parse_datetime(self, value) -> Optional[datetime]:
|
|
if not value:
|
|
return None
|
|
if isinstance(value, datetime):
|
|
return value
|
|
if isinstance(value, str):
|
|
try:
|
|
return datetime.strptime(value, "%Y-%m-%d %H:%M:%S")
|
|
except ValueError:
|
|
try:
|
|
return datetime.strptime(value, "%Y-%m-%d")
|
|
except ValueError:
|
|
return None
|
|
return None
|
|
|
|
def _resolve_location(self, name_ar: str) -> Optional[LegacyLocation]:
|
|
if not name_ar:
|
|
return None
|
|
return LegacyLocation.objects.filter(name_ar=name_ar).first()
|
|
|
|
def _resolve_section(self, name_ar: str) -> Optional[LegacyMainSection]:
|
|
if not name_ar:
|
|
return None
|
|
return LegacyMainSection.objects.filter(name_ar=name_ar).first()
|
|
|
|
def _resolve_subsection(self, name_ar: str) -> Optional[LegacySubSection]:
|
|
if not name_ar:
|
|
return None
|
|
return LegacySubSection.objects.filter(name_ar=name_ar).first()
|
|
|
|
def _get_or_create_data_entry_user(self, arabic_name: str) -> Optional[User]:
|
|
if not arabic_name:
|
|
return None
|
|
|
|
try:
|
|
from unidecode import unidecode
|
|
except ImportError:
|
|
return None
|
|
|
|
parts = arabic_name.split()
|
|
if len(parts) >= 2:
|
|
first, last = parts[0], parts[-1]
|
|
else:
|
|
first, last = arabic_name, "staff"
|
|
|
|
username_first = re.sub(r"[^a-z0-9]", "", unidecode(first).lower().strip())
|
|
username_last = re.sub(r"[^a-z0-9]", "", unidecode(last).lower().strip())
|
|
|
|
if not username_first:
|
|
username_first = "user"
|
|
if not username_last:
|
|
username_last = "staff"
|
|
|
|
username = f"{username_first}.{username_last}"
|
|
|
|
user = User.objects.filter(username=username).first()
|
|
if user:
|
|
return user
|
|
|
|
try:
|
|
user = User(
|
|
username=username,
|
|
first_name=arabic_name,
|
|
last_name="",
|
|
email=f"{username}@alhammadi.med.sa",
|
|
is_active=True,
|
|
)
|
|
user.save()
|
|
return user
|
|
except Exception:
|
|
return None
|
|
|
|
def _build_title(self, row_data: Dict) -> str:
|
|
desc = row_data.get("description_en") or row_data.get("description_ar") or ""
|
|
return desc[:500] if desc else "No description"
|
|
|
|
def _build_description(self, row_data: Dict) -> str:
|
|
desc_en = row_data.get("description_en") or ""
|
|
desc_ar = row_data.get("description_ar") or ""
|
|
|
|
if desc_en and desc_ar:
|
|
return f"{desc_en}\n\n[Arabic]:\n{desc_ar}"
|
|
return desc_en or desc_ar or "No description provided"
|
|
|
|
def _print_report(self):
|
|
self.stdout.write("\n" + "=" * 60)
|
|
self.stdout.write("IMPORT REPORT - 2025 BASIC")
|
|
self.stdout.write("=" * 60)
|
|
self.stdout.write(f"Sheet: {self.sheet_name}")
|
|
self.stdout.write(f"Mode: {'DRY RUN' if self.dry_run else 'ACTUAL'}")
|
|
self.stdout.write(f"\nProcessed: {self.stats['processed']}")
|
|
self.stdout.write(f"Success: {self.stats['success']}")
|
|
self.stdout.write(f"Failed: {self.stats['failed']}")
|
|
|
|
if self.unmapped_arabic_depts:
|
|
self.stdout.write("\n--- Unmapped Arabic Sub-Departments ---")
|
|
sorted_unmapped = sorted(self.unmapped_arabic_depts.items(), key=lambda x: -x[1])
|
|
for name, count in sorted_unmapped:
|
|
self.stdout.write(f" {count:4d}x {name}")
|
|
|
|
if self.errors:
|
|
self.stdout.write(f"\nErrors: {len(self.errors)}")
|
|
for error in self.errors[:5]:
|
|
self.stdout.write(f" Row {error['row']}: {error['error']}")
|