← Files VeraARCHIVED FILE

modules/business-planning/scripts/planning_presentation.py

27.7 KB · Oct 2, 2026 · 00:29 UTC

↓ Download file

"""Shared report presentation with mechanically checked figure/source bindings.

Authors choose useful comparisons and decision criteria. Code only checks their
references and arithmetic and renders the same material in HTML and PDF.
"""

from __future__ import annotations

import html
from decimal import Decimal
from typing import Any, Callable
from urllib.parse import urlsplit

from planning_workflow import indexed, number, require

__all__ = [
    "language",
    "label",
    "format_number",
    "format_unit",
    "validate_presentation",
    "render_tables",
    "comparison_rows",
    "render_actions",
    "render_sources",
]

ITALIAN = {
    "Period": "Periodo",
    "Comparison": "Confronto",
    "Print current view": "Stampa la vista corrente",
    "client": "Cliente",
    "Report sections": "Sezioni del report",
    "Income statement": "Conto economico",
    "Cash": "Cassa",
    "Sources": "Fonti",
    "Current planning question": "Domanda di questa iterazione",
    "What we learned": "Che cosa abbiamo appreso",
    "Decision in this round": "Decisione di questa iterazione",
    "Next test or action": "Prossima verifica o azione",
    "When to revisit": "Quando riesaminare il piano",
    "Planning history and input changes": "Iterazioni precedenti e modifiche agli input",
    "Financing assessment": "Valutazione del finanziamento",
    "Bank debt": "Credito bancario",
    "Venture equity": "Capitale di venture capital",
    "Provider not yet selected": "Finanziatore da individuare",
    "Financing conclusion pending evidence and review": "Conclusione sul finanziamento in attesa di evidenze e revisione",
    "Recommendation pending evidence and review": "Raccomandazione in attesa di evidenze e revisione",
    "Explore financing": "Esplorare il finanziamento",
    "Prepare the request": "Preparare la richiesta",
    "Revise the request": "Rivedere la richiesta",
    "Requested financing is not suitable": "Il finanziamento richiesto non è adatto",
    "Business credibility and demand": "Credibilità del business e domanda",
    "Use of funds and proposed loan terms": "Impiego dei fondi e condizioni del prestito",
    "Repayment from operating cash": "Rimborso con la cassa operativa",
    "Repayment under adverse conditions": "Rimborso in condizioni sfavorevoli",
    "Borrower, existing debt and sponsor contribution": "Impresa, debiti esistenti e apporto dei soci",
    "Guarantees and collateral": "Garanzie personali e reali",
    "Lender requirements and missing evidence": "Richieste della banca ed evidenze mancanti",
    "Market opportunity and customer evidence": "Opportunità di mercato ed evidenze sui clienti",
    "Competition and defensible advantage": "Concorrenza e vantaggio difendibile",
    "Traction and repeatable growth": "Risultati commerciali e crescita ripetibile",
    "Team and ability to execute": "Team e capacità di esecuzione",
    "Use of funds and funded milestones": "Impiego dei fondi e traguardi finanziati",
    "Cash runway and further funding": "Autonomia di cassa e finanziamenti successivi",
    "Ownership, dilution and investor return": "Quote, diluizione e rendimento per l’investitore",
    "Investor fit and missing evidence": "Coerenza con l’investitore ed evidenze mancanti",
    "The forecast does not yet cover the stated repayment or funding milestone.": "Le previsioni non coprono ancora la scadenza di rimborso o il traguardo finanziato dichiarato.",
    "This assessment does not establish lender or investor approval.": "Questa valutazione non attesta l’approvazione della banca o dell’investitore.",
    "Business plan": "Business plan",
    "Audience": "Destinatari",
    "internal": "Uso interno",
    "Recommendation": "Raccomandazione",
    "Business assessment withheld": "Valutazione del business sospesa",
    "Correct the conflicting calculations and supporting claims, then review the recommendation again. The submitted assessment is retained in the case record; it is not presented as a conclusion.": "Correggere i calcoli e le affermazioni in conflitto, quindi riesaminare la raccomandazione. La valutazione ricevuta resta nel fascicolo e non viene presentata come conclusione.",
    "Proceed": "Procedere",
    "Test": "Testare prima del lancio",
    "Redesign": "Riprogettare",
    "Stop": "Fermarsi",
    "What this judgment depends on": "Condizioni per proseguire",
    "What would change the recommendation": "Quali prove cambierebbero la decisione",
    "The business and its customers": "Prodotto e clienti",
    "Demand and route to market": "Domanda e canali di vendita",
    "How the business would operate": "Produzione, persone e operazioni",
    "Prices, costs and sustainable sales": "Prezzi, costi e redditività",
    "Cash, investment and financing": "Cassa e finanziamento",
    "Alternatives worth considering": "Alternative realistiche",
    "What to do next": "Prossimi passi e responsabilità",
    "Material uncertainties and limitations": "Incertezze rilevanti per la decisione",
    "Provisional assessment. Material evidence or review remains open; this is not a finalized business plan.": "Valutazione provvisoria per discussione: restano dati o verifiche aperte. Non è un piano approvato per investire o richiedere credito.",
    "Basis and review": "Fonti e stato di revisione",
    "Provisional interpretation — professional review pending": "Interpretazione provvisoria; revisione professionale da acquisire",
    "Reviewed interpretation": "Interpretazione revisionata",
    "Sources and calculation references": "Fonti e riferimenti ai calcoli",
    "Source": "Fonte",
    "Reference": "Riferimento",
    "Value": "Valore",
    "Claim / use": "Affermazione / utilizzo",
    "Location": "Pagina o celle",
    "Not specified": "Non specificato",
    "Action": "Azione",
    "Owner": "Responsabile",
    "When": "Quando",
    "Undated operating period": "Periodo operativo senza data",
    "Evidence / decision criterion": "Prova / criterio di decisione",
    "Supporting evidence, calculations and review record": "Appendice tecnica: evidenze, calcoli e verifiche",
    "Chart data and calculation lineage": "Dati del grafico e riferimenti ai calcoli",
    "Month": "Mese",
    "Series": "Serie",
    "Calculation ID": "Riferimento di calcolo",
    "Before new financing": "Prima dei nuovi fondi",
    "After scheduled financing": "Dopo i fondi ipotizzati",
    "EBITDA by scenario": "EBITDA mensile per scenario",
    "Monthly cash": "Cassa mensile",
    "Calculation": "Calcolo",
    "Source observation": "Dato riportato dalla fonte",
    "Sum": "Somma",
    "Ratio": "Rapporto",
    "Draft for discussion": "Bozza per discussione",
    "Page": "Pagina",
}


