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)


def default_scenario_path(persona: dict[str, Any]) -> str | None:
    """Scenario path from the persona, or `data/synthetic/<name>.jsonl` guessed from a `tok-<name>` token."""
    if persona.get("scenario"):
        return str(persona["scenario"])
    token = str(persona.get("token", ""))
    if token.startswith("tok-"):
        guess = f"data/synthetic/{token.removeprefix('tok-')}.jsonl"
        if resolve_path(guess).exists():
            return guess
    return None