PTSRaiAgent/dashboard/console/scenario.py

"""Demo scenario files `data/synthetic/<persona>.jsonl` (format agreed with #18).

One JSON object per line: `{"day": <therapy day>, "note": "<presenter hint>", "actions": [...]}` where an action
is exactly one of `{"advance_to": "HH:MM"}`, `{"press": "<label substring>"}`, `{"text": "<message>"}`,
`{"expect_color": "green|yellow|red|grey"}`. The console also accepts `{"start": "<connect code or empty>"}`
(sends a `start` event) so a scenario can open the dialog itself.

The parser is lenient on purpose: a broken line or action is reported and skipped, the rest still plays —
a typo in the scenario must not take the live demo down.
"""

from __future__ import annotations

import json
import re
from dataclasses import dataclass, field
from pathlib import Path
from typing import Any, Literal

ActionKind = Literal["advance_to", "press", "text", "expect_color", "start"]
ACTION_KINDS: tuple[ActionKind, ...] = ("advance_to", "press", "text", "expect_color", "start")
COLORS = ("green", "yellow", "red", "grey")
_HHMM = re.compile(r"^([01]?\d|2[0-3]):([0-5]\d)$")

REPO_ROOT = Path(__file__).resolve().parents[2]


@dataclass(frozen=True)
class Action:
    kind: ActionKind
    value: str

    def describe(self) -> str:
        """Short Russian description for the step log."""
        return {
            "advance_to": f"⏩ часы → {self.value}",
            "press": f"🔘 нажать «{self.value}»",
            "text": f"💬 написать «{self.value}»",
            "expect_color": f"🎯 ожидаем цвет {self.value}",
            "start": "🚀 начать диалог" + (f" (код {self.value})" if self.value else ""),
        }[self.kind]


@dataclass(frozen=True)
class DayScript:
    day: int
    note: str
    actions: tuple[Action, ...]


@dataclass
class Scenario:
    days: list[DayScript] = field(default_factory=list)
    errors: list[str] = field(default_factory=list)

    def day(self, number: int) -> DayScript | None:
        return next((d for d in self.days if d.day == number), None)

    @property
    def day_numbers(self) -> list[int]:
        return [d.day for d in self.days]


def parse_hhmm(value: str) -> tuple[int, int]:
    """'8:05' / '08:05' → (8, 5); raises ValueError on anything else."""
    m = _HHMM.match(value.strip())
    if m is None:
        raise ValueError(f"время должно быть в формате ЧЧ:ММ, получено «{value}»")
    return int(m.group(1)), int(m.group(2))


def parse_action(raw: Any) -> Action:
    """Validate one action object; raises ValueError with a Russian message."""
    if not isinstance(raw, dict) or len(raw) != 1:
        raise ValueError(f"действие должно быть объектом с одним ключом, получено {raw!r}")
    ((kind, value),) = raw.items()
    if kind not in ACTION_KINDS:
        raise ValueError(f"неизвестное действие «{kind}»")
    if value is None and kind == "start":
        value = ""
    if not isinstance(value, str):
        raise ValueError(f"значение «{kind}» должно быть строкой")
    value = value.strip()
    if kind == "advance_to":
        h, m = parse_hhmm(value)
        value = f"{h:02d}:{m:02d}"
    elif kind == "expect_color":
        value = value.lower()
        if value not in COLORS:
            raise ValueError(f"цвет должен быть одним из {', '.join(COLORS)}, получено «{value}»")
    elif kind in ("press", "text") and not value:
        raise ValueError(f"пустое значение «{kind}»")
    return Action(kind, value)  # type: ignore[arg-type]


def parse_scenario(text: str) -> Scenario:
    """Parse JSONL text. Days are returned sorted; duplicates keep the first occurrence."""
    scenario = Scenario()
    seen: set[int] = set()
    for lineno, line in enumerate(text.splitlines(), start=1):
        line = line.strip()
        if not line or line.startswith("//") or line.startswith("#"):
            continue
        where = f"строка {lineno}"
        try:
            obj = json.loads(line)
        except json.JSONDecodeError as exc:
            scenario.errors.append(f"{where}: не JSON ({exc.msg})")
            continue
        if not isinstance(obj, dict):
            scenario.errors.append(f"{where}: ожидается объект")
            continue
        day = obj.get("day")
        if isinstance(day, bool) or not isinstance(day, int) or day < 0:
            scenario.errors.append(f"{where}: поле day должно быть целым ≥ 0")
            continue
        if day in seen:
            scenario.errors.append(f"{where}: день {day} повторяется, строка пропущена")
            continue
        raw_actions = obj.get("actions", [])
        if not isinstance(raw_actions, list):
            scenario.errors.append(f"{where}: actions должно быть списком")
            continue
        actions: list[Action] = []
        for i, raw in enumerate(raw_actions, start=1):
            try:
                actions.append(parse_action(raw))
            except ValueError as exc:
                scenario.errors.append(f"{where}, день {day}, действие {i}: {exc} — пропущено")
        note = obj.get("note", "")
        seen.add(day)
        scenario.days.append(DayScript(day=day, note=str(note) if note is not None else "", actions=tuple(actions)))
    scenario.days.sort(key=lambda d: d.day)
    return scenario


def resolve_path(path: str | Path, root: Path = REPO_ROOT) -> Path:
    p = Path(path)
    return p if p.is_absolute() else root / p


def load_scenario(path: str | Path, root: Path = REPO_ROOT) -> Scenario:
    """Read and parse a scenario file; a missing/unreadable file yields an empty scenario with an error."""
    full = resolve_path(path, root)
    try:
        text = full.read_text(encoding="utf-8")
    except FileNotFoundError:
        return Scenario(errors=[f"файл сценария не найден: {path}"])
    except OSError as exc:
        return Scenario(errors=[f"не удалось прочитать {path}: {type(exc).__name__}"])
    return parse_scenario(text)


_TRANSLIT = {
    "а": "a", "б": "b", "в": "v", "г": "g", "д": "d", "е": "e", "ё": "e", "ж": "zh", "з": "z", "и": "i",
    "й": "y", "к": "k", "л": "l", "м": "m", "н": "n", "о": "o", "п": "p", "р": "r", "с": "s", "т": "t",
    "у": "u", "ф": "f", "х": "kh", "ц": "ts", "ч": "ch", "ш": "sh", "щ": "shch", "ъ": "", "ы": "y", "ь": "",
    "э": "e", "ю": "yu", "я": "ya",
}  # fmt: skip


def persona_slug(display_name: str) -> str:
    """'Андрей С.' → 'andrey': the backend names scenario files by the transliterated first name (#12)."""
    first = (display_name.split() or [""])[0].lower()
    return "".join(_TRANSLIT.get(ch, ch if ch.isascii() and ch.isalnum() else "") for ch in first)


def default_scenario_path(persona: dict[str, Any], root: Path = REPO_ROOT) -> str | None:
    """Scenario path from the persona, else `data/synthetic/<slug>.jsonl` guessed from the name or `tok-<slug>`."""
    if persona.get("scenario"):
        return str(persona["scenario"])
    token = str(persona.get("token", ""))
    slugs = [persona_slug(str(persona.get("display_name", "")))]
    if token.startswith("tok-"):
        slugs.append(token.removeprefix("tok-"))
    for slug in filter(None, slugs):
        guess = f"data/synthetic/{slug}.jsonl"
        if resolve_path(guess, root).exists():
            return guess
    return None