weirsoe-party-protocol/party/spion.py

"""Rum til Spionen.

Deler forbindelser, timer, udsendelse og speakerens kø med de andre spil (`rooms.BaseRoom`). Speakeren siger
aldrig ordet eller hvem der er spion, før dommen og resultatet viser det på storskærmen: replikkerne til dommen
spørger kun ("er det spionen?"), svaret kommer med gættet eller resultatet.
"""

from __future__ import annotations

import asyncio
import json
import logging
import time
from pathlib import Path

from spionen.engine import MAX_PLAYERS, MIN_PLAYERS, ROUND_CHOICES, RULES, Game

from . import rooms, tts
from .narrator import Narrator, generic_texts, join_names
from .themes import spy_themes

log = logging.getLogger(__name__)

NARRATOR_DIR = Path(__file__).resolve().parents[1] / "spionen" / "narrator"
LINES = json.loads((NARRATOR_DIR / "lines.json").read_text(encoding="utf-8"))
GENERIC = json.loads((NARRATOR_DIR / "generic.json").read_text(encoding="utf-8"))  # se party.narrator
HURRY_BEFORE = 10
# Replikker uden navne og uden ordet: dem kan tts_warm lave på forhånd (resten har en generisk afløser).
FIXED_LINES = ("lobby_welcome", "round_first", "round_last", "clue_hurry", "clue_all_done", "clues_intro",
               "clues_missing", "talk_start", "vote_start", "vote_all_done", "verdict_tie", "verdict_none",
               "result_guessed", "result_missed", "result_timeout", "result_free", "play_again")
AWARD_LINES = {"liar": "award_liar", "sharp": "award_sharp", "suspected": "award_suspected"}


class SpionRoom(rooms.BaseRoom):
    kind = "spion"

    def __init__(self, code: str, game: Game, host_username: str):
        super().__init__(code, game, host_username)
        self.narrator = Narrator(lines=LINES, generic=GENERIC)
        self._last_join_line = 0.0
        self._lobby_waiting_said = False

    # ── handlinger ───────────────────────────────────────────────────────────
    async def play_action(self, conn, member_id: str, action: str, content: dict, now: float):
        g = self.game
        if action == "start":
            g.start(member_id, now)
        elif action == "submit_clue":
            g.submit_clue(member_id, str(content.get("text", "")), now)
        elif action == "vote":
            g.vote(member_id, str(content["player_id"]), now)
        elif action == "spy_guess":
            g.spy_guess(member_id, str(content.get("text", "")), now)
        elif action == "react":
            reaction = g.react(member_id, str(content.get("emoji", "")), now)
            if reaction:
                await self.forward_reaction(reaction)
        elif action == "play_again":
            g.play_again(member_id, now)
        else:
            await conn.send_error("bad_request", "Ukendt handling.")

    # ── speakeren ────────────────────────────────────────────────────────────
    def host_opened(self):
        if tts.enabled() and not self._welcomed and self.game.phase == "lobby" and not self.game.players:
            self._welcomed = True
            self.say(self.narrator.line("lobby_welcome"))

    def handle_events(self):
        events = self.game.drain_events()
        if not tts.enabled():
            return
        skip_start = any(e["type"] == "play_again" for e in events)
        for ev in events:
            t = ev["type"]
            if t == "game_start" and skip_start:
                continue
            if t == "rules":
                asyncio.get_running_loop().create_task(self._narrate_rules())
                continue
            if t == "join":
                now = time.time()
                if now - self._last_join_line < 2.5:  # otte på én gang skal ikke alle kommenteres
                    continue
                self._last_join_line = now
                self._maybe_schedule_lobby_waiting()
            for text in self.lines_for(ev):
                self.say(text)
            if t == "round_start":
                self._hurry()

    def lines_for(self, ev: dict) -> list[str]:
        t, line = ev["type"], self.narrator.line
        if t == "join" and ev["role"] == "player" and self.game.phase == "lobby":
            return [line("lobby_join", navn=ev["name"])]
        if t == "game_start":
            return [line("game_start", antal_spillere=ev["players"])]
        if t == "round_start":
            head = (line("round_first") if ev["round"] == 1 else line("round_last") if ev["last"]
                    else line("round_next", runde=ev["round"]))
            return [head, line("theme", tema=ev["theme"])]
        if t == "clues":
            return [line("clues_intro")] + ([line("clues_missing")] if ev["missing"] else [])
        if t in ("clue_all_done", "vote_all_done", "play_again"):
            return [line(t)]
        if t == "talk":
            return [line("talk_start")]
        if t == "vote":
            return [line("vote_start")]
        if t == "verdict":
            if ev["no_votes"]:
                return [line("verdict_none")]
            return [line("verdict_tie")] if ev["tie"] else [line("verdict_accused", navn=ev["accused"])]
        if t == "guess":
            return [line("guess_start", navn=ev["name"])]
        if t == "result":
            out = []
            if ev["accused_innocent"]:
                out.append(line("result_innocent", navn=self._accused_name()))
            elif ev["caught"]:
                g = self.game
                out.append(line("result_guessed" if ev["guessed"] else "result_missed" if g.guess else "result_timeout"))
            else:
                out.append(line("result_free"))
            out.append(line("result_reveal", spioner=join_names(ev["spies"]), ord=ev["word"]))
            return out
        if t == "final":
            out = [line("winner", vinder=ev["winner"], point=ev["points"])]
            return out + [line(AWARD_LINES[a["kind"]], navn=a["name"]) for a in ev["awards"]]
        return []

    def _accused_name(self) -> str:
        p = self.game.players.get(self.game.accused or "")
        return p.name if p else ""

    def _hurry(self):
        g = self.game
        ident = g.phase_key()
        delay = (g.deadline or 0) - HURRY_BEFORE - time.time()
        if delay <= 1:
            return

        def fire():
            if g.phase_key() == ident and not g._all_done():
                self.say(self.narrator.line("clue_hurry"))
        self._side_timers.append(asyncio.get_running_loop().call_later(delay, fire))
        self._side_timers = [h for h in self._side_timers if not h.cancelled()][-20:]

    def _maybe_schedule_lobby_waiting(self):
        g = self.game
        if self._lobby_waiting_said or g.phase != "lobby" or len(g.players) < MIN_PLAYERS:
            return
        self._lobby_waiting_said = True

        def fire():
            vip = g.players.get(g.vip_id() or "")
            if g.phase == "lobby" and vip:
                self.say(self.narrator.line("lobby_waiting", vip=vip.name))
        self._side_timers.append(asyncio.get_running_loop().call_later(rooms.LOBBY_WAITING_AFTER, fire))