def language(case: dict[str, Any]) -> str:
    """Use an explicit presentation language, never infer locale from client data."""
    return case.get("presentation", {}).get("language", "en")


def label(text: str, lang: str) -> str:
    from planning_french import FRENCH

    return {"it": ITALIAN, "fr": FRENCH}.get(lang, {}).get(text, text)


def format_unit(unit: str, lang: str) -> str:
    """Localize display units without changing the canonical calculation record."""
    if lang != "it":
        return unit
    if unit.endswith("/unit"):
        return unit.removesuffix("/unit") + "/unità"
    return {"unit": "unità", "units": "unità", "months": "mesi"}.get(unit, unit)


def format_number(
    value: Any, lang: str, decimals: int = 0, style: str = "number"
) -> str:
    amount = number(format(Decimal(str(value)), "f")) * (
        100 if style == "percent" else 1
    )
    rendered = f"{amount:,.{decimals}f}"
    if lang == "it":
        rendered = rendered.translate(str.maketrans({",": ".", ".": ","}))
    elif lang == "fr":
        rendered = rendered.translate(str.maketrans({",": "\u202f", ".": ","}))
    return rendered + ("%" if style == "percent" else "")


def _cell_value(cell: dict[str, Any], plan: dict[str, Any]) -> tuple[Decimal, str]:
    """Check exact proposed values against canonical figures or source observations."""
    if "observation_id" in cell:
        observations = {r["id"]: r for r in plan["case"]["observations"]}
        require(cell["observation_id"] in observations, "Unknown table observation")
        require(
            "operation" not in cell, "Observation cells cannot specify an operation"
        )
        row = observations[cell["observation_id"]]
        return number(row["value"]), row["unit"]
    ids = cell.get("calculation_ids", [])
    require(
        isinstance(ids, list) and bool(ids) and len(ids) == len(set(ids)),
        "Table calculation IDs must be unique and nonempty",
    )
    calcs = plan["calculations"]
    require(
        all(i in calcs and calcs[i]["value"] is not None for i in ids),
        "Table calculation unavailable",
    )
    rows = [calcs[i] for i in ids]
    require(len({r["unit"] for r in rows}) == 1, "Table calculation units differ")
    operation = cell.get("operation", "value")
    require(
        operation in {"value", "sum", "ratio", "difference"}, "Unknown table operation"
    )
    require(operation != "value" or len(rows) == 1, "Value cell needs one calculation")
    if operation == "ratio":
        require(
            len(rows) == 2 and number(rows[1]["value"]) != 0,
            "Ratio needs two values and a nonzero denominator",
        )
        return number(rows[0]["value"]) / number(rows[1]["value"]), "ratio"
    if operation == "difference":
        require(len(rows) >= 2, "Difference needs at least two values")
        return (
            number(rows[0]["value"])
            - sum((number(r["value"]) for r in rows[1:]), Decimal(0)),
            rows[0]["unit"],
        )
    return sum((number(r["value"]) for r in rows), Decimal(0)), rows[0]["unit"]


