"""In-memory transcript of the patient chat shown by the console (kept in Streamlit session state). Entries are plain dicts: `{"dir": "out", "text", "kind", "buttons"}` for bot replies (the `Reply` contract) and `{"dir": "in", "text", "pressed": bool}` for what the patient sent. """ from __future__ import annotations from collections.abc import Iterable from typing import Any MAX_ENTRIES = 400 def normalize_reply(raw: dict[str, Any]) -> dict[str, Any]: """Coerce an API reply into a safe shape: missing fields get defaults, malformed buttons are dropped.""" rows: list[list[dict[str, str]]] = [] for row in raw.get("buttons") or []: if not isinstance(row, list): continue clean = [ {"label": str(b.get("label", "")), "data": str(b.get("data", ""))} for b in row if isinstance(b, dict) and b.get("data") is not None ] if clean: rows.append(clean) return { "dir": "out", "text": str(raw.get("text") or ""), "kind": str(raw.get("kind") or "normal"), "buttons": rows, } class ChatLog: """Wraps a list owned by the caller (session state), so every page sees the same transcript.""" def __init__(self, entries: list[dict[str, Any]] | None = None): self.entries: list[dict[str, Any]] = entries if entries is not None else [] def _trim(self) -> None: if len(self.entries) > MAX_ENTRIES: del self.entries[: len(self.entries) - MAX_ENTRIES] def add_in(self, text: str, *, pressed: bool = False) -> None: self.entries.append({"dir": "in", "text": text, "pressed": pressed}) self._trim() def add_replies(self, replies: Iterable[dict[str, Any]]) -> list[dict[str, Any]]: added = [normalize_reply(r) for r in replies if isinstance(r, dict)] self.entries.extend(added) self._trim() return added def add_history(self, messages: Iterable[dict[str, Any]]) -> None: """Prefill from `GET /dialog/{token}/history` (MessageOut: no buttons).""" for m in messages: if not isinstance(m, dict): continue if m.get("direction") == "in": self.entries.append({"dir": "in", "text": str(m.get("text", "")), "pressed": False}) else: self.entries.append(normalize_reply({"text": m.get("text"), "kind": m.get("kind")})) self._trim() def recent_replies(self, n: int = 10) -> list[dict[str, Any]]: """Last `n` bot replies, oldest first.""" outs = [e for e in self.entries if e.get("dir") == "out"] return outs[-n:] if n > 0 else [] def last_buttons_index(self) -> int | None: """Index of the newest bot reply that carries buttons — only its keyboard is clickable in the web chat.""" for i in range(len(self.entries) - 1, -1, -1): e = self.entries[i] if e.get("dir") == "out" and e.get("buttons"): return i return None def clear(self) -> None: self.entries.clear()