platform/harness/views.py

"""harness/views.py — saved views + spawned dashboards: the store and the RE-RUNNER (OM-3 viewer).

The Analyst's save_view/compose_dashboard tools persist chart SPECS + their semantic QUERY (never
rows) into data/store/views.json. This module is the read side: load the specs, re-execute each
view's governed query against the tenant store at render time, and hand the page a fresh artifact
dict in exactly the shape `_render_analyst_artifact` already renders. The viewer therefore shows
LIVE (store-fresh) numbers, not the numbers from whenever the view was saved — same discipline as
every dashboard (a saved view is a query, not a snapshot).

Legacy note: views saved before 2026-07-11 carry no 'query' (only a session-scoped result_id) and
cannot be re-run — run_view raises a readable error the page surfaces per view.
"""
import json
import time

import harness.semantic as SEM
from harness.tools import VIEWS_PATH

_QUERY_KEYS = ("topic", "measures", "group_by", "grain", "date_from", "date_to",
               "team_id", "filters", "sort", "limit", "exclude_services")


def load():
    if VIEWS_PATH.exists():
        return json.loads(VIEWS_PATH.read_text(encoding="utf-8"))
    return {"views": {}, "dashboards": {}}


def _save(d):
    VIEWS_PATH.parent.mkdir(parents=True, exist_ok=True)
    VIEWS_PATH.write_text(json.dumps(d, indent=1), encoding="utf-8")


def views():
    return load().get("views", {})


def dashboards():
    return load().get("dashboards", {})


def delete_view(name):
    d = load()
    d.get("views", {}).pop(name, None)
    for dash in d.get("dashboards", {}).values():
        dash["views"] = [v for v in dash.get("views", []) if v != name]
    _save(d)


def delete_dashboard(name):
    d = load()
    d.get("dashboards", {}).pop(name, None)
    _save(d)


# ------------------------------------------------------------------ UPDATE (owner directive
# 2026-07-12: the Analyst can CREATE **and UPDATE** each workbook in the workspace)

_QUERY_PATCH_KEYS = ("measures", "group_by", "grain", "date_from", "date_to", "team_id",
                     "filters", "sort", "limit", "exclude_services", "topic")
_SPEC_PATCH_KEYS = ("title", "kind", "x", "y", "series", "metric",
                    "y2", "size", "value", "facet", "columns", "transforms")


def update_view(name, changes):
    """Patch a saved view's chart spec and/or query, VALIDATED BY RE-EXECUTION before saving —
    an update that cannot run does not land. `changes` may contain spec keys (title/kind/x/y/
    series/metric) and/or a `query` dict of query-key patches (a None value REMOVES the key)."""
    d = load()
    v = d.get("views", {}).get(name)
    if not v:
        raise SEM.ModelError(f"unknown view {name!r} (use list_workspace)")
    spec = dict(v.get("chart") or {})
    q = dict(spec.get("query") or {})
    if not q:
        raise SEM.ModelError(f"view {name!r} carries no query — rebuild it instead")
    changes = dict(changes or {})
    qpatch = changes.pop("query", None) or {}
    bad = [k for k in changes if k not in _SPEC_PATCH_KEYS]
    bad += [k for k in qpatch if k not in _QUERY_PATCH_KEYS]
    if bad:
        raise SEM.ModelError(f"unknown patch keys {bad} (spec: {list(_SPEC_PATCH_KEYS)}; "
                             f"query: {list(_QUERY_PATCH_KEYS)})")
    for k, val in qpatch.items():
        if val is None:
            q.pop(k, None)
        else:
            q[k] = val
    res = _run_query(q)                                  # validation: it must RUN
    new_spec = {**spec, **changes, "query": q}
    rows = res["rows"]
    if new_spec.get("transforms"):                       # the transform chain must replay too
        import harness.transforms as TR
        rows, _ = TR.apply({**res, "query": q}, new_spec["transforms"], run_query=_run_query)
    cols = set(rows[0]) if rows else set()
    for ref in ("x", "y", "series", "y2", "size", "value", "facet"):
        if new_spec.get(ref) and cols and new_spec[ref] not in cols:
            raise SEM.ModelError(f"{ref}={new_spec[ref]!r} not in the patched result columns "
                                 f"{sorted(cols)} — patch {ref} too")
    d["views"][name] = {"chart": new_spec,
                        "saved_at": time.strftime("%Y-%m-%d %H:%M")}
    _save(d)
    return {"updated": name, "row_count": res["row_count"],
            "query": q, "note": "patched query re-ran successfully before saving"}