def validate_presentation(plan: dict[str, Any]) -> None:
    """Reject broken bindings and unsafe links, not provisional business judgment."""
    p = plan["case"].get("presentation", {})
    require(
        isinstance(p, dict)
        and set(p)
        <= {"language", "tables", "actions", "source_notes", "comparison_groups"},
        "Unexpected presentation fields",
    )
    require(language(plan["case"]) in {"en", "it", "fr"}, "Unsupported report language")
    from planning_assessment import SECTIONS

    # Blocked reports retain submitted bindings for diagnostic validation only;
    # their assessment, tables and actions are not rendered as conclusions.
    narrative = {
        n["id"]
        for n in (
            plan["case"]["narrative"]
            if plan["status"] == "blocked"
            else plan["accepted_narrative"]
        )
    }
    tables = indexed(p.get("tables", []), "presentation table")
    for table in tables.values():
        require(
            set(table)
            <= {
                "id",
                "title",
                "section",
                "headers",
                "rows",
                "caption_id",
                "comparison",
            },
            "Unexpected table fields",
        )
        require(
            table.get("section") in SECTIONS and bool(table.get("title")),
            "Table needs a title and business section",
        )
        headers = table.get("headers")
        require(
            isinstance(headers, list)
            and 1 <= len(headers) <= 8
            and all(isinstance(h, str) and h for h in headers),
            "Table requires readable column headings",
        )
        require(
            isinstance(table.get("rows"), list) and bool(table["rows"]),
            "Table requires rows",
        )
        require(
            table.get("caption_id") in narrative,
            "Table needs an available narrative explaining scope and assumptions",
        )
        for row in table["rows"]:
            require(
                isinstance(row, list) and len(row) == len(headers),
                "Table row width differs",
            )
            for cell in row:
                require(isinstance(cell, dict), "Table cells must be typed")
                if "text" in cell:
                    require(
                        set(cell) == {"text"} and isinstance(cell["text"], str),
                        "Text cells contain labels only",
                    )
                    continue
                require(
                    set(cell)
                    <= {
                        "observation_id",
                        "calculation_ids",
                        "operation",
                        "value",
                        "decimals",
                        "style",
                    },
                    "Unexpected numeric cell fields",
                )
                require(
                    ("observation_id" in cell) != ("calculation_ids" in cell),
                    "Choose one table figure source",
                )
                require(
                    isinstance(cell.get("decimals", 0), int)
                    and 0 <= cell.get("decimals", 0) <= 4,
                    "Invalid table precision",
                )
                require(
                    cell.get("style", "number") in {"number", "percent"},
                    "Unknown number style",
                )
                amount, unit = _cell_value(cell, plan)
                require(
                    "value" in cell and amount == number(cell["value"]),
                    "Table figure disagrees with its source calculations",
                )
                require(
                    cell.get("style") != "percent" or unit == "ratio",
                    "Percent formatting requires a ratio",
                )
        if "comparison" in table:
            comparison_rows(table, plan)
    from planning_interaction import validate_groups

    validate_groups(p, tables)
    require(isinstance(p.get("actions", []), list), "Actions must be a list")
    for action in p.get("actions", []):
        require(isinstance(action, dict), "Action must be an object")
        require(
            set(action) == {"action_id", "owner", "when", "criterion_id"},
            "Unexpected action fields",
        )
        require(
            action["action_id"] in narrative and action["criterion_id"] in narrative,
            "Action references unavailable narrative",
        )
        require(
            all(
                isinstance(action[k], str) and action[k].strip()
                for k in ("owner", "when")
            ),
            "Action needs a responsible role and timing",
        )
    sources = {s["id"] for s in plan["case"]["sources"]}
    require(isinstance(p.get("source_notes", []), list), "Source notes must be a list")
    for note in p.get("source_notes", []):
        require(isinstance(note, dict), "Source note must be an object")
        require(
            set(note) <= {"source_id", "claim", "locator", "url"},
            "Unexpected source note fields",
        )
        require(note.get("source_id") in sources, "Unknown cited source")
        require(
            all(
                isinstance(note.get(k), str) and note[k].strip()
                for k in ("claim", "locator")
            ),
            "Source note needs a claim and locator",
        )
        if "url" in note:
            require(isinstance(note["url"], str), "Source URL must be text")
            url = urlsplit(note["url"])
            require(
                url.scheme in {"https", "http"}
                and bool(url.hostname)
                and not url.username
                and not url.password,
                "Source URL must be an HTTP(S) reference without credentials",
            )


