"""Import the official RUFORUM MEL Framework workbook into a Results Framework. The RUFORUM team's ``RUFORUM MEL FRAMEWORK.xlsm`` carries the AHESTI results chain on its visible sheet:: AHESTI 10-Year Objective (2018-2030) → Strategic Objective (a LogFrame) AHESTI 10-Year Outcome (2018-2030) → Impact (chain root, no parent) AHESTI Intermediate Outcome (2024-2028)→ Intermediate Outcome RUFORUM High-Level Output → Output Indicator + Disaggregation → Indicator (DRAFT) + categories Each distinct objective becomes its own Strategic Objective (LogFrame); the ``--slug`` argument is the base slug (suffixed ``-1``, ``-2``, … when the workbook carries more than one objective). Hierarchy columns are merged/blank on continuation rows, so values are forward-filled. Indicators are created in DRAFT state with their disaggregation *categories* stubbed from the workbook's free-text column — officers then add the values, baselines and targets, and the publish gate (M&E SRS Table 64) enforces completeness before the indicator goes live. Idempotent: rows and indicators are matched by title/name within the target framework, so re-running updates nothing and creates no duplicates. Usage: manage.py import_ruforum_mel_framework /path/to/framework.xlsm [--dry-run] """ from __future__ import annotations import re from django.core.management.base import BaseCommand, CommandError from django.db import transaction FRAME_SHEET = "RUFORUM MEL FRAME" def _clean(value) -> str: return re.sub(r"\s+", " ", str(value)).strip() if value not in (None, "") else "" def _split_categories(spec: str) -> list[str]: """Split a free-text disaggregation spec into category names. Splits on commas/semicolons only *outside* parentheses so phrases like "by level (undergraduate, TVET, postgraduate)" stay one category. """ parts: list[str] = [] buf: list[str] = [] depth = 0 for ch in spec: if ch in "([": depth += 1 elif ch in ")]": depth = max(0, depth - 1) if ch in ",;" and depth == 0: parts.append("".join(buf)) buf = [] else: buf.append(ch) parts.append("".join(buf)) out = [] for part in parts: name = _clean(part).strip("-• ") if name: out.append(name) return out class Command(BaseCommand): help = "Import the RUFORUM MEL Framework workbook as Strategic Objectives." def add_arguments(self, parser): parser.add_argument("workbook", help="Path to RUFORUM MEL FRAMEWORK.xlsm") parser.add_argument( "--dry-run", action="store_true", help="Parse and report without writing anything.", ) parser.add_argument( "--slug", default="ruforum-mel-framework", help="Base slug for the target Strategic Objective(s) (default: ruforum-mel-framework).", ) @transaction.atomic def handle(self, *args, **opts): try: import openpyxl except ImportError as exc: # pragma: no cover raise CommandError("openpyxl is required for this import.") from exc from apps.mel.indicators.models import ( Indicator, IndicatorState, LogFrame, LogFrameLevel, LogFrameRow, ) try: wb = openpyxl.load_workbook(opts["workbook"], data_only=True) except Exception as exc: raise CommandError(f"Cannot open workbook: {exc}") from exc if FRAME_SHEET not in wb.sheetnames: raise CommandError( f"Sheet {FRAME_SHEET!r} not found — got {wb.sheetnames}." ) ws = wb[FRAME_SHEET] # ── Parse with forward-fill over the merged hierarchy columns ────── parsed: list[dict] = [] current = {"objective": "", "outcome": "", "intermediate": "", "output": ""} for row in ws.iter_rows(min_row=2, max_col=7): objective = _clean(row[1].value) outcome = _clean(row[2].value) intermediate = _clean(row[3].value) output = _clean(row[4].value) indicator = _clean(row[5].value) disaggregation = _clean(row[6].value) if objective: current["objective"] = objective if outcome: current["outcome"] = outcome if intermediate: current["intermediate"] = intermediate if output: current["output"] = output if not indicator: continue parsed.append({**current, "indicator": indicator, "disagg": disaggregation}) if not parsed: raise CommandError("No indicator rows parsed — is this the right workbook?") self.stdout.write( f"Parsed {len(parsed)} indicators across " f"{len({p['objective'] for p in parsed})} objectives / " f"{len({p['output'] for p in parsed})} outputs." ) if opts["dry_run"]: for p in parsed: self.stdout.write( f" [{p['output'][:40]}…] {p['indicator'][:70]}" f" ⇒ categories: {_split_categories(p['disagg'])}" ) transaction.set_rollback(True) return # ── Strategic Objectives (one framework per objective) ──────────── # In the M&E framework the Strategic Objective IS the framework # container. Each distinct AHESTI objective therefore becomes its own # LogFrame; its Impact → Intermediate Outcome → Output chain hangs off it. objectives: list[str] = [] for p in parsed: if p["objective"] and p["objective"] not in objectives: objectives.append(p["objective"]) base_slug = opts["slug"] single = len(objectives) <= 1 lf_by_objective: dict[str, "LogFrame"] = {} lf_created_count = 0 for idx, objective in enumerate(objectives, start=1): obj_slug = base_slug if single else f"{base_slug}-{idx}" logframe, lf_created = LogFrame.objects.get_or_create( slug=obj_slug, defaults={ "name": objective[:255], "code": f"AHESTI-SO-{idx:02d}", "planning_period": "2024–2028", "programme": "RUFORUM Secretariat", "description": ( "Official RUFORUM MEL Framework as shared by the RUFORUM " "team — AHESTI 10-year Strategic Objective with its " "2024–2028 Impacts, Intermediate Outcomes and Outputs." ), }, ) lf_by_objective[objective] = logframe if lf_created: lf_created_count += 1 stats = {"rows": 0, "indicators": 0, "categories": 0} def get_row(logframe, level, title, parent): row, created = LogFrameRow.objects.get_or_create( logframe=logframe, level=level, title=title[:255], defaults={"parent": parent}, ) if created: stats["rows"] += 1 return row seq = ( Indicator.objects.filter(code__startswith="ruf-mel-") .count() ) for p in parsed: logframe = lf_by_objective[p["objective"]] # Impact is the chain root — it hangs directly off the Strategic # Objective container (no parent row). impact = get_row(logframe, LogFrameLevel.IMPACT, p["outcome"], None) outcome = get_row(logframe, LogFrameLevel.OUTCOME, p["intermediate"], impact) output = get_row(logframe, LogFrameLevel.OUTPUT, p["output"], outcome) indicator = Indicator.objects.filter( logframe_row__logframe=logframe, name__iexact=p["indicator"][:255] ).first() if indicator is None: seq += 1 indicator = Indicator.objects.create( logframe_row=output, code=f"ruf-mel-{seq:02d}", name=p["indicator"][:255], definition=p["indicator"], unit="count", calculation_method="As defined in the RUFORUM MEL Framework.", data_source="manual", workflow_status=IndicatorState.DRAFT, ) stats["indicators"] += 1 for name in _split_categories(p["disagg"]): # Uppercase only the first character — "sex" → "Sex" without # mangling embedded acronyms like "(SME, corporate)". label = (name[0].upper() + name[1:])[:120] _, created = indicator.disaggregation_categories.get_or_create( name=label ) if created: stats["categories"] += 1 self.stdout.write( self.style.SUCCESS( f"Imported {len(objectives)} Strategic Objective(s) " f"({lf_created_count} newly created): " f"+{stats['rows']} rows, +{stats['indicators']} draft indicators, " f"+{stats['categories']} disaggregation categories." ) ) self.stdout.write( "Indicators are DRAFT — add disaggregation values, baselines and " "targets, then publish (the M&E SRS gate enforces completeness)." )