"""Load + normalise the authoritative close-checklist seed (129 task templates). The raw snapshot lives in ``close_seed/close_checklist_seed.json`` (captured from the finance team's live tool). This module parses it into normalised rows for the template table — pure, no DB — so the seed migration and the runtime re-seed share one source of truth. Enum normalisation (raw → canonical stored value): frequency : "On going"→ongoing, "Monthly"→monthly, "Quarterly"→quarterly, "Annual"→annual aiWork : "Y-AI Assist"→assist, "Y-AI Perform"→perform, "No"→no, "NA"→na, ""→none The ``calendarDay`` token ("Day+1" / "Day-3" / "On going") is kept verbatim — that is exactly what :func:`close_calendar.resolve_anchor` consumes. """ from __future__ import annotations import functools import json from pathlib import Path _SEED_PATH = Path(__file__).with_name("close_seed") / "close_checklist_seed.json" _FREQ = { "On going": "ongoing", "Monthly": "monthly", "Quarterly": "quarterly", "Annual": "annual", } _AI = { "Y-AI Assist": "assist", "Y-AI Perform": "perform", "No": "no", "NA": "na", "": "none", } # Canonical stored enum values (also used by schema validation). FREQUENCIES = ("ongoing", "monthly", "quarterly", "annual") AI_WORK_VALUES = ("assist", "perform", "no", "na", "none") PROCESSES = ( "AP", "COGS", "OPEX", "Rev rec", "Collection", "Tax", "Cost Center update", "Payroll", "Intercompany Accounting", "Treasury Accounting", "Tax Accounting", "Reporting", ) @functools.lru_cache(maxsize=1) def load_seed() -> dict: """Parse and cache the raw seed JSON.""" with _SEED_PATH.open(encoding="utf-8") as fh: data: dict = json.load(fh) return data def normalise_frequency(raw: str) -> str: return _FREQ.get((raw or "").strip(), "monthly") def normalise_ai_work(raw: str) -> str: return _AI.get((raw or "").strip(), "none") def owner_email_hints() -> dict[str, str]: return dict(load_seed().get("ownerEmailHints", {})) def entity_codes() -> list[str]: return list(load_seed().get("entityCodes", [])) def close_config_defaults() -> dict: return dict(load_seed().get("closeConfig", {})) def template_rows() -> list[dict]: """Normalised template rows in seed order (``sort_order`` = index).""" from app.domains.finance.close_signal_keys import DEFAULT_SIGNAL_BINDINGS seed = load_seed() rows: list[dict] = [] for idx, t in enumerate(seed["templates"]): rows.append({ "source_key": t["id"], "process": t["process"], "sub_process": (t.get("subProcess") or "").strip(), "description": t["task"], "owner_label": (t.get("owner") or "").strip(), "frequency": normalise_frequency(t.get("frequency", "")), "calendar_day": (t.get("calendarDay") or "On going").strip(), "estimated_hours": t.get("estimatedHours"), "submit_to_fpa": bool(t.get("submitToFPA")), "ai_work": normalise_ai_work(t.get("aiWork", "")), "link": (t.get("link") or "").strip(), "entity_codes": list(t.get("entities", [])), "sort_order": idx, "signal_key": DEFAULT_SIGNAL_BINDINGS.get(t["id"]), }) return rows