| |
| """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() |
|
|
|
|
| |
| |
| 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: |
| 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": |
| 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 |
|
|
|
|
| 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"]) |
| _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() |
|
|
