Blame for app.py

#!/usr/bin/env python3
"""Hugging Face Space Gradio client for the public The Answer v2 API (https://api.dadler.app/theanswer/ask).

Layout follows the oc debug playground (status, answer, claims table with verified
column, episode triage table, warnings); styling follows the 2026-09-28 redesign.
"""

from __future__ import annotations

import os
import threading
import time
from pathlib import Path

import gradio as gr
import httpx
import yaml

from render import render_notice, render_result
from textproc import DisplayOpts
from view_model import map_api_response

_STR_LIST = "str_list"

_REQUIRED: list[tuple[str, object]] = [
    ("api_url", str),
    ("local_server_name", str),
    ("local_server_port", int),
    ("analytics_enabled", bool),
    ("font_family", str),
    ("font_css_url", str),
    ("font_fallbacks", _STR_LIST),
    ("asks_per_min", int),
    ("rate_window_s", int),
    ("max_question_chars", int),
    ("http_timeout_s", float),
    ("concurrency_limit", int),
    ("session_prune_threshold", int),
    ("display.max_quote_chars", int),
    ("display.ellipsis", str),
    ("display.min_term_chars", int),
    ("display.min_stem_chars", int),
    ("display.max_prefix_chars", int),
    ("display.stopwords", _STR_LIST),
    ("display.prefixes", _STR_LIST),
    ("ui.title", str),
    ("ui.title_markdown", str),
    ("ui.theme", str),
    ("ui.primary_hue", str),
    ("ui.secondary_hue", str),
    ("ui.neutral_hue", str),
    ("ui.label_question", str),
    ("ui.label_status", str),
    ("ui.label_answer", str),
    ("ui.label_claims", str),
    ("ui.label_triage", str),
    ("ui.label_warnings", str),
    ("ui.warnings_note", str),
    ("ui.claims_headers", _STR_LIST),
    ("ui.triage_headers", _STR_LIST),
    ("ui.verified_badge", str),
    ("ui.verified_title", str),
    ("ui.unverified_badge", str),
    ("ui.unverified_title", str),
    ("ui.status_labels.answered", str),
    ("ui.status_labels.partial", str),
    ("ui.status_labels.not_in_podcast", str),
    ("ui.status_labels.false_premise", str),
    ("ui.status_labels.refused", str),
    ("ui.status_labels.error", str),
    ("ui.triage_labels.answers", str),
    ("ui.triage_labels.related", str),
    ("ui.triage_labels.unrelated", str),
    ("ui.button", str),
    ("ui.rtl", bool),
    ("ui.question_lines", int),
    ("ui.question_max_lines", int),
    ("ui.examples_auto_submit", bool),
    ("ui.examples", _STR_LIST),
    ("ui.msg_empty", str),
    ("ui.msg_hint", str),
    ("ui.footer_html", str),
    ("ui.msg_too_long", str),
    ("ui.msg_rate", str),
    ("ui.msg_auth", str),
    ("ui.msg_down", str),
    ("ui.msg_network", str),
    ("ui.msg_unconfigured", str),
    ("ui.msg_503", str),
    ("ui.msg_refusal", str),
    ("ui.missing_title", str),
    ("ui.head_html", str),
    ("ui.css", str),
]


class SpaceConfigError(RuntimeError):
    def __init__(self, path: Path, problems: list[str]) -> None:
        self.path = path
        self.problems = problems
        lines = "\n".join(f"- {item}" for item in problems)
        super().__init__(f"Invalid Space config {path}:\n{lines}")


def _dig(data: dict, dotted: str):
    cur = data
    for part in dotted.split("."):
        if not isinstance(cur, dict) or part not in cur or cur[part] is None:
            return False, None
        cur = cur[part]
    return True, cur


def _type_ok(value: object, spec: object) -> bool:
    if spec is bool:
        return isinstance(value, bool)
    if spec is int:
        return isinstance(value, int) and not isinstance(value, bool)
    if spec is float:
        return isinstance(value, (int, float)) and not isinstance(value, bool)
    if spec is str:
        return isinstance(value, str) and bool(value.strip())
    if spec == _STR_LIST:
        return (
            isinstance(value, list)
            and len(value) > 0
            and all(isinstance(item, str) and item.strip() for item in value)
        )
    return False


def _type_name(spec: object) -> str:
    if spec is bool:
        return "bool"
    if spec is int:
        return "int"
    if spec is float:
        return "float"
    if spec is str:
        return "str"
    if spec == _STR_LIST:
        return "list[str]"
    return str(spec)


