NotSoToothless
91 lines
3.1 KiB
Python
91 lines
3.1 KiB
Python
"""TSSBOT autolog matcher.
|
|
|
|
For each game received from the Spectra TSS feed, match it against every
|
|
subscribing guild's ``tss-team`` preference entries and work out which channels
|
|
should be notified, deduping per session.
|
|
|
|
NOTE: building and sending the scoreboard embed is intentionally a TODO for now
|
|
(it needs significant rework). The matching + dedup pipeline below is complete
|
|
and logs the targets it *would* post to. Wire the send in at the marked spot.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
from typing import Any, Optional
|
|
|
|
import discord
|
|
|
|
from . import preferences
|
|
|
|
log = logging.getLogger("tssbot.autolog")
|
|
|
|
# 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,
|
|
# mirrors SREBOT's _sent_channels_by_session).
|
|
_sent_channels_by_session: dict[str, set[int]] = {}
|
|
|
|
|
|
def set_bot(bot: discord.Client) -> None:
|
|
"""Register the Discord client so the matcher can post (and run at all)."""
|
|
global _bot
|
|
_bot = bot
|
|
|
|
|
|
def _present_teams(game: dict[str, Any]) -> tuple[set[str], set[str]]:
|
|
"""Return stable TSS team IDs and 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
|
|
|
|
|
|
async def process_game(game: dict[str, Any]) -> None:
|
|
"""Match one received game against guild preferences and notify channels.
|
|
|
|
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_ids, present_team_names = _present_teams(game)
|
|
|
|
sent = _sent_channels_by_session.setdefault(session_id, set())
|
|
|
|
for guild_id, prefs in preferences.iter_guild_preferences():
|
|
for entity_id, entry in prefs.items():
|
|
if not isinstance(entry, dict) or entry.get("Type") != "tss-team":
|
|
continue
|
|
channel_id, enabled = preferences.parse_channel(entry.get("Logs"))
|
|
if not channel_id or not enabled:
|
|
continue
|
|
|
|
name = (entry.get("Name") or "").lower()
|
|
matched = str(entity_id) in present_team_ids or (name and name in present_team_names)
|
|
|
|
if not matched or channel_id in sent:
|
|
continue
|
|
sent.add(channel_id)
|
|
|
|
# TODO: build & send the scoreboard embed to `channel_id`.
|
|
# The matched target is fully resolved here; the only thing missing
|
|
# is the rendered scoreboard (out of scope for now).
|
|
log.info(
|
|
"autolog match (send TODO): session=%s guild=%s channel=%s team=%s",
|
|
session_id, guild_id, channel_id, entity_id,
|
|
)
|