214 lines
8.0 KiB
Python
214 lines
8.0 KiB
Python
"""
|
|
Backfill source (PXSource FK) for complaints that were imported without one.
|
|
|
|
Handles two import batches:
|
|
1. historical_excel_2022: metadata.source contains the Arabic source value
|
|
2. 2025_excel: metadata has no source — re-reads the Excel file to extract it
|
|
|
|
Usage:
|
|
# Preview what would change
|
|
python manage.py backfill_complaint_sources --dry-run
|
|
|
|
# Execute the backfill
|
|
python manage.py backfill_complaint_sources
|
|
|
|
# Only process one batch
|
|
python manage.py backfill_complaint_sources --batch=historical_excel_2022
|
|
python manage.py backfill_complaint_sources --batch=2025_excel
|
|
"""
|
|
|
|
import logging
|
|
|
|
from django.core.management.base import BaseCommand
|
|
from django.db import transaction
|
|
|
|
from apps.complaints.models import Complaint
|
|
from apps.complaints.management.commands.complaint_source_mapping import (
|
|
resolve_px_source,
|
|
get_complaint_source_type,
|
|
)
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
EXCEL_2025_PATH = "data/Complaints Report - 2025.xlsx"
|
|
|
|
# Header aliases for 2025 Excel column detection
|
|
HEADER_ALIASES_2025 = {
|
|
"complaint_num": ["رقم الشكوى"],
|
|
"source": ["جهة الشكوى"],
|
|
}
|
|
|
|
|
|
class Command(BaseCommand):
|
|
help = "Backfill PXSource for complaints imported without one."
|
|
|
|
def add_arguments(self, parser):
|
|
parser.add_argument("--dry-run", action="store_true", help="Preview without saving")
|
|
parser.add_argument(
|
|
"--batch",
|
|
choices=["historical_excel_2022", "2025_excel"],
|
|
default=None,
|
|
help="Only process a specific import batch",
|
|
)
|
|
parser.add_argument(
|
|
"--excel-file",
|
|
default=EXCEL_2025_PATH,
|
|
help=f"Path to 2025 Excel file (default: {EXCEL_2025_PATH})",
|
|
)
|
|
|
|
def handle(self, *args, **options):
|
|
self.dry_run = options["dry_run"]
|
|
self.excel_file = options["excel_file"]
|
|
batch_filter = options["batch"]
|
|
|
|
stats = {"total": 0, "updated": 0, "unmapped": 0, "no_source_value": 0, "skipped_2026": 0}
|
|
|
|
no_source_qs = Complaint.objects.filter(source__isnull=True)
|
|
|
|
if batch_filter:
|
|
no_source_qs = no_source_qs.filter(metadata__import_source=batch_filter)
|
|
|
|
self.stdout.write(f"\nFound {no_source_qs.count()} complaints with no source.")
|
|
if self.dry_run:
|
|
self.stdout.write(self.style.WARNING("DRY RUN — no changes will be saved.\n"))
|
|
|
|
# Build 2025 Excel lookup if needed
|
|
excel_lookup = {}
|
|
needs_excel = not batch_filter or batch_filter == "2025_excel"
|
|
if needs_excel:
|
|
self.stdout.write("Reading 2025 Excel for source column...")
|
|
excel_lookup = self._build_2025_excel_lookup()
|
|
self.stdout.write(f" Built lookup with {len(excel_lookup)} entries.\n")
|
|
|
|
for complaint in no_source_qs.iterator():
|
|
stats["total"] += 1
|
|
|
|
# Skip 2026 test data
|
|
if complaint.created_at.year == 2026:
|
|
stats["skipped_2026"] += 1
|
|
continue
|
|
|
|
import_source = complaint.metadata.get("import_source", "")
|
|
source_value = None
|
|
|
|
if import_source == "historical_excel_2022":
|
|
source_value = complaint.metadata.get("source")
|
|
|
|
elif import_source == "2025_excel":
|
|
complaint_num = complaint.metadata.get("complaint_num")
|
|
original_sheet = complaint.metadata.get("original_sheet", "")
|
|
lookup_key = (complaint_num, original_sheet)
|
|
source_value = excel_lookup.get(lookup_key)
|
|
|
|
else:
|
|
# Unknown import source — try metadata.source as fallback
|
|
source_value = complaint.metadata.get("source")
|
|
|
|
if not source_value:
|
|
stats["no_source_value"] += 1
|
|
continue
|
|
|
|
px_source = resolve_px_source(source_value)
|
|
|
|
if px_source is None:
|
|
stats["unmapped"] += 1
|
|
self.stdout.write(
|
|
self.style.WARNING(
|
|
f" UNMAPPED: Ref={complaint.reference_number} source_value=\"{source_value}\""
|
|
)
|
|
)
|
|
continue
|
|
|
|
if not self.dry_run:
|
|
with transaction.atomic():
|
|
complaint.source = px_source
|
|
complaint.complaint_source_type = get_complaint_source_type(px_source)
|
|
complaint.save(update_fields=["source", "complaint_source_type"])
|
|
|
|
# Store source in metadata for 2025_excel if missing
|
|
if import_source == "2025_excel" and "source" not in complaint.metadata:
|
|
complaint.metadata["source"] = source_value
|
|
complaint.save(update_fields=["metadata"])
|
|
|
|
stats["updated"] += 1
|
|
|
|
self._print_report(stats)
|
|
|
|
def _build_2025_excel_lookup(self):
|
|
"""Read all sheets from 2025 Excel and build (complaint_num, sheet) → source_value map."""
|
|
import openpyxl
|
|
|
|
try:
|
|
wb = openpyxl.load_workbook(self.excel_file, read_only=True, data_only=True)
|
|
except FileNotFoundError:
|
|
self.stdout.write(self.style.ERROR(f"Excel file not found: {self.excel_file}"))
|
|
return {}
|
|
|
|
lookup = {}
|
|
|
|
for sheet_name in wb.sheetnames:
|
|
if sheet_name == "DropDown":
|
|
continue
|
|
|
|
ws = wb[sheet_name]
|
|
col_map = self._detect_columns(ws)
|
|
|
|
if "complaint_num" not in col_map or "source" not in col_map:
|
|
self.stdout.write(self.style.WARNING(f" Sheet '{sheet_name}': could not detect columns, skipping"))
|
|
continue
|
|
|
|
complaint_col = col_map["complaint_num"]
|
|
source_col = col_map["source"]
|
|
|
|
for row in ws.iter_rows(min_row=3, values_only=True):
|
|
comp_num = row[complaint_col - 1] if complaint_col - 1 < len(row) else None
|
|
source_val = row[source_col - 1] if source_col - 1 < len(row) else None
|
|
|
|
if comp_num and source_val:
|
|
normalized_source = str(source_val).strip()
|
|
if normalized_source and not normalized_source.replace(".", "", 1).isdigit():
|
|
lookup[(comp_num, sheet_name)] = normalized_source
|
|
|
|
wb.close()
|
|
return lookup
|
|
|
|
def _detect_columns(self, ws):
|
|
"""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 = ws.cell(r, c).value
|
|
if val:
|
|
if val not in row_values:
|
|
row_values[val] = c
|
|
|
|
if "رقم الشكوى" in row_values:
|
|
for field, aliases in HEADER_ALIASES_2025.items():
|
|
for alias in aliases:
|
|
if alias in row_values:
|
|
mapping[field] = row_values[alias]
|
|
break
|
|
break
|
|
|
|
return mapping
|
|
|
|
def _print_report(self, stats):
|
|
self.stdout.write(f"\n{'=' * 60}")
|
|
self.stdout.write(self.style.SUCCESS("Backfill Report"))
|
|
self.stdout.write(f"{'=' * 60}")
|
|
self.stdout.write(f"Total complaints scanned: {stats['total']}")
|
|
self.stdout.write(f"Updated: {stats['updated']}")
|
|
self.stdout.write(f"Unmapped source value: {stats['unmapped']}")
|
|
self.stdout.write(f"No source value found: {stats['no_source_value']}")
|
|
self.stdout.write(f"Skipped (2026 test data): {stats['skipped_2026']}")
|
|
|
|
if stats["updated"] > 0:
|
|
if self.dry_run:
|
|
self.stdout.write(self.style.WARNING("\nDry run complete — no changes saved."))
|
|
else:
|
|
self.stdout.write(self.style.SUCCESS(f"\nSuccessfully updated {stats['updated']} complaints."))
|
|
|
|
if stats["unmapped"] > 0:
|
|
self.stdout.write(self.style.WARNING(f"\n{stats['unmapped']} complaints had source values that could not be mapped."))
|