def _semantic_problems(data: dict) -> list[str]:
    problems: list[str] = []
    if len(data["ui"]["claims_headers"]) != 6:
        problems.append("ui.claims_headers must have 6 entries (claim, quote, verified, episode, date, time)")
    if len(data["ui"]["triage_headers"]) != 4:
        problems.append("ui.triage_headers must have 4 entries (episode, date, label, reason)")
    if str(data["font_family"]) not in str(data["ui"]["css"]):
        problems.append("ui.css does not reference font_family")
    if str(data["font_css_url"]) not in str(data["ui"]["head_html"]):
        problems.append("ui.head_html does not include font_css_url")
    question_lines = int(data["ui"]["question_lines"])
    question_max_lines = int(data["ui"]["question_max_lines"])
    if question_lines != 1:
        problems.append(
            "ui.question_lines must be 1 (Gradio submits on Enter only when lines is 1)"
        )
    if question_max_lines <= question_lines:
        problems.append("ui.question_max_lines must be greater than ui.question_lines")
    too_long = str(data["ui"]["msg_too_long"])
    if "{max_chars}" not in too_long:
        problems.append("ui.msg_too_long must contain {max_chars}")
    else:
        try:
            rendered = too_long.format(max_chars=int(data["max_question_chars"]))
        except (KeyError, ValueError, IndexError) as exc:
            problems.append(f"ui.msg_too_long format error: {exc}")
        else:
            if str(int(data["max_question_chars"])) not in rendered:
                problems.append("ui.msg_too_long did not include max_question_chars")
    if int(data["display"]["max_quote_chars"]) < 1:
        problems.append("display.max_quote_chars must be >= 1")
    if int(data["display"]["min_term_chars"]) < 1:
        problems.append("display.min_term_chars must be >= 1")
    if int(data["display"]["min_stem_chars"]) < 1:
        problems.append("display.min_stem_chars must be >= 1")
    if int(data["display"]["max_prefix_chars"]) < 1:
        problems.append("display.max_prefix_chars must be >= 1")
    port = int(data["local_server_port"])
    if port < 1 or port > 65535:
        problems.append("local_server_port out of range")
    theme_name = str(data["ui"]["theme"])
    theme_cls = getattr(gr.themes, theme_name, None)
    if not isinstance(theme_cls, type):
        problems.append(f"unknown theme ui.theme={theme_name}")
        return problems
    try:
        theme_cls(
            primary_hue=str(data["ui"]["primary_hue"]),
            secondary_hue=str(data["ui"]["secondary_hue"]),
            neutral_hue=str(data["ui"]["neutral_hue"]),
            font=[str(data["font_family"]), *[str(item) for item in data["font_fallbacks"]]],
        )
    except Exception as exc:
        problems.append(f"invalid theme settings for ui.theme={theme_name}: {exc}")
    return problems


def load_space_config(path: Path | None = None) -> dict:
    """Load space/config.yaml. SPACE_CONFIG overrides the path. No tunable defaults."""
    if path is None:
        raw = (os.environ.get("SPACE_CONFIG") or "").strip()
        path = Path(raw).expanduser() if raw else Path(__file__).resolve().parent / "config.yaml"
    if not path.is_file():
        raise SpaceConfigError(path, [f"file not found: {path}"])
    try:
        data = yaml.safe_load(path.read_text(encoding="utf-8"))
    except yaml.YAMLError as exc:
        raise SpaceConfigError(path, [f"YAML parse error: {exc}"]) from exc
    if not isinstance(data, dict):
        raise SpaceConfigError(path, ["root document must be a mapping"])
    problems: list[str] = []
    for dotted, spec in _REQUIRED:
        ok, value = _dig(data, dotted)
        if not ok:
            problems.append(f"missing key {dotted}")
        elif not _type_ok(value, spec):
            problems.append(
                f"invalid type for {dotted}: expected {_type_name(spec)}, "
                f"got {type(value).__name__}"
            )
    if not problems:
        problems.extend(_semantic_problems(data))
    if problems:
        raise SpaceConfigError(path, problems)
    return data


def build_theme(cfg: dict):
    ui = cfg["ui"]
    theme_cls = getattr(gr.themes, str(ui["theme"]))
    return theme_cls(
        primary_hue=str(ui["primary_hue"]),
        secondary_hue=str(ui["secondary_hue"]),
        neutral_hue=str(ui["neutral_hue"]),
        font=[str(cfg["font_family"]), *[str(item) for item in cfg["font_fallbacks"]]],
    )


def display_opts(cfg: dict) -> DisplayOpts:
    display = cfg["display"]
    return DisplayOpts(
        max_quote_chars=int(display["max_quote_chars"]),
        ellipsis=str(display["ellipsis"]),
        min_term_chars=int(display["min_term_chars"]),
        min_stem_chars=int(display["min_stem_chars"]),
        max_prefix_chars=int(display["max_prefix_chars"]),
        stopwords=frozenset(str(item).casefold() for item in display["stopwords"]),
        prefixes=tuple(str(item) for item in display["prefixes"]),
    )


_VIEW_KEYS = (
    "label_status", "label_answer", "label_claims", "label_triage", "label_warnings", "warnings_note",
    "claims_headers", "triage_headers", "verified_badge", "verified_title", "unverified_badge",
    "unverified_title", "status_labels", "triage_labels", "msg_refusal", "missing_title",
)


def view_from(cfg: dict) -> dict:
    return {key: cfg["ui"][key] for key in _VIEW_KEYS}


_CFG = load_space_config()
_OPTS = display_opts(_CFG)
_VIEW = view_from(_CFG)
_THEME = build_theme(_CFG)

_SESSION_HITS: dict[str, list[float]] = {}
_SESSION_LOCK = threading.Lock()


def _ui(name: str) -> str:
    return str(_CFG["ui"][name])


def _msg_too_long() -> str:
    return _ui("msg_too_long").format(max_chars=int(_CFG["max_question_chars"]))


def _api_url() -> str:
    override = (os.environ.get("THEANSWER_API_URL") or "").strip()
    if override:
        return override
    return str(_CFG["api_url"]).strip()


# API protocol headers (v2/api/public.py): the response says whether it came from the question cache;
# at the session limit the Space asks for a cache-only answer, so cached repeats never hit the limit.
CACHE_HEADER = "X-TheAnswer-Cache"
CACHE_ONLY_HEADER = "X-TheAnswer-Cache-Only"


def _session_room(session_hash: str, timestamps: list[float]) -> tuple[bool, list[float]]:
    """Whether this session may trigger a new (non-cached) answer now. Records nothing."""
    now = time.monotonic()
    cap = int(_CFG["asks_per_min"])
    window_s = float(_CFG["rate_window_s"])
    window = [stamp for stamp in (timestamps or []) if now - stamp < window_s]
    key = session_hash or ""
    with _SESSION_LOCK:
        prune_at = int(_CFG["session_prune_threshold"])
        if len(_SESSION_HITS) > prune_at:
            stale = [
                bucket
                for bucket, hits in _SESSION_HITS.items()
                if not hits or now - hits[-1] >= window_s
            ]
            for bucket in stale:
                _SESSION_HITS.pop(bucket, None)
        backup = [stamp for stamp in _SESSION_HITS.get(key, []) if now - stamp < window_s]
        _SESSION_HITS[key] = backup
        room = cap > 0 and len(window) < cap and len(backup) < cap
    return room, window


def _session_record(session_hash: str, window: list[float]) -> list[float]:
    """Count one non-cached answer against this session."""
    now = time.monotonic()
    key = session_hash or ""
    with _SESSION_LOCK:
        _SESSION_HITS.setdefault(key, []).append(now)
    return list(window) + [now]


def _friendly_http_error(status: int) -> str:
    if status == 401:
        return _ui("msg_auth")
    if status == 413:
        return _msg_too_long()
    if status == 429:
        return _ui("msg_rate")
    if status == 503:
        return _ui("msg_503")
    if status >= 500:
        return _ui("msg_down")
    return _ui("msg_down")


def _render_payload(question: str, data: dict) -> str:
    view = map_api_response(data)
    if view.is_error and not view.answer_markdown:
        return render_notice(_ui("msg_down"))
    return render_result(view, question, _VIEW, _OPTS)


def ask(
    question: str,
    timestamps: list[float],
    request: gr.Request,
) -> tuple[str, list[float]]:
    question = (question or "").strip()
    if not question:
        return render_notice(_ui("msg_empty")), timestamps or []
    if len(question) > int(_CFG["max_question_chars"]):
        return render_notice(_msg_too_long()), timestamps or []

    session_hash = getattr(request, "session_hash", None) or ""
    room, timestamps = _session_room(session_hash, timestamps or [])

    token = os.environ.get("THEANSWER_API_TOKEN") or ""
    if not token.strip():
        return render_notice(_ui("msg_unconfigured")), timestamps

    headers = {"Authorization": f"Bearer {token}"}
    if not room:  # at the session limit: only a cached answer may come back (no new Gemini call)
        headers[CACHE_ONLY_HEADER] = "1"
    try:
        resp = httpx.post(
            _api_url(),
            json={"question": question},
            headers=headers,
            timeout=float(_CFG["http_timeout_s"]),
        )
    except (httpx.TimeoutException, httpx.NetworkError, httpx.HTTPError):
        if room:
            timestamps = _session_record(session_hash, timestamps)
        return render_notice(_ui("msg_network")), timestamps
    if room and resp.headers.get(CACHE_HEADER) != "hit":  # only non-cached answers count
        timestamps = _session_record(session_hash, timestamps)

    if resp.status_code != 200:
        return render_notice(_friendly_http_error(resp.status_code)), timestamps

    try:
        data = resp.json()
    except ValueError:
        return render_notice(_ui("msg_down")), timestamps
    if not isinstance(data, dict):
        return render_notice(_ui("msg_down")), timestamps
    return _render_payload(question, data), timestamps


def _install_launch(demo: gr.Blocks) -> None:
    original = demo.launch
    host = str(_CFG["local_server_name"])
    port = int(_CFG["local_server_port"])
    css = str(_CFG["ui"]["css"])
    head = str(_CFG["ui"]["head_html"])

    def launch(*args, **kwargs):
        kwargs.setdefault("theme", _THEME)
        kwargs.setdefault("css", css)
        kwargs.setdefault("head", head)
        kwargs.setdefault("share", False)
        if "server_name" not in kwargs and not (os.environ.get("GRADIO_SERVER_NAME") or "").strip():
            kwargs["server_name"] = host
        if "server_port" not in kwargs and not (os.environ.get("GRADIO_SERVER_PORT") or "").strip():
            kwargs["server_port"] = port
        return original(*args, **kwargs)

    demo.launch = launch  # type: ignore[method-assign]


def build_ui() -> gr.Blocks:
    ui = _CFG["ui"]
    rtl = bool(ui["rtl"])
    with gr.Blocks(title=_ui("title"), analytics_enabled=bool(_CFG["analytics_enabled"])) as demo:
        gr.Markdown(_ui("title_markdown"), rtl=rtl, elem_classes=["hero"])
        with gr.Row(elem_classes=["examples"]):
            example_buttons = [
                gr.Button(
                    example,
                    elem_classes=["example-btn"],
                    size="sm",
                    variant="secondary",
                )
                for example in ui["examples"]
            ]
        question = gr.Textbox(
            label=_ui("label_question"),
            lines=int(ui["question_lines"]),
            max_lines=int(ui["question_max_lines"]),
            rtl=rtl,
            elem_id="question",
        )
        button = gr.Button(_ui("button"), variant="primary", elem_id="ask")
        result = gr.HTML(value=render_notice(_ui("msg_hint"), "hint"), elem_id="result")
        stamps = gr.State([])
        button.click(
            ask,
            inputs=[question, stamps],
            outputs=[result, stamps],
            show_progress="full",
        )
        question.submit(
            ask,
            inputs=[question, stamps],
            outputs=[result, stamps],
            show_progress="full",
        )
        auto_submit = bool(ui["examples_auto_submit"])
        for example, example_button in zip(ui["examples"], example_buttons):
            if auto_submit:
                example_button.click(
                    _example_ask(example),
                    inputs=[stamps],
                    outputs=[question, result, stamps],
                    show_progress="full",
                )
            else:
                example_button.click(_example_fill(example), outputs=[question])
        gr.HTML(_ui("footer_html"), elem_classes=["site-footer"])  # last component = bottom of the page
    _install_launch(demo)
    return demo


def _example_fill(example: str):
    def _fill():
        return example

    return _fill


def _example_ask(example: str):
    def _run(timestamps: list[float], request: gr.Request):
        html_out, updated = ask(example, timestamps, request)
        return example, html_out, updated

    return _run


demo = build_ui()
demo.queue(default_concurrency_limit=int(_CFG["concurrency_limit"]))

if __name__ == "__main__":
    demo.launch()