NotSoToothless
100 lines
3.4 KiB
Python
100 lines
3.4 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, storage
|
|
|
|
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_team_tags(game: dict[str, Any]) -> set[str]:
|
|
"""Return the set of team tags present in a game dict."""
|
|
players = game.get("players") or {}
|
|
tags: set[str] = set()
|
|
for p in players.values():
|
|
if not isinstance(p, dict):
|
|
continue
|
|
tag_raw = p.get("tag") or ""
|
|
tag = tag_raw[1:-1] if len(tag_raw) > 2 else tag_raw
|
|
if tag:
|
|
tags.add(tag)
|
|
return tags
|
|
|
|
|
|
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_tags = _present_team_tags(game)
|
|
tags_lower = {t.lower() for t in present_tags}
|
|
|
|
# Resolve present tags → team_ids so team subscriptions (keyed by team_id) match.
|
|
present_team_ids: set[str] = set()
|
|
for tag in present_tags:
|
|
try:
|
|
tid = await storage.resolve_team_id_for_tag(tag)
|
|
except Exception as exc:
|
|
log.error("tag resolve failed for %s: %s", tag, exc)
|
|
tid = None
|
|
if tid is not None:
|
|
present_team_ids.add(str(tid))
|
|
|
|
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 tags_lower)
|
|
|
|
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,
|
|
)
|