def comparison_rows(
    table: dict[str, Any], plan: dict[str, Any]
) -> tuple[list[dict[str, Any]], str]:
    """Compute both variances from explicitly selected, source-bound value columns.

    Exact arithmetic and unit checks are mechanical. The author owns period
    comparability, row meaning and whether an increase is favorable.
    """
    comparison = table["comparison"]
    require(
        isinstance(comparison, dict)
        and set(comparison)
        <= {
            "baseline_column",
            "comparison_column",
            "favorable_directions",
            "row_types",
        },
        "Unexpected comparison fields",
    )
    baseline_index = comparison.get("baseline_column")
    current_index = comparison.get("comparison_column")
    require(
        type(baseline_index) is int
        and type(current_index) is int
        and {baseline_index, current_index} == {1, 2}
        and len(table["headers"]) == 3,
        "Comparison needs a row label and two value columns",
    )
    directions = comparison.get(
        "favorable_directions", ["neutral"] * len(table["rows"])
    )
    row_types = comparison.get("row_types", ["detail"] * len(table["rows"]))
    require(
        isinstance(directions, list)
        and len(directions) == len(table["rows"])
        and all(d in {"higher", "lower", "neutral"} for d in directions),
        "Invalid comparison favorable directions",
    )
    require(
        isinstance(row_types, list)
        and len(row_types) == len(table["rows"])
        and all(t in {"detail", "subtotal", "total"} for t in row_types),
        "Invalid comparison row types",
    )
    rows = []
    units = set()
    for row, direction, row_type in zip(table["rows"], directions, row_types):
        require(
            "text" in row[0]
            and "text" not in row[baseline_index]
            and "text" not in row[current_index],
            "Comparison needs two available numeric values",
        )
        baseline, baseline_unit = _cell_value(row[baseline_index], plan)
        current, current_unit = _cell_value(row[current_index], plan)
        require(
            baseline_unit == current_unit == plan["case"]["reporting_currency"],
            "Financial comparison requires the same reporting currency",
        )
        units.add(baseline_unit)
        delta = current - baseline
        rows.append(
            {
                "row_label": row[0]["text"],
                "baseline_value": str(baseline),
                "comparison_value": str(current),
                "absolute_variance": str(delta),
                "relative_variance": (
                    str(delta / baseline * 100) if baseline > 0 else None
                ),
                "favorable_direction": direction,
                "row_type": row_type,
            }
        )
    require(len(units) == 1, "Comparison rows must share one currency")
    return rows, units.pop()


