Echo-Tavern/game/schemas.py

from __future__ import annotations

import json
from typing import Any

ALLOWED_TOP_LEVEL_KEYS = {
    "narration",
    "npc_dialogues",
    "state_updates",
    "private_player_updates",
    "choices",
    "should_end_scene",
    "should_end_game",
    "ending_summary",
}


def fallback_ai_payload(reason: str = "AI 响应异常，主持规则兜底推进。") -> dict[str, Any]:
    return {
        "narration": reason,
        "npc_dialogues": [],
        "state_updates": {
            "danger_delta": 0,
            "add_clues": [],
            "set_flags": {},
            "npc_changes": {},
            "inventory_add": {},
            "inventory_remove": {},
        },
        "private_player_updates": [],
        "choices": ["继续探索大厅", "分头调查可疑 NPC", "检查已获得线索"],
        "should_end_scene": False,
        "should_end_game": False,
        "ending_summary": None,
    }


def parse_json_payload(raw_text: str) -> tuple[dict[str, Any] | None, str | None]:
    if not raw_text:
        return None, "empty_response"
    text = raw_text.strip()

    candidates = [text]
    start = text.find("{")
    end = text.rfind("}")
    if start != -1 and end != -1 and end > start:
        candidates.append(text[start : end + 1])

    for candidate in candidates:
        try:
            parsed = json.loads(candidate)
            if isinstance(parsed, dict):
                return parsed, None
        except json.JSONDecodeError:
            continue
    return None, "invalid_json"


def _ensure_list_of_str(value: Any) -> list[str]:
    if not isinstance(value, list):
        return []
    output: list[str] = []
    for item in value:
        if isinstance(item, str):
            output.append(item.strip())
    return output


def validate_ai_payload(payload: dict[str, Any]) -> tuple[bool, list[str], dict[str, Any]]:
    errors: list[str] = []
    normalized = fallback_ai_payload(reason="")

    unknown_keys = set(payload.keys()) - ALLOWED_TOP_LEVEL_KEYS
    if unknown_keys:
        errors.append(f"unknown_top_keys:{sorted(list(unknown_keys))}")

    narration = payload.get("narration")
    if isinstance(narration, str) and narration.strip():
        normalized["narration"] = narration.strip()
    else:
        errors.append("invalid_narration")

    npc_dialogues = payload.get("npc_dialogues", [])
    normalized_dialogues = []
    if isinstance(npc_dialogues, list):
        for row in npc_dialogues:
            if not isinstance(row, dict):
                continue
            npc = row.get("npc")
            line = row.get("line")
            tone = row.get("tone", "neutral")
            if isinstance(npc, str) and isinstance(line, str):
                normalized_dialogues.append(
                    {"npc": npc.strip(), "line": line.strip(), "tone": str(tone).strip()}
                )
    else:
        errors.append("invalid_npc_dialogues")
    normalized["npc_dialogues"] = normalized_dialogues

    state_updates = payload.get("state_updates", {})
    if not isinstance(state_updates, dict):
        errors.append("invalid_state_updates")
        state_updates = {}

    danger_delta = state_updates.get("danger_delta", 0)
    if not isinstance(danger_delta, int):
        errors.append("invalid_danger_delta")
        danger_delta = 0
    danger_delta = max(-2, min(3, danger_delta))

    add_clues = _ensure_list_of_str(state_updates.get("add_clues", []))
    set_flags_raw = state_updates.get("set_flags", {})
    set_flags = set_flags_raw if isinstance(set_flags_raw, dict) else {}
    if not isinstance(set_flags_raw, dict):
        errors.append("invalid_set_flags")

    npc_changes_raw = state_updates.get("npc_changes", {})
    npc_changes = npc_changes_raw if isinstance(npc_changes_raw, dict) else {}
    if not isinstance(npc_changes_raw, dict):
        errors.append("invalid_npc_changes")

    inventory_add_raw = state_updates.get("inventory_add", {})
    inventory_add: dict[str, list[str]] = {}
    if isinstance(inventory_add_raw, dict):
        for player_id, items in inventory_add_raw.items():
            if isinstance(player_id, str):
                inventory_add[player_id] = _ensure_list_of_str(items)
    else:
        errors.append("invalid_inventory_add")

    inventory_remove_raw = state_updates.get("inventory_remove", {})
    inventory_remove: dict[str, list[str]] = {}
    if isinstance(inventory_remove_raw, dict):
        for player_id, items in inventory_remove_raw.items():
            if isinstance(player_id, str):
                inventory_remove[player_id] = _ensure_list_of_str(items)
    else:
        errors.append("invalid_inventory_remove")

    normalized["state_updates"] = {
        "danger_delta": danger_delta,
        "add_clues": add_clues,
        "set_flags": set_flags,
        "npc_changes": npc_changes,
        "inventory_add": inventory_add,
        "inventory_remove": inventory_remove,
    }

    private_updates = payload.get("private_player_updates", [])
    normalized_private = []
    if isinstance(private_updates, list):
        for row in private_updates:
            if isinstance(row, dict) and isinstance(row.get("player_id"), str) and isinstance(
                row.get("message"), str
            ):
                normalized_private.append(
                    {
                        "player_id": row["player_id"].strip(),
                        "message": row["message"].strip(),
                    }
                )
    else:
        errors.append("invalid_private_updates")
    normalized["private_player_updates"] = normalized_private

    choices = _ensure_list_of_str(payload.get("choices", []))[:4]
    if not choices:
        choices = ["继续探索", "与 NPC 交谈", "整理线索"]
    normalized["choices"] = choices

    should_end_scene = payload.get("should_end_scene", False)
    if not isinstance(should_end_scene, bool):
        errors.append("invalid_should_end_scene")
        should_end_scene = False
    normalized["should_end_scene"] = should_end_scene

    should_end_game = payload.get("should_end_game", False)
    if not isinstance(should_end_game, bool):
        errors.append("invalid_should_end_game")
        should_end_game = False
    normalized["should_end_game"] = should_end_game

    ending_summary = payload.get("ending_summary")
    if ending_summary is not None and not isinstance(ending_summary, str):
        errors.append("invalid_ending_summary")
        ending_summary = None
    normalized["ending_summary"] = ending_summary

    if not normalized["narration"]:
        normalized["narration"] = "主持规则接管：本轮叙述采用系统兜底。"

    return len(errors) == 0, errors, normalized