def update_dashboard(name, views_list=None, new_name=None):
    """Update a workbook (dashboard): recompose its views (order = display order) and/or rename
    it. Renames follow through to scheduled reports that reference it."""
    d = load()
    dash = d.get("dashboards", {}).get(name)
    if dash is None:
        raise SEM.ModelError(f"unknown dashboard {name!r} (use list_workspace)")
    if views_list is not None:
        missing = [v for v in views_list if v not in d.get("views", {})]
        if missing:
            raise SEM.ModelError(f"unknown views {missing} — save_view them first")
        dash["views"] = list(views_list)
    dash["updated_at"] = time.strftime("%Y-%m-%d %H:%M")
    if new_name and new_name != name:
        if new_name in d.get("dashboards", {}) or new_name in d.get("views", {}):
            raise SEM.ModelError(f"{new_name!r} already exists — pick another name")
        d["dashboards"][new_name] = d["dashboards"].pop(name)
        try:                                             # follow the rename into schedules
            import harness.routines as R
            rt = R.load()
            if name in rt.get("reports", {}):
                rt["reports"][new_name] = rt["reports"].pop(name)
                R._save(rt)
        except Exception:
            pass
        name = new_name
    _save(d)
    return {"dashboard": name, "views": d["dashboards"][name]["views"]}


def _run_query(q):
    return SEM.store_query(**{k: q.get(k) for k in _QUERY_KEYS if q.get(k) is not None})


def run_view(name):
    """Re-execute one saved view's governed query — then REPLAY its recorded transform chain —
    and return a fresh artifact dict ({'chart': {..., 'rows': [...]}}, {'table': {...}} or
    {'kpi': {...}}) for the house renderer."""
    v = views().get(name)
    if not v:
        raise SEM.ModelError(f"unknown view {name!r}")
    spec = v.get("chart") or {}
    q = spec.get("query")
    if not q:
        raise SEM.ModelError(f"view {name!r} was saved without its query (pre-2026-07-11) — "
                             "ask the Analyst to rebuild and re-save it")
    res = _run_query(q)
    if spec.get("kind") == "kpi" or (spec.get("metric") and spec.get("kind") != "table"):
        metric = spec.get("metric") or spec.get("y")       # KPI-shaped view
        val = (res["rows"][0].get(metric) if res["rows"] else None)
        kpi = {"metric": metric, "value": val or 0}
        cq = spec.get("compare_query")
        if cq:
            prev_rows = _run_query(cq)["rows"]
            prev = prev_rows[0].get(metric) if prev_rows else None
            if prev:
                kpi["delta_pct"] = ((val or 0) - prev) / abs(prev)
        return {"kpi": kpi}
    rows = res["rows"]
    if spec.get("transforms"):                             # a saved view replays its transforms
        import harness.transforms as TR
        rows, _ = TR.apply({**res, "query": q}, spec["transforms"], run_query=_run_query)
    if spec.get("kind") == "table":
        return {"table": {**{k: spec.get(k) for k in ("kind", "title", "columns")},
                          "rows": rows}}
    return {"chart": {**{k: spec.get(k) for k in ("kind", "x", "y", "series", "title",
                                                  "y2", "size", "value", "facet")},
                      "rows": rows}}


def run_dashboard(name):
    """All of a dashboard's views, freshly re-queried. Returns [(view_name, artifact|None, err)]."""
    dash = dashboards().get(name)
    if dash is None:
        raise SEM.ModelError(f"unknown dashboard {name!r}")
    out = []
    for vn in dash.get("views", []):
        try:
            out.append((vn, run_view(vn), None))
        except Exception as e:
            out.append((vn, None, str(e)[:200]))
    return out


def store_freshness():
    """Newest sync checkpoint across entities — the staleness footer ('data as of …')."""
    try:
        import harness.datastore as DS
        ts = [s.get("updated") for s in DS.status().values() if s.get("updated")]
        return max(ts) if ts else None
    except Exception:
        return None