def warm_texts(themes: list[dict] | None = None) -> list[str]:
    """Spionens faste replikker præcis som rummet siger dem (standardstemmen): de faste, reglerne, temaerne,
    rundernes nummer og spillets start for 4-10 spillere. Replikker med navne eller ordet er ikke med; de har en
    generisk afløser (warm_generic)."""
    themes = spy_themes() if themes is None else themes
    n = Narrator(lines=LINES, generic=GENERIC)
    out = [t for ev in FIXED_LINES for t in n.variants(ev)]
    out += [r["say"] for r in RULES]
    out += [t for th in themes for t in n.variants("theme", tema=th["name"])]
    out += [t for k in range(2, max(ROUND_CHOICES)) for t in n.variants("round_next", runde=k)]
    out += [t for k in range(MIN_PLAYERS, MAX_PLAYERS + 1) for t in n.variants("game_start", antal_spillere=k)]
    return list(dict.fromkeys(out))


def warm_generic() -> list[str]:
    return list(dict.fromkeys(t for t, _ in generic_texts(GENERIC, None)))


def create_spion_room(host_username: str, public_url: str, code: str | None = None, show_rules: bool = True,
                      rounds: int = 4) -> SpionRoom:
    """code: et kørende rums kode = skift spil i det rum (se rooms.add_room)."""
    rooms.cleanup()
    code = code or rooms.new_code()
    game = Game(code, spy_themes(), join_url=f"{public_url}/?code={code}", rounds=rounds)
    if not show_rules:
        game.show_rules = False
    room = SpionRoom(code, game, host_username)
    rooms.add_room(room)
    log.info("Spion-rum %s oprettet af %s (%d runder)", code, host_username, game.rounds)
    return room