"""Single-call combined response + widget generation — Claude-style architecture.
Instead of two sequential LLM calls (response then widget), this module
lets GPT-4 generate BOTH in one pass:
Output format:
[answer text here — follows primitive format rule]
[either complete self-contained HTML OR JSON UI schema (depending on WIDGET_MODE)]
The parser splits these apart. Text goes to chat; widget payload goes to renderer.
This matches Claude's architecture: one model, one generation, no sequential delay.
"""
from __future__ import annotations
import json
import re
from typing import Any, Tuple
from . import config
from .widget_prompt import inject_design_system
def strip_widget_markdown_fences(raw: str) -> str:
"""Remove ```json ... ``` wrappers often emitted inside despite instructions."""
s = (raw or "").strip()
if not s or "```" not in s:
return s
fence = re.search(r"```(?:json|html|javascript|js)?\s*(.*?)```", s, re.DOTALL | re.IGNORECASE)
if fence:
return fence.group(1).strip()
return re.sub(r"```\w*", "", s).strip()
def parse_widget_schema_object(s: str) -> Any | None:
"""Parse first JSON value; tolerate leading/trailing prose via JSONDecoder.raw_decode."""
s = strip_widget_markdown_fences(s).strip()
if not s:
return None
dec = json.JSONDecoder()
for start_ch in ("{", "["):
idx = s.find(start_ch)
if idx == -1:
continue
try:
obj, _ = dec.raw_decode(s, idx)
return obj
except json.JSONDecodeError:
continue
try:
return json.loads(s)
except json.JSONDecodeError:
return None
_BLOCK_TYPES = frozenset({"text", "kpi_row", "chart", "table", "action_row", "image"})
_TYPE_ALIASES: dict[str, str] = {
"markdown": "text",
"md": "text",
"paragraph": "text",
"rich_text": "text",
"kpi": "kpi_row",
"kpis": "kpi_row",
"kpirow": "kpi_row",
"metrics": "kpi_row",
"metric_row": "kpi_row",
"data_table": "table",
"grid": "table",
"actions": "action_row",
"action": "action_row",
"plot": "chart",
"graph": "chart",
"line_chart": "chart",
"bar_chart": "chart",
"photo": "image",
"picture": "image",
"img": "image",
"illustration": "image",
}
def _normalize_layout_block(entry: Any) -> dict[str, Any]:
"""Turn primitives, malformed entries, and common aliases into valid block dicts."""
if entry is None:
return {"type": "text", "content": ""}
if isinstance(entry, (str, int, float, bool)):
return {"type": "text", "content": str(entry)}
if isinstance(entry, list):
return {"type": "text", "content": json.dumps(entry, ensure_ascii=False)}
if not isinstance(entry, dict):
return {"type": "text", "content": str(entry)}
b: dict[str, Any] = dict(entry)
raw_t = str(b.get("type") or "").strip().lower().replace("-", "_")
t = _TYPE_ALIASES.get(raw_t, raw_t)
if t in _BLOCK_TYPES:
b["type"] = t
return b
items = b.get("items")
if isinstance(items, list) and items:
first = items[0]
if isinstance(first, dict) and isinstance(first.get("label"), str):
v = first.get("value")
if isinstance(v, (str, int, float, bool)):
b["type"] = "kpi_row"
return b
if isinstance(b.get("chart"), dict):
b["type"] = "chart"
return b
src_val = b.get("src")
if isinstance(src_val, str) and src_val.strip():
b["type"] = "image"
return b
if isinstance(b.get("rows"), list):
b["type"] = "table"
return b
if isinstance(b.get("buttons"), list):
b["type"] = "action_row"
return b
for key in ("content", "body", "text", "markdown", "md"):
v = b.get(key)
if isinstance(v, str):
return {"type": "text", "content": v}
return {"type": "text", "content": json.dumps(b, ensure_ascii=False)}
def _sanitize_layout(layout: list[Any]) -> list[dict[str, Any]]:
return [_normalize_layout_block(e) for e in layout]
def coerce_widget_schema_root(obj: Any) -> dict[str, Any] | None:
"""Ensure root has layout: [] (alias blocks/components; wrap bare arrays)."""
if obj is None:
return None
if isinstance(obj, list):
return {"version": "1.0", "layout": obj}
if not isinstance(obj, dict):
return None
out = dict(obj)
# Unwrap { "widget": { "layout": [...] } } or { "schema": {...} }
if isinstance(out.get("widget"), dict):
inner = dict(out.pop("widget"))
out = {**out, **inner}
if isinstance(out.get("schema"), dict):
inner = dict(out.pop("schema"))
out = {**out, **inner}
if "layout" not in out or not isinstance(out.get("layout"), list):
if isinstance(out.get("blocks"), list):
out["layout"] = out["blocks"]
elif isinstance(out.get("components"), list):
out["layout"] = out["components"]
elif isinstance(out.get("Layout"), list):
out["layout"] = out.pop("Layout")
# layout is a single block object (common model mistake)
lay = out.get("layout")
if isinstance(lay, dict) and lay.get("type"):
out["layout"] = [lay]
# Root is one block with no layout key
if not isinstance(out.get("layout"), list) and out.get("type") in _BLOCK_TYPES:
ver = out.pop("version", None)
block = dict(out)
out = {"version": str(ver or "1.0"), "layout": [block]}
if isinstance(out.get("layout"), list):
out["layout"] = _sanitize_layout(out["layout"])
return out
def widget_schema_json_is_valid(schema_str: str) -> bool:
if not (schema_str or "").strip():
return False
try:
o = json.loads(schema_str)
except json.JSONDecodeError:
return False
return isinstance(o, dict) and isinstance(o.get("layout"), list)
def extract_embeddable_html_document(raw: str) -> str | None:
"""
If the model put HTML/JS widgets inside while WIDGET_MODE=json, return a full HTML
document suitable for inject_design_system + iframe. Otherwise None.
"""
s = strip_widget_markdown_fences((raw or "").strip())
if not s or "<" not in s or ">" not in s:
return None
st = s.lstrip()
if st.startswith("{") and "]*>", "", s, flags=re.IGNORECASE).strip()
if any(
tag in low
for tag in (
"]*>", "", s, flags=re.IGNORECASE).strip()
if "{inner}