ui/app.py

"""AI usage evaluator — UI shell (Gradio). Run with:  python -m ui.app

One gr.Blocks, three screens as visibility-toggled columns (INTAKE -> PROCESSING -> RESULT).
The processing generator reveals REAL parsed facts (Task 2) one at a time, then the DummyScorer
(Task 3) derives the card from the upload. No URL routes — result depends on the upload, which
only lives in this session's state.
"""
from __future__ import annotations

import threading
import time

import gradio as gr

from .data import get_stub_card
from .theme import base_theme, HEAD, CSS, DRAW_RADAR_JS
from .components.card import render_card, render_accordion_body
from .scoring import get_scorer, score_to_card
from .screens import intake, processing as P, result

HIDE = gr.update(visible=False)
SHOW = gr.update(visible=True)
NOOP = gr.update()


def _provenance(scorer) -> str:
    """Honesty-tag provenance from the scorer that actually produced the result."""
    if type(scorer).__name__ != "ObservableScorer":
        return "placeholder"
    try:
        from .scoring.observable import _lora_axes
        return "real-lora" if _lora_axes() else "real-base"
    except Exception:
        return "real-base"


def _frame(lines, *, intake_v, proc_v, result_v, card=NOOP, bodies=None):
    bodies = bodies if bodies is not None else [NOOP] * 5
    return (intake_v, proc_v, result_v, P.wrap(lines), card, *bodies)


def run(name, parsed, rm_flag):
    """Generator: staged reveal of real facts, then the scored result card."""
    reduce = (rm_flag == "reduce")

    def nap(t):
        if not reduce:
            time.sleep(t)

    lines: list[str] = []
    yield _frame(lines, intake_v=HIDE, proc_v=SHOW, result_v=HIDE)

    if parsed is None:  # safety net; button is gated on a valid parse
        lines.append(P.fact("Could not read this export — go back and try another file.", muted=True))
        yield _frame(lines, intake_v=SHOW, proc_v=HIDE, result_v=HIDE)
        return

    nap(0.4)
    lines.append(P.fact(f"{parsed.source.title()} export detected"))
    yield _frame(lines, intake_v=HIDE, proc_v=SHOW, result_v=HIDE)
    nap(1.0)

    total = parsed.turn_count
    slot = len(lines)
    lines.append(P.fact_num(0, "turns analyzed"))
    if reduce:
        lines[slot] = P.fact_num(total, "turns analyzed")
        yield _frame(lines, intake_v=HIDE, proc_v=SHOW, result_v=HIDE)
    else:
        for f in (0.3, 0.6, 0.85, 1.0):
            lines[slot] = P.fact_num(int(total * f), "turns analyzed")
            yield _frame(lines, intake_v=HIDE, proc_v=SHOW, result_v=HIDE)
            nap(0.25)
    nap(0.5)

    dr = parsed.date_range()
    if dr:
        lines.append(P.fact(dr))
        yield _frame(lines, intake_v=HIDE, proc_v=SHOW, result_v=HIDE)
        nap(1.0)

    lines.append(P.lang_block(parsed.english_turn_count, parsed.other_turn_count))
    yield _frame(lines, intake_v=HIDE, proc_v=SHOW, result_v=HIDE)
    nap(1.0)

    busy = parsed.busiest_slot()
    if busy:
        lines.append(P.fact(busy))
        yield _frame(lines, intake_v=HIDE, proc_v=SHOW, result_v=HIDE)
        nap(1.0)

    lines.append(P.fact("Warming up models…", muted=True))
    yield _frame(lines, intake_v=HIDE, proc_v=SHOW, result_v=HIDE)
    nap(0.6)

    # Real backend when configured (OPENBMB_* set), else the heuristic placeholder. Scoring the whole
    # history is thousands of model calls, so run it in a worker thread and stream a REAL progress bar
    # (ticked per completed call) rather than freezing the screen. A live failure degrades to the dummy.
    scorer = get_scorer()
    # HONEST FAILURE: never show fake "demo" scores. If there's no real backend, or scoring errors out,
    # surface a clear message — not a DummyScorer card.
    if type(scorer).__name__ != "ObservableScorer":
        lines.append(P.error("Scoring backend not configured. Set OPENBMB_BASE_URL / OPENBMB_TOKEN "
                             "(see DEPLOY.md), then reload and try again."))
        yield _frame(lines, intake_v=HIDE, proc_v=SHOW, result_v=HIDE)
        return

    from prompt_card.observable_pipeline import Progress
    progress = Progress()
    box: dict = {}

    def _work():
        try:
            box["result"] = scorer.score(parsed, progress=progress)
        except Exception as e:  # captured, surfaced as a real error frame below
            box["error"] = e

    bar = len(lines)
    lines.append(P.progress_bar(0, progress.total))
    yield _frame(lines, intake_v=HIDE, proc_v=SHOW, result_v=HIDE)

    th = threading.Thread(target=_work, daemon=True)
    th.start()
    while th.is_alive():
        done, total = progress.snapshot()
        lines[bar] = P.progress_bar(done, total)
        yield _frame(lines, intake_v=HIDE, proc_v=SHOW, result_v=HIDE)
        time.sleep(0.4)
    th.join()

    if "error" in box:
        e = box["error"]
        lines[bar] = P.error(f"Scoring failed — {type(e).__name__}: {str(e)[:240]}")
        lines.append(P.fact("No demo scores are shown. Check the model endpoint / secrets, then reload "
                            "and try again.", muted=True))
        yield _frame(lines, intake_v=HIDE, proc_v=SHOW, result_v=HIDE)
        return

    done, total = progress.snapshot()
    lines[bar] = P.progress_bar(total or done, total or done)  # snap to 100%

    result = box["result"]
    card = score_to_card(result, name)
    card.provenance = _provenance(scorer)
    card.single_conversation = (getattr(parsed, "source", "") == "paste")
    body_updates = [gr.update(value=render_accordion_body(a, card.provenance)) for a in card.axes]
    yield _frame(lines, intake_v=HIDE, proc_v=HIDE, result_v=SHOW,
                 card=gr.update(value=render_card(card)), bodies=body_updates)