def render_tables(
    plan: dict[str, Any], section: str, render_paragraph: Callable[[str], str]
) -> str:
    """Render author-selected comparisons without a case-specific HTML wrapper."""
    from planning_report import _table

    lang = language(plan["case"])

    def render_table(
        table: dict[str, Any], scale_rows: list[dict[str, Any]] | None = None
    ) -> str:
        if "comparison" in table:
            from reporting_table import render_reporting_table

            rows, unit = comparison_rows(table, plan)
            comparison = table["comparison"]
            note = (
                "Scostamento = confronto − base. Percentuale sulla base; n/d con base zero o negativa. Colori secondo la convenzione della singola voce; grigio se non definita."
                if lang == "it"
                else label(
                    "Variance = comparison − baseline. Percent uses the baseline; n/a for zero or negative baselines. Colors follow each row's convention; gray when unspecified.",
                    lang,
                )
            )
            component = render_reporting_table(
                row_header=table["headers"][0],
                rows=rows,
                baseline_label=table["headers"][comparison["baseline_column"]],
                comparison_label=table["headers"][comparison["comparison_column"]],
                entity_label=plan["case"]["entity_name"],
                comparison_caption=table["title"],
                metric=unit,
                source_label=note,
                language=lang,
                fragment=True,
                row_label_width=220,
                scale_rows=scale_rows,
            )
            return f'<div id="table-{html.escape(table["id"], quote=True)}">{component}{render_paragraph(table["caption_id"])}</div>'
        rows = []
        for row in table["rows"]:
            cells = []
            for cell in row:
                if "text" in cell:
                    cells.append(cell["text"])
                else:
                    amount, _ = _cell_value(cell, plan)
                    cells.append(
                        format_number(
                            str(amount),
                            lang,
                            cell.get("decimals", 0),
                            cell.get("style", "number"),
                        )
                    )
            rows.append(cells)
        return f'<div class="decision-table" id="table-{html.escape(table["id"])}"><h3>{html.escape(table["title"])}</h3>{_table(table["headers"], rows)}{render_paragraph(table["caption_id"])}</div>'

    from planning_interaction import render_group

    presentation = plan["case"].get("presentation", {})
    tables = {t["id"]: t for t in presentation.get("tables", [])}
    groups = presentation.get("comparison_groups", [])
    grouped = {v["table_id"] for g in groups for v in g["views"]}
    output = [
        render_table(t)
        for t in tables.values()
        if t["section"] == section and t["id"] not in grouped
    ]
    for group in groups:
        if tables[group["views"][0]["table_id"]]["section"] == section:
            output.append(
                render_group(
                    group,
                    tables,
                    lang,
                    render_table,
                    lambda t: comparison_rows(t, plan)[0],
                )
            )
    return "".join(output)


def render_actions(plan: dict[str, Any], render_paragraph: Callable[[str], str]) -> str:
    lang = language(plan["case"])
    actions = plan["case"].get("presentation", {}).get("actions", [])
    if not actions:
        return ""
    heads = ["Action", "Owner", "When", "Evidence / decision criterion"]
    output = [
        '<div class="action-table table-scroll"><table><thead><tr>'
        + "".join(f"<th>{label(h, lang)}</th>" for h in heads)
        + "</tr></thead><tbody>"
    ]
    for a in actions:
        output.append(
            f'<tr><td>{render_paragraph(a["action_id"])}</td><td>{html.escape(a["owner"])}</td><td>{html.escape(a["when"])}</td><td>{render_paragraph(a["criterion_id"])}</td></tr>'
        )
    return "".join(output) + "</tbody></table></div>"


def render_sources(plan: dict[str, Any]) -> str:
    """Keep filenames, locators and used figure methods readable in standalone PDF."""
    from planning_report import _table

    case = plan["case"]
    lang = language(case)
    e = html.escape
    sources = {s["id"]: s for s in case["sources"]}
    notes = case.get("presentation", {}).get("source_notes", [])
    rows = []
    for source in sources.values():
        matching = [n for n in notes if n["source_id"] == source["id"]]
        if not matching:
            matching = [
                {
                    "claim": label("Not specified", lang),
                    "locator": label("Not specified", lang),
                }
            ]
        for note in matching:
            rows.append(
                [
                    source["id"]
                    + ": "
                    + source["path"].replace("\\", "/").rsplit("/", 1)[-1]
                    + " — "
                    + source["version"],
                    note["claim"],
                    note["locator"],
                ]
            )
    output = [
        f'<section id="reader-sources"><h2>{label("Sources and calculation references", lang)}</h2>',
        _table([label(h, lang) for h in ("Source", "Claim / use", "Location")], rows),
    ]
    for note in notes:
        if note.get("url"):
            output.append(
                f'<p><a href="{e(note["url"], quote=True)}">{e(note["claim"])}</a><br><span class="source-url">{e(note["url"])}</span></p>'
            )
    used: dict[str, dict[str, Any]] = {}
    for entry in plan["accepted_narrative"]:
        for claim in entry["claims"].values():
            if "calculation_id" in claim:
                used[claim["calculation_id"]] = plan["calculations"][
                    claim["calculation_id"]
                ]
    for table in case.get("presentation", {}).get("tables", []):
        for row in table["rows"]:
            for cell in row:
                for cid in cell.get("calculation_ids", []):
                    used[cid] = plan["calculations"][cid]
    bindings = []
    observations = {o["id"]: o for o in case["observations"]}
    for table in case.get("presentation", {}).get("tables", []):
        for index, row in enumerate(table["rows"], 1):
            for header, cell in zip(table["headers"], row):
                if "text" in cell:
                    continue
                if "observation_id" in cell:
                    obs = observations[cell["observation_id"]]
                    reference = (
                        label("Source observation", lang)
                        + ": "
                        + obs["id"]
                        + " ("
                        + obs["basis"]
                        + ")"
                    )
                else:
                    ids = cell["calculation_ids"]
                    reference = cell.get("operation", "value") + ": " + "; ".join(ids)
                    if len(ids) > 2 and cell.get("operation") == "sum":
                        reference = (
                            label("Sum", lang)
                            + ": "
                            + ids[0]
                            + " … "
                            + ids[-1]
                            + " ("
                            + str(len(ids))
                            + ")"
                        )
                bindings.append(
                    [
                        table["title"] + " / " + str(index) + " / " + header,
                        format_number(
                            cell["value"],
                            lang,
                            cell.get("decimals", 0),
                            cell.get("style", "number"),
                        ),
                        reference,
                    ]
                )
    if bindings:
        output.append(
            _table(
                [
                    label("Reference", lang),
                    label("Value", lang),
                    label("Calculation", lang),
                ],
                bindings,
            )
        )
    if used:
        # One method per metric, sharing scenario/period coverage rather than
        # repeating the same filenames and formula for every scenario.
        groups: dict[tuple[str, str, str], list[dict[str, Any]]] = {}
        for c in used.values():
            key = (c["metric"], c["formula"], ", ".join(c["source_ids"]))
            groups.setdefault(key, []).append(c)
        methods = []
        for (metric, formula, ids), values in groups.items():
            periods = sorted({v["period"] for v in values if v["period"] is not None})
            scenarios = ", ".join(sorted({v["scenario"] for v in values}))
            methods.append(
                [
                    metric + " / " + scenarios,
                    (
                        periods[0] + " — " + periods[-1]
                        if periods
                        else label("Undated operating period", lang)
                    ),
                    formula,
                    ids,
                ]
            )
        output.append(
            _table(
                [
                    label("Calculation", lang),
                    label("When", lang),
                    label("Reference", lang),
                    label("Source", lang),
                ],
                methods,
            )
        )
    return "".join(output) + "</section>"

SHA-256: f9ccfc89c1ecf912e22b989b033bd7abd95964b26c7b5706e213f89464576e3a