""" 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."))