def run_paste(name, pasted, rm_flag):
    """Single-conversation quick path: parse the pasted transcript, then reuse the staged reveal +
    scorer. Invalid paste shows an error and stays on the intake screen."""
    from .parsing import parse_paste, ParseError, is_share_url, fetch_share
    try:
        text = pasted or ""
        parsed = fetch_share(text) if is_share_url(text) else parse_paste(text)
    except ParseError as e:
        yield _frame([P.error(str(e))], intake_v=HIDE, proc_v=SHOW, result_v=HIDE)
        return
    yield from run(name, parsed, rm_flag)


def build_demo() -> gr.Blocks:
    # Gradio 6 moved theme/css/head from the Blocks constructor to launch(); see launch_app().
    with gr.Blocks(title="AI usage card") as demo:
        parsed_state = gr.State(None)
        rm_flag = gr.Textbox("motion", visible=False, elem_id="omc-rm-flag")

        intake_col, ic = intake.build()
        proc_col, pc = P.build()
        result_col, rc = result.build(get_stub_card(""))

        outputs = [intake_col, proc_col, result_col, pc["log"], rc["card_html"], *rc["bodies"]]

        ic["file"].change(intake.validate, inputs=[ic["file"]],
                          outputs=[ic["analyze"], ic["error"], parsed_state])
        ev = ic["analyze"].click(run, inputs=[ic["name"], parsed_state, rm_flag], outputs=outputs)
        ev.then(fn=None, inputs=None, outputs=None, js=DRAW_RADAR_JS)
        # Paste path: single-conversation quick analysis (parses on click; no file needed).
        ev_p = ic["paste_btn"].click(run_paste, inputs=[ic["name"], ic["paste"], rm_flag], outputs=outputs)
        ev_p.then(fn=None, inputs=None, outputs=None, js=DRAW_RADAR_JS)

        # Detect prefers-reduced-motion on load so the reveal can show instantly when requested.
        demo.load(fn=None, inputs=None, outputs=[rm_flag],
                  js="() => (window.matchMedia && window.matchMedia('(prefers-reduced-motion: reduce)').matches) ? 'reduce' : 'motion'")
    return demo


def _load_local_env():
    """Dev convenience so `python -m ui.app` uses the REAL scorer with no manual exports: load
    eval/.secrets.env (gitignored) into the environment for any keys not already set (a real shell
    export or HF Space secret always wins). No-op when the file is absent (e.g. on the Space, where
    the same vars are set as Space secrets). This is why a plain run was showing the heuristic
    placeholder — without OPENBMB_* set, get_scorer() falls back to DummyScorer."""
    import os
    from pathlib import Path
    f = Path(__file__).resolve().parent.parent / "eval" / ".secrets.env"
    try:
        text = f.read_text()
    except OSError:
        return
    for raw in text.splitlines():
        line = raw.strip()
        if not line or line.startswith("#") or "=" not in line:
            continue
        k, _, v = line.partition("=")
        k, v = k.strip(), v.strip()
        if k and k not in os.environ:
            os.environ[k] = v


def launch_app(**kwargs):
    """Launch with the visual config Gradio 6 expects on launch() (theme/css/head)."""
    _load_local_env()
    from .scoring import get_scorer
    s = get_scorer()
    if type(s).__name__ == "ObservableScorer":
        from .scoring.observable import _lora_axes
        print(f"[ui] scoring backend: REAL MiniCPM ({'LoRA hybrid' if _lora_axes() else 'base'}) "
              f"@ {__import__('os').environ.get('OPENBMB_BASE_URL')}", flush=True)
    else:
        print("[ui] scoring backend: DummyScorer (heuristic placeholder) — no OPENBMB_* configured. "
              "Set OPENBMB_BASE_URL/OPENBMB_TOKEN (or add them to eval/.secrets.env) for real scores.",
              flush=True)
    return build_demo().launch(theme=base_theme(), css=CSS, head=HEAD, **kwargs)


if __name__ == "__main__":
    launch_app()