"""Measures computed from summed inputs (2026-11-09, docs/plans/RATIO_MEASURES_REQUIREMENTS.md).

Over several periods or units the plant measures PA as ONE ratio of summed hours,
`100 * (Σ total hours - Σ breakdown) / Σ total hours` (STD_2026_006_MNO p. 4, EMA's
`PAPerformanceMonthly.yml`), not as the average of the per-month `pa` values. The
query IR can compute that (`AggSelect.computed`); this module decides WHEN, from data:

1. `bind_computed_measures` (read time, M3): an approved measure whose formula entry
   carries a `compute` block, e.g.
       {"column": "pa",
        "numerator": [["+", "total_inpr_hour"], ["-", "breakdown_duration_total"]],
        "denominator": [["+", "total_inpr_hour"]],
        "scale": 100, "label": {"id": "PA dari total jam", "en": "PA from summed hours"}}
   marks every catalog table holding all those columns (by NAME, numeric terms only)
   with `Column.measure_method = {"kind": "ratio", ...}` in column ids. An approved
   measure WITH a formula but no usable block marks a same-named column
   `{"kind": "plain"}`, so the answer can say it is a plain average (U9).
2. `mark_computed` (plan validation, M4): every `avg` of a ratio-marked column gets the
   `computed` spec. The planner keeps writing `avg(pa)`; it is never taught the form.
3. `wants_plain_average` (M8): the user explicitly asked for a plain average, so the
   rewrite is skipped for the turn (U10).

The definition lives once, on the approved formula entry (one home, §5.22); no prompt
text. Every function here is fail-open: a bad block or a slip leaves the plain `avg`,
which is today's behaviour.
"""

from __future__ import annotations

import re
from typing import Any

from src.middlewares.logging import get_logger

logger = get_logger("knowledge_computed")

_NUMERIC = frozenset({"int", "decimal"})
_SIGNS = frozenset({"+", "-"})
_SCALES = frozenset({1, 100})
_MAX_TERMS = 4


def _terms(raw: Any) -> list[tuple[str, str]] | None:
    """[[sign, column], ...] → [(sign, column_name)], or None when malformed."""
    if not isinstance(raw, list) or not 1 <= len(raw) <= _MAX_TERMS:
        return None
    out: list[tuple[str, str]] = []
    for item in raw:
        if not (isinstance(item, list | tuple) and len(item) == 2):
            return None
        sign, name = item
        if sign not in _SIGNS or not isinstance(name, str) or not name.strip():
            return None
        out.append((sign, name.strip()))
    return out


def _parse(spec: Any) -> dict | None:
    if not isinstance(spec, dict):
        return None
    column = spec.get("column")
    numerator, denominator = _terms(spec.get("numerator")), _terms(spec.get("denominator"))
    try:
        scale = int(spec.get("scale", 1))
    except (TypeError, ValueError):
        return None
    label = spec.get("label") if isinstance(spec.get("label"), dict) else {}
    if not isinstance(column, str) or not numerator or not denominator or scale not in _SCALES:
        return None
    return {"column": column.strip(), "numerator": numerator, "denominator": denominator,
            "scale": scale, "label": {k: str(v) for k, v in label.items() if v}}


def bind_computed_measures(measures: list[Any], catalog: Any) -> int:
    """Mark catalog columns with how their measure is computed. Returns tables bound.

    In place on the catalog's Column objects (catalogs are rebuilt per read, so this
    never leaks across requests). Idempotent. Never raises.
    """
    try:
        return _bind(measures, catalog)
    except Exception as e:  # an enrichment must never cost the answer
        logger.warning(
            "computed_measures_bind_failed",
            extra={"error": repr(e), "degraded_seam": "computed_measures"},
        )
        return 0


def _bind(measures: list[Any], catalog: Any) -> int:
    ratios: list[tuple[str, dict]] = []
    plain_names: dict[str, str] = {}
    for m in measures or []:
        if not getattr(m, "approved", False):
            continue
        spec = _parse(getattr(m, "compute", None))
        if spec is not None:
            ratios.append((m.name, spec))
        elif getattr(m, "compute", None) is not None:
            logger.warning("computed_measure_spec_invalid", extra={"measure": m.name})
        if getattr(m, "formula_latex", None):
            for form in [m.name, *getattr(m, "surface_forms", [])]:
                plain_names.setdefault(form.strip().casefold(), m.name)

    bound = 0
    for source in getattr(catalog, "sources", []) or []:
        for table in getattr(source, "tables", []) or []:
            by_name = {c.name.strip().casefold(): c for c in table.columns}
            ratio_cols: set[str] = set()
            for measure, spec in ratios:
                target = by_name.get(spec["column"].casefold())
                terms = [*spec["numerator"], *spec["denominator"]]
                cols = [by_name.get(name.casefold()) for _, name in terms]
                if target is None or any(c is None or c.data_type not in _NUMERIC for c in cols):
                    continue
                ids = iter(c.column_id for c in cols)
                num = [{"column_id": next(ids), "sign": s} for s, _ in spec["numerator"]]
                den = [{"column_id": next(ids), "sign": s} for s, _ in spec["denominator"]]
                target.measure_method = {
                    "kind": "ratio",
                    "measure": measure,
                    "numerator": num,
                    "denominator": den,
                    "scale": spec["scale"],
                    "label": spec["label"],
                }
                ratio_cols.add(target.column_id)
                bound += 1
            for name, column in by_name.items():
                measure = plain_names.get(name)
                if measure and column.column_id not in ratio_cols:
                    column.measure_method = {"kind": "plain", "measure": measure}
    logger.info("computed_measures_bound", extra={"n_bound": bound, "n_defined": len(ratios)})
    return bound


def mark_computed(ir: Any, catalog: Any) -> list[str]:
    """Give every `avg` of a ratio-marked column its `computed` spec. In place.

    Returns the aliases (or column ids) it marked. Leaves sum/min/max/count, raw
    columns and an already-computed select alone.
    """
    from src.query.ir.models import AggSelect, ComputedRatio

    columns = {
        c.column_id: c
        for s in getattr(catalog, "sources", []) or []
        if s.source_id == ir.source_id
        for t in s.tables
        for c in t.columns
    }
    marked: list[str] = []
    for item in ir.select:
        if not isinstance(item, AggSelect) or item.fn != "avg" or item.computed is not None:
            continue
        method = getattr(columns.get(item.column_id), "measure_method", None) or {}
        if method.get("kind") != "ratio":
            continue
        item.computed = ComputedRatio(
            numerator=method["numerator"],
            denominator=method["denominator"],
            scale=method["scale"],
            label=method.get("label") or {},
        )
        marked.append(item.alias or item.column_id)
    return marked


# Explicit requests for the plain average (U10). Matched on the user's words, EN + ID.
_PLAIN_AVERAGE = re.compile(
    r"rata[\s-]?rata\s+(sederhana|biasa|aritmatika|aritmetika)"
    r"|bukan\s+(dari\s+)?(total|jumlah)\s+jam"
    r"|tanpa\s+bobot|tidak\s+(di)?bobot"
    r"|simple\s+(average|mean)|plain\s+(average|mean)|arithmetic\s+(average|mean)"
    r"|unweighted|not\s+weighted|not\s+from\s+(summed|total)\s+hours",
    re.IGNORECASE,
)


def wants_plain_average(*texts: str | None) -> bool:
    """True when the user's own words ask for a plain (unweighted) average."""
    return any(t and _PLAIN_AVERAGE.search(t) for t in texts)