"""TSSBOT autolog matcher + scoreboard dispatch. For each game received from the Spectra TSS feed, match it against every subscribing guild's ``tss-team`` / ``tss-player`` preference entries, render the scoreboard once per session/color/language, and post it to each subscribed channel. Matching is per-entity (a team by id/name, a player by uid); sending is deduped per channel per session so a channel that subscribes both a team and one of its players only receives one scoreboard for a given game. """ from __future__ import annotations import asyncio import logging from pathlib import Path from typing import Any, Optional import discord from . import preferences, scoreboard, transform from .storage import STORAGE_DIR log = logging.getLogger("tssbot.autolog") REPLAYS_TSS_DIR: Path = STORAGE_DIR / "REPLAYS" / "TSS" # Registered by start_bot once the client exists; standalone tss_ws leaves it None. _bot: Optional[discord.Client] = None # session_id -> set of channel_ids already handled (in-memory idempotency). _sent_channels_by_session: dict[str, set[int]] = {} # session_id -> lock guarding the one-time PNG render. _render_locks: dict[str, asyncio.Lock] = {} def set_bot(bot: discord.Client) -> None: """Register the Discord client so the matcher can post (and run at all).""" global _bot _bot = bot # --------------------------------------------------------------------------- # Present-entity extraction # --------------------------------------------------------------------------- def _present_teams(game: dict[str, Any]) -> tuple[set[str], set[str]]: """Return stable TSS team IDs and lowercased names embedded in a replay.""" tss = game.get("tss") or {} team_ids: set[str] = set() team_names: set[str] = set() for slot in ("1", "2"): team = tss.get(slot) if not isinstance(team, dict): continue if team.get("team_id") is not None: team_ids.add(str(team["team_id"])) name = str(team.get("team_name") or team.get("name") or "").strip() if name: team_names.add(name.lower()) return team_ids, team_names def _present_players(game: dict[str, Any]) -> set[str]: """Return the uids of every player in a replay (players dict + tss rosters).""" uids: set[str] = {str(u) for u in (game.get("players") or {}).keys()} tss = game.get("tss") or {} for slot in ("1", "2"): team = tss.get(slot) if isinstance(team, dict): for entry in team.get("players") or []: if entry.get("uid") is not None: uids.add(str(entry["uid"])) return uids def _bar_color(game: dict[str, Any], guild_id: int) -> str: """Header/separator tint from the guild's own team vs the result. Compared by team *name* (case-insensitive) — TSS team ids are per-tournament. """ if game.get("draw"): return "draw" guild_team = preferences.get_guild_team(guild_id) my_name = str((guild_team or {}).get("TM_Name") or "").casefold() if not my_name: return "not_set" tss = game.get("tss") or {} slot_names = {s: str((tss.get(s) or {}).get("team_name") or "").casefold() for s in ("1", "2")} winner = str(game.get("winner") or "") if slot_names.get(winner) == my_name: return "win" if my_name in slot_names.values(): return "loss" return "not_involved" # --------------------------------------------------------------------------- # Scoreboard view + render/send # --------------------------------------------------------------------------- def build_tss_scoreboard_view(session_id: str) -> discord.ui.View: """Link buttons under a scoreboard: in-game replay + the TSS website.""" view = discord.ui.View(timeout=None) try: replay_url = f"https://warthunder.com/en/tournament/replay/{int(session_id, 16)}" view.add_item(discord.ui.Button(label="View Replay", style=discord.ButtonStyle.link, url=replay_url)) except (ValueError, TypeError): pass view.add_item(discord.ui.Button( label="View on Website", style=discord.ButtonStyle.link, url=f"https://tss.pawjob.us/games/{session_id}", emoji="🌐", )) return view async def _render_scoreboard(game: dict[str, Any], session_id: str, bar_color: str, lang_column: str) -> Optional[Path]: """Render (once, cached on disk) the scoreboard for a session/color/language.""" model = transform.build_scoreboard_model(game, lang_column) if not model: log.warning("[TSS-AUTOLOG] could not build model for %s", session_id) return None lang_tag = lang_column.strip("<>").lower() or "english" out_dir = REPLAYS_TSS_DIR / session_id out_path = out_dir / f"scoreboard-{bar_color}-{lang_tag}.png" lock = _render_locks.setdefault(session_id, asyncio.Lock()) async with lock: if not out_path.exists(): out_dir.mkdir(parents=True, exist_ok=True) await scoreboard.create_scoreboard(model, str(out_path), bar_color=bar_color) return out_path if out_path.exists() else None async def _send_scoreboard(game: dict[str, Any], guild_id: int, channel_id: int, session_id: str, sent: set[int]) -> None: """Render and post the scoreboard for one subscribed channel (deduped).""" bot = _bot if bot is None or channel_id in sent: return sent.add(channel_id) # claim the channel before any await so concurrent matches can't double-send bar_color = _bar_color(game, guild_id) lang_column = preferences.guild_language_column(guild_id) out_path = await _render_scoreboard(game, session_id, bar_color, lang_column) if out_path is None: return channel = bot.get_channel(channel_id) if channel is None: try: channel = await bot.fetch_channel(channel_id) except (discord.NotFound, discord.Forbidden) as e: log.warning("[TSS-AUTOLOG] channel %s unavailable (%s)", channel_id, e) return if not isinstance(channel, discord.abc.Messageable): return try: with open(out_path, "rb") as f: await channel.send( file=discord.File(f, filename="scoreboard.png"), view=build_tss_scoreboard_view(session_id), ) log.info("[TSS-AUTOLOG] sent %s -> guild=%s channel=%s", session_id, guild_id, channel_id) except discord.HTTPException as e: log.error("[TSS-AUTOLOG] send failed %s -> %s: %s", session_id, channel_id, e) # --------------------------------------------------------------------------- # Entry point # --------------------------------------------------------------------------- async def process_game(game: dict[str, Any]) -> None: """Match one received game against guild prefs and post scoreboards. Safe to call from the standalone WS listener (no-ops if no bot registered). """ if _bot is None: return session_id = str(game.get("_id") or "") if not session_id: return _, present_team_names = _present_teams(game) present_uids = _present_players(game) sent = _sent_channels_by_session.setdefault(session_id, set()) for guild_id, prefs in preferences.iter_guild_preferences(): for _pref_key, entry in prefs.items(): if not isinstance(entry, dict): continue type_ = entry.get("Type") if type_ not in ("tss-team", "tss-player", "tss-wildcard"): continue channel_id, enabled = preferences.parse_channel(entry.get("Logs")) if not channel_id or not enabled or channel_id in sent: continue if type_ == "tss-wildcard": matched = True elif type_ == "tss-team": name = (entry.get("Name") or "").lower() matched = bool(name) and name in present_team_names else: # tss-player matched = str(entry.get("UID") or "") in present_uids if not matched: continue try: await _send_scoreboard(game, guild_id, channel_id, session_id, sent) except Exception as e: # never let one channel break the rest log.error("[TSS-AUTOLOG] dispatch error guild=%s channel=%s: %s", guild_id, channel_id, e)