"""Management command: import_apqc_pcf Imports APQC PCF taxonomy from the Excel file produced by the strategy team. Expected Excel structure (sheet names configurable via options): Sheet ``process`` columns: PCF ID | سطح فرایند (level) | Hierarchy ID | Process element (EN name) | عنصر فرایند (FA name) | تعریف (FA definition) | Definition (EN definition) Sheet ``metrics`` columns: Process Element ID | Hierarchy ID | Process element | Metric Category | Metric ID | Metric name | Formula | Units Usage examples -------------- # Import from default Excel path with auto-detected framework code: python manage.py import_apqc_pcf \\ --file ".docs/mydocs/APQC/APQC PCF 7.2.1 Automotive - Glossary Terms - 5_ zare.xlsx" # Full options: python manage.py import_apqc_pcf \\ --file path/to/file.xlsx \\ --framework-code PCF-AUTO-7.2.1-EN \\ --framework-name "APQC PCF Automotive v7.2.1" \\ --industry automotive \\ --version 7.2.1 \\ --process-sheet process \\ --metrics-sheet metrics \\ --update """ from __future__ import annotations import re from pathlib import Path from typing import Any from django.core.management.base import BaseCommand, CommandError from django.db import transaction class Command(BaseCommand): help = "Import APQC PCF taxonomy from an Excel file." # ------------------------------------------------------------------ # Argument definitions # ------------------------------------------------------------------ def add_arguments(self, parser) -> None: parser.add_argument( "--file", required=True, help="Path to the Excel (.xlsx) file containing PCF data.", ) parser.add_argument( "--framework-code", default=None, help='Unique code for the PCFFramework record (e.g. "PCF-CI-7.2.1-EN"). ' "Auto-derived from filename if omitted.", ) parser.add_argument( "--framework-name", default=None, help="Human-readable name for the framework. Auto-derived if omitted.", ) parser.add_argument( "--industry", default="cross_industry", choices=[ "cross_industry", "automotive", "healthcare", "education", "financial", "energy", "retail", "telecom", "government", "construction", "manufacturing", "other", ], help="Industry type for this framework (default: cross_industry).", ) parser.add_argument( "--pcf-version", default="7.2.1", dest="version", help='PCF version string (default: "7.2.1").', ) parser.add_argument( "--language", default="en", help='Primary language code (default: "en").', ) parser.add_argument( "--process-sheet", default="Processes", help='Name of the Excel sheet holding process elements (default: "Processes"). Case-insensitive.', ) parser.add_argument( "--metrics-sheet", default="Metrics", help='Name of the Excel sheet holding metrics (default: "Metrics"). Case-insensitive.', ) parser.add_argument( "--operational-sheet", default="operational sequences", help='Sheet with operational sequences — used for FA names/triggers (optional). Case-insensitive.', ) parser.add_argument( "--update", action="store_true", default=False, help="Update existing records instead of skipping them.", ) parser.add_argument( "--dry-run", action="store_true", default=False, help="Parse and validate without writing to the database.", ) # ------------------------------------------------------------------ # Main entry # ------------------------------------------------------------------ def handle(self, *args: Any, **options: Any) -> None: try: import openpyxl # noqa: F401 except ImportError: raise CommandError( "openpyxl is required to read Excel files.\n" "Install it with: pip install openpyxl" ) file_path = Path(options["file"]) if not file_path.exists(): raise CommandError(f"File not found: {file_path}") framework_code = options["framework_code"] or self._derive_code(file_path, options) framework_name = options["framework_name"] or f"APQC PCF {options['version']} ({options['industry']})" self.stdout.write(self.style.NOTICE(f"Reading: {file_path}")) self.stdout.write(self.style.NOTICE(f"Framework code: {framework_code}")) rows, metrics_rows, op_rows = self._read_excel( file_path, options["process_sheet"], options["metrics_sheet"], options["operational_sheet"], ) if options["dry_run"]: self._dry_run_report(rows, metrics_rows) return with transaction.atomic(): framework = self._upsert_framework( code=framework_code, name=framework_name, industry=options["industry"], version=options["version"], language=options["language"], update=options["update"], ) element_map = self._import_elements(framework, rows, op_rows, update=options["update"]) self._import_metrics(element_map, metrics_rows, update=options["update"]) self.stdout.write(self.style.SUCCESS( f"Done. Framework '{framework_code}' — " f"{len(element_map)} elements, {len(metrics_rows)} metrics." )) # ------------------------------------------------------------------ # Excel reading # ------------------------------------------------------------------ def _read_excel( self, path: Path, proc_sheet: str, metrics_sheet: str, op_sheet: str ) -> tuple[list[dict], list[dict], list[dict]]: import openpyxl wb = openpyxl.load_workbook(path, read_only=True, data_only=True) proc_rows = self._sheet_to_dicts(wb, proc_sheet) metrics_rows = self._sheet_to_dicts(wb, metrics_sheet) op_rows = self._sheet_to_dicts(wb, op_sheet) if any(s.lower().strip() == op_sheet.lower().strip() for s in wb.sheetnames) else [] wb.close() return proc_rows, metrics_rows, op_rows @staticmethod def _sheet_to_dicts(wb, sheet_name: str) -> list[dict]: # Case-insensitive sheet lookup actual_name = next( (n for n in wb.sheetnames if n.lower().strip() == sheet_name.lower().strip()), None, ) if actual_name is None: return [] ws = wb[actual_name] rows = list(ws.iter_rows(values_only=True)) if not rows: return [] headers = [str(h).strip() if h is not None else f"col_{i}" for i, h in enumerate(rows[0])] return [ {headers[i]: (cell if cell is not None else "") for i, cell in enumerate(row)} for row in rows[1:] if any(cell for cell in row) ] # ------------------------------------------------------------------ # Framework upsert # ------------------------------------------------------------------ def _upsert_framework( self, code: str, name: str, industry: str, version: str, language: str, update: bool ): from simorgh.apps.bpm.models import PCFFramework fw, created = PCFFramework.objects.get_or_create( code=code, defaults={ "name": name, "industry": industry, "version": version, "language": language, "is_active": True, }, ) if not created and update: fw.name = name fw.industry = industry fw.version = version fw.language = language fw.save(update_fields=["name", "industry", "version", "language", "updated_at"]) status = "created" if created else ("updated" if update else "skipped") self.stdout.write(f" Framework {status}: {fw.code}") return fw # ------------------------------------------------------------------ # Element import # ------------------------------------------------------------------ _LEVEL_MAP = {1: 1, 2: 2, 3: 3, 4: 4} def _import_elements( self, framework, rows: list[dict], op_rows: list[dict], update: bool ) -> dict[str, Any]: """Import PCFElement rows; return {hierarchy_id: instance} map.""" from simorgh.apps.bpm.models import PCFElement # Build FA-name lookup from operational sequences sheet fa_name_map: dict[str, str] = {} for row in op_rows: hid = str(row.get("Hierarchy ID", "") or row.get("hierarchy id", "")).strip() fa = str(row.get("نام فارسی", "") or "").strip() if hid and fa: fa_name_map[hid] = fa element_map: dict[str, PCFElement] = {} created_count = updated_count = skipped_count = 0 # Sort rows by hierarchy depth so parents are inserted before children def _depth(r: dict) -> int: hid = str(r.get("Hierarchy ID", "") or r.get("hierarchy id", "")).strip() return len(hid.split(".")) if hid else 0 for row in sorted(rows, key=_depth): pcf_id_raw = row.get("PCF ID", row.get("pcf id", "")) level_raw = row.get("سطح فرایند", row.get("level", row.get("Level", ""))) hierarchy_id = str(row.get("Hierarchy ID", row.get("hierarchy id", "")) or "").strip() name_en = str(row.get("Process element", row.get("process element", "")) or "").strip() name_fa = str(row.get("عنصر فرایند", row.get("عنصر فرآیند", "")) or "").strip() def_en = str(row.get("Definition", row.get("definition", "")) or "").strip() def_fa = str(row.get("تعریف", row.get("تعریف (ترجمه گوگل)", "")) or "").strip() if not hierarchy_id or not name_en: continue # Normalise hierarchy_id — handle integers (1 → "1") and floats (1.0 → "1") try: hid_float = float(str(hierarchy_id)) if hid_float == int(hid_float) and "." not in str(hierarchy_id): hierarchy_id = str(int(hid_float)) else: hierarchy_id = str(hierarchy_id).strip() except (ValueError, TypeError): hierarchy_id = str(hierarchy_id).strip() try: pcf_id = int(float(str(pcf_id_raw))) if pcf_id_raw else 0 except (ValueError, TypeError): pcf_id = 0 try: # Level may be an integer, a float, or "2 Process Group" style string level_str = str(level_raw).strip() if level_raw else "" m = re.match(r"^(\d+)", level_str) if m: level = int(m.group(1)) else: level = int(float(level_str)) if level_str else len(hierarchy_id.split(".")) except (ValueError, TypeError): level = len(hierarchy_id.split(".")) # Fallback FA name from operational sheet if not name_fa: name_fa = fa_name_map.get(hierarchy_id, "") # Derive order from the last segment of hierarchy_id try: order = int(re.split(r"[.\-]", hierarchy_id)[-1]) except (ValueError, IndexError): order = 0 # Resolve parent parent = None if "." in hierarchy_id: parent_hid = hierarchy_id.rsplit(".", 1)[0] parent = element_map.get(parent_hid) defaults = { "level": level, "parent": parent, "name_en": name_en, "name_fa": name_fa, "definition_en": def_en, "definition_fa": def_fa, "order": order, "is_active": True, } lookup = {"framework": framework, "hierarchy_id": hierarchy_id} if pcf_id: lookup["pcf_id"] = pcf_id try: obj, created = PCFElement.objects.get_or_create( framework=framework, hierarchy_id=hierarchy_id, defaults={**defaults, "pcf_id": pcf_id or 0}, ) except PCFElement.MultipleObjectsReturned: obj = PCFElement.objects.filter(framework=framework, hierarchy_id=hierarchy_id).first() created = False if not created and update: for k, v in defaults.items(): setattr(obj, k, v) if pcf_id: obj.pcf_id = pcf_id obj.save() updated_count += 1 elif created: created_count += 1 else: skipped_count += 1 element_map[hierarchy_id] = obj self.stdout.write( f" Elements — created: {created_count}, " f"updated: {updated_count}, skipped: {skipped_count}" ) return element_map # ------------------------------------------------------------------ # Metrics import # ------------------------------------------------------------------ def _import_metrics( self, element_map: dict, rows: list[dict], update: bool ) -> None: from simorgh.apps.bpm.models import PCFMetric _cat_map = { "process efficiency": "efficiency", "process effectiveness": "effectiveness", "process cycle time": "cycle_time", "process cost": "cost", "process quality": "quality", } created_count = skipped_count = 0 for row in rows: hierarchy_id = str(row.get("Hierarchy ID", row.get("hierarchy id", "")) or "").strip() metric_id = str(row.get("Metric ID", row.get("metric id", "")) or "").strip() cat_raw = str(row.get("Metric Category", row.get("metric category", "")) or "").strip().lower() name = str(row.get("Metric name", row.get("metric name", "")) or "").strip() formula = str(row.get("Formula", row.get("formula", "")) or "").strip() unit = str(row.get("Units", row.get("units", "")) or "").strip() # Normalize hierarchy_id (handles "1.0" → "1") try: hf = float(hierarchy_id) if hf == int(hf) and "." not in hierarchy_id: hierarchy_id = str(int(hf)) else: # "1.0" → "1", "1.1" stays "1.1" parts = hierarchy_id.split(".") if len(parts) == 2 and parts[1] == "0": hierarchy_id = parts[0] except (ValueError, TypeError): pass if not metric_id or not name: continue element = element_map.get(hierarchy_id) if element is None: self.stdout.write( self.style.WARNING(f" Metric {metric_id}: element '{hierarchy_id}' not found, skipping.") ) continue category = _cat_map.get(cat_raw, "other") obj, created = PCFMetric.objects.get_or_create( pcf_element=element, metric_id=metric_id, defaults={"category": category, "name": name, "formula": formula, "unit": unit}, ) if not created and update: obj.category = category obj.name = name obj.formula = formula obj.unit = unit obj.save(update_fields=["category", "name", "formula", "unit", "updated_at"]) if created: created_count += 1 else: skipped_count += 1 self.stdout.write(f" Metrics — created: {created_count}, skipped: {skipped_count}") # ------------------------------------------------------------------ # Helpers # ------------------------------------------------------------------ @staticmethod def _derive_code(path: Path, options: dict) -> str: industry = options["industry"].replace("_", "-").upper()[:4] version = options["version"].replace(".", "_") lang = options["language"].upper() return f"PCF-{industry}-{version}-{lang}" def _dry_run_report(self, rows: list[dict], metrics_rows: list[dict]) -> None: self.stdout.write(self.style.WARNING("DRY RUN — no changes written.")) self.stdout.write(f" Process rows : {len(rows)}") self.stdout.write(f" Metrics rows : {len(metrics_rows)}") levels: dict[int, int] = {} for row in rows: lv_raw = row.get("سطح فرایند", row.get("level", row.get("Level", ""))) try: lv_str = str(lv_raw).strip() if lv_raw else "" m = re.match(r"^(\d+)", lv_str) lv = int(m.group(1)) if m else (int(float(lv_str)) if lv_str else 0) except (ValueError, TypeError): lv = 0 levels[lv] = levels.get(lv, 0) + 1 for lv, cnt in sorted(levels.items()): self.stdout.write(f" Level {lv}: {cnt} elements")