#!/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()