Spaces:
Sleeping
Sleeping
Download main.py from wuhp/devstalker: direct link, hf CLI and curl.
- Browser
- Download file 63.7 kB
-
https://huggingface.co/spaces/wuhp/devstalker/resolve/main/main.py
- Command line
-
hf download hf://spaces/wuhp/devstalker/main.py
-
curl -L -o main.py https://huggingface.co/spaces/wuhp/devstalker/resolve/main/main.py
63.7 kB
| from __future__ import annotations | |
| import asyncio | |
| import json | |
| import logging | |
| import os | |
| import sqlite3 | |
| import socket | |
| import time | |
| import base64 | |
| from collections import Counter | |
| from io import BytesIO | |
| from datetime import datetime, timezone, timedelta | |
| from pathlib import Path | |
| from typing import Any, Optional | |
| import discord | |
| from discord import app_commands | |
| from discord.ext import tasks | |
| from fastapi import FastAPI | |
| import aiohttp | |
| import uvicorn | |
| import requests | |
| import yarl | |
| import matplotlib | |
| matplotlib.use("Agg") | |
| import matplotlib.pyplot as plt | |
| from tracker_client import TrackerClient, TrackerError | |
| APP_DIR = Path(__file__).resolve().parent | |
| DB_PATH = Path(os.getenv("DB_PATH", str(APP_DIR / "rocket_watch.db"))) | |
| POLL_SECONDS = max(60, int(os.getenv("POLL_SECONDS", "120"))) | |
| ALERT_MAX_AGE_MINUTES = max(5, int(os.getenv("ALERT_MAX_AGE_MINUTES", "30"))) | |
| DISCORD_BOT_TOKEN = os.getenv("DISCORD_BOT_TOKEN", "").strip() | |
| GUILD_ID = os.getenv("DISCORD_GUILD_ID", "").strip() | |
| DISCORD_LOGIN_TIMEOUT = max(5, int(os.getenv("DISCORD_LOGIN_TIMEOUT", "20"))) | |
| DISCORD_READY_TIMEOUT = max(10, int(os.getenv("DISCORD_READY_TIMEOUT", "30"))) | |
| DISCORD_RELAY_ORIGIN = (os.getenv("DISCORD_RELAY_ORIGIN") or os.getenv("DISCORD_PROXY_ORIGIN", "")).strip().rstrip("/") | |
| DISCORD_RELAY_SECRET = (os.getenv("DISCORD_RELAY_SECRET") or os.getenv("DISCORD_PROXY_SECRET", "")).strip().strip("/") | |
| GHT = os.getenv("GHT", "").strip() | |
| GITHUB_DATA_REPO = os.getenv("GITHUB_DATA_REPO", "Wuhpondiscord/rldevs").strip() | |
| GITHUB_DATA_BRANCH = os.getenv("GITHUB_DATA_BRANCH", "main").strip() or "main" | |
| GITHUB_SYNC_ACTIVITY_LIMIT = max(100, int(os.getenv("GITHUB_SYNC_ACTIVITY_LIMIT", "5000"))) | |
| GITHUB_AUTO_SYNC_SECONDS = max(60, int(os.getenv("GITHUB_AUTO_SYNC_SECONDS", "300"))) | |
| logging.basicConfig( | |
| level=os.getenv("LOG_LEVEL", "INFO").upper(), | |
| format="%(asctime)s | %(levelname)s | %(name)s | %(message)s", | |
| ) | |
| log = logging.getLogger("rocket-watch") | |
| def configure_discord_relay() -> Optional[str]: | |
| """Route discord.py REST + Gateway traffic through the external Deno relay. | |
| The relay uses a secret path segment so it cannot be used as an open Discord | |
| proxy by anyone who does not know the shared secret. The bot token is not | |
| stored in the relay; discord.py sends it as the normal Authorization header. | |
| """ | |
| if not DISCORD_RELAY_ORIGIN and not DISCORD_RELAY_SECRET: | |
| return None | |
| if not DISCORD_RELAY_ORIGIN or not DISCORD_RELAY_SECRET: | |
| raise RuntimeError( | |
| "Set both DISCORD_RELAY_ORIGIN and DISCORD_RELAY_SECRET, or neither." | |
| ) | |
| if not DISCORD_RELAY_ORIGIN.startswith("https://"): | |
| raise RuntimeError("DISCORD_RELAY_ORIGIN must start with https://") | |
| from discord.http import Route | |
| from discord.gateway import DiscordWebSocket | |
| rest_base = f"{DISCORD_RELAY_ORIGIN}/p/{DISCORD_RELAY_SECRET}/api/v10" | |
| gateway = DISCORD_RELAY_ORIGIN.replace("https://", "wss://", 1) | |
| gateway = f"{gateway}/p/{DISCORD_RELAY_SECRET}/gateway" | |
| Route.BASE = rest_base | |
| DiscordWebSocket.DEFAULT_GATEWAY = yarl.URL(gateway) | |
| # Discord's READY event contains a direct resume_gateway_url. discord.py stores | |
| # that URL for later RESUME attempts. Restore our proxy URL immediately after | |
| # READY/RESUMED processing so reconnects never escape the relay. | |
| original_received_message = DiscordWebSocket.received_message | |
| async def proxied_received_message(self, msg): | |
| result = await original_received_message(self, msg) | |
| if getattr(self, "session_id", None) and str(getattr(self, "gateway", "")) != gateway: | |
| self.gateway = yarl.URL(gateway) | |
| return result | |
| if not getattr(DiscordWebSocket.received_message, "_rocket_watch_proxy_patch", False): | |
| proxied_received_message._rocket_watch_proxy_patch = True | |
| DiscordWebSocket.received_message = proxied_received_message | |
| log.info("Discord relay enabled through %s", DISCORD_RELAY_ORIGIN) | |
| return gateway | |
| DISCORD_RELAY_GATEWAY = configure_discord_relay() | |
| def utc_now_iso() -> str: | |
| return datetime.now(timezone.utc).isoformat() | |
| def parse_dt(value: Optional[str]) -> Optional[datetime]: | |
| if not value: | |
| return None | |
| try: | |
| dt = datetime.fromisoformat(value.replace("Z", "+00:00")) | |
| if dt.tzinfo is None: | |
| dt = dt.replace(tzinfo=timezone.utc) | |
| return dt.astimezone(timezone.utc) | |
| except ValueError: | |
| return None | |
| def age_text(value: Optional[str]) -> str: | |
| dt = parse_dt(value) | |
| if not dt: | |
| return "unknown" | |
| sec = max(0, int((datetime.now(timezone.utc) - dt).total_seconds())) | |
| if sec < 60: | |
| return f"{sec}s ago" | |
| if sec < 3600: | |
| return f"{sec // 60}m ago" | |
| if sec < 86400: | |
| return f"{sec // 3600}h {(sec % 3600) // 60}m ago" | |
| return f"{sec // 86400}d ago" | |
| def discord_time(value: Optional[str], style: str = "F") -> str: | |
| """Render a timestamp in each Discord viewer's own locale/timezone.""" | |
| dt = parse_dt(value) | |
| if not dt: | |
| return "unknown" | |
| return f"<t:{int(dt.timestamp())}:{style}>" | |
| def queue_window_text(value: Optional[str]) -> str: | |
| """Conservative inference from Tracker collection time; never claims live presence.""" | |
| dt = parse_dt(value) | |
| if not dt: | |
| return "Unknown" | |
| age_min = max(0, (datetime.now(timezone.utc) - dt).total_seconds() / 60) | |
| if age_min <= 3: | |
| return "**Very fresh result** — strongest window to try the same playlist/MMR" | |
| if age_min <= 8: | |
| return "**Fresh** — they may still be playing or re-queueing" | |
| if age_min <= 20: | |
| return "**Recent** — useful signal, but queue status is uncertain" | |
| return "**Stale** — weak signal for current matchmaking" | |
| def activity_label(value: Optional[str]) -> str: | |
| dt = parse_dt(value) | |
| if not dt: | |
| return "UNKNOWN" | |
| age = (datetime.now(timezone.utc) - dt).total_seconds() | |
| if age <= 8 * 60: | |
| return "HOT" | |
| if age <= 20 * 60: | |
| return "RECENT" | |
| if age <= 60 * 60: | |
| return "COOLING" | |
| return "INACTIVE" | |
| def trunc(s: str, n: int) -> str: | |
| return s if len(s) <= n else s[: n - 1] + "…" | |
| class Store: | |
| def __init__(self, path: Path): | |
| self.path = path | |
| self.path.parent.mkdir(parents=True, exist_ok=True) | |
| self._init() | |
| def conn(self): | |
| conn = sqlite3.connect(self.path, timeout=30) | |
| conn.row_factory = sqlite3.Row | |
| return conn | |
| def _init(self): | |
| with self.conn() as c: | |
| c.executescript( | |
| """ | |
| PRAGMA journal_mode=WAL; | |
| CREATE TABLE IF NOT EXISTS players ( | |
| id INTEGER PRIMARY KEY, | |
| guild_id INTEGER NOT NULL, | |
| label TEXT NOT NULL, | |
| platform TEXT NOT NULL, | |
| identifier TEXT NOT NULL, | |
| handle TEXT, | |
| tracker_player_id INTEGER, | |
| alert_channel_id INTEGER, | |
| enabled INTEGER NOT NULL DEFAULT 1, | |
| created_at TEXT NOT NULL, | |
| alert_watermark_at TEXT, | |
| UNIQUE(guild_id, platform, identifier) | |
| ); | |
| CREATE TABLE IF NOT EXISTS snapshots ( | |
| id INTEGER PRIMARY KEY, | |
| player_id INTEGER NOT NULL, | |
| observed_at TEXT NOT NULL, | |
| tracker_updated_at TEXT, | |
| playlist_json TEXT NOT NULL, | |
| FOREIGN KEY(player_id) REFERENCES players(id) | |
| ); | |
| CREATE INDEX IF NOT EXISTS idx_snapshots_player_time | |
| ON snapshots(player_id, observed_at DESC); | |
| CREATE TABLE IF NOT EXISTS matches ( | |
| id INTEGER PRIMARY KEY, | |
| player_id INTEGER NOT NULL, | |
| external_id TEXT NOT NULL, | |
| collected_at TEXT, | |
| playlist TEXT, | |
| result TEXT, | |
| rating INTEGER, | |
| rating_delta INTEGER, | |
| tier TEXT, | |
| division TEXT, | |
| is_grouped INTEGER, | |
| matches_played INTEGER, | |
| wins INTEGER, | |
| goals INTEGER, | |
| assists INTEGER, | |
| saves INTEGER, | |
| shots INTEGER, | |
| mvps INTEGER, | |
| first_seen_at TEXT NOT NULL, | |
| raw_json TEXT, | |
| FOREIGN KEY(player_id) REFERENCES players(id), | |
| UNIQUE(player_id, external_id) | |
| ); | |
| CREATE INDEX IF NOT EXISTS idx_matches_player_collected | |
| ON matches(player_id, collected_at DESC); | |
| """ | |
| ) | |
| cols = {row["name"] for row in c.execute("PRAGMA table_info(players)")} | |
| if "alert_watermark_at" not in cols: | |
| c.execute("ALTER TABLE players ADD COLUMN alert_watermark_at TEXT") | |
| def add_player(self, guild_id: int, label: str, platform: str, identifier: str, channel_id: int) -> int: | |
| with self.conn() as c: | |
| c.execute( | |
| """ | |
| INSERT INTO players(guild_id,label,platform,identifier,alert_channel_id,created_at) | |
| VALUES(?,?,?,?,?,?) | |
| ON CONFLICT(guild_id, platform, identifier) DO UPDATE SET | |
| label=excluded.label, | |
| alert_channel_id=excluded.alert_channel_id, | |
| enabled=1 | |
| """, | |
| (guild_id, label, platform, identifier, channel_id, utc_now_iso()), | |
| ) | |
| row = c.execute( | |
| "SELECT id FROM players WHERE guild_id=? AND platform=? AND identifier=?", | |
| (guild_id, platform, identifier), | |
| ).fetchone() | |
| return int(row["id"]) | |
| def remove_player(self, guild_id: int, query: str) -> bool: | |
| with self.conn() as c: | |
| cur = c.execute( | |
| "DELETE FROM players WHERE guild_id=? AND (label=? COLLATE NOCASE OR identifier=? COLLATE NOCASE)", | |
| (guild_id, query, query), | |
| ) | |
| return cur.rowcount > 0 | |
| def players(self, guild_id: Optional[int] = None): | |
| with self.conn() as c: | |
| if guild_id is None: | |
| return c.execute("SELECT * FROM players WHERE enabled=1 ORDER BY guild_id,label").fetchall() | |
| return c.execute( | |
| "SELECT * FROM players WHERE enabled=1 AND guild_id=? ORDER BY label COLLATE NOCASE", | |
| (guild_id,), | |
| ).fetchall() | |
| def get_player(self, guild_id: int, query: str): | |
| with self.conn() as c: | |
| return c.execute( | |
| """ | |
| SELECT * FROM players | |
| WHERE guild_id=? AND enabled=1 | |
| AND (label=? COLLATE NOCASE OR handle=? COLLATE NOCASE OR identifier=? COLLATE NOCASE) | |
| LIMIT 1 | |
| """, | |
| (guild_id, query, query, query), | |
| ).fetchone() | |
| def update_identity(self, player_id: int, handle: str, tracker_player_id: Optional[int]): | |
| with self.conn() as c: | |
| c.execute( | |
| "UPDATE players SET handle=?, tracker_player_id=? WHERE id=?", | |
| (handle, tracker_player_id, player_id), | |
| ) | |
| def set_alert_watermark(self, player_id: int, value: Optional[str]) -> None: | |
| if not value: | |
| return | |
| with self.conn() as c: | |
| c.execute("UPDATE players SET alert_watermark_at=? WHERE id=?", (value, player_id)) | |
| def ensure_alert_watermark(self, player_id: int) -> Optional[str]: | |
| with self.conn() as c: | |
| row = c.execute("SELECT alert_watermark_at, created_at FROM players WHERE id=?", (player_id,)).fetchone() | |
| if not row: | |
| return None | |
| if row["alert_watermark_at"]: | |
| return row["alert_watermark_at"] | |
| latest = c.execute( | |
| "SELECT collected_at FROM matches WHERE player_id=? AND collected_at IS NOT NULL ORDER BY collected_at DESC LIMIT 1", | |
| (player_id,), | |
| ).fetchone() | |
| value = latest["collected_at"] if latest else row["created_at"] | |
| c.execute("UPDATE players SET alert_watermark_at=? WHERE id=?", (value, player_id)) | |
| return value | |
| def add_snapshot(self, player_id: int, profile) -> None: | |
| with self.conn() as c: | |
| c.execute( | |
| "INSERT INTO snapshots(player_id,observed_at,tracker_updated_at,playlist_json) VALUES(?,?,?,?)", | |
| ( | |
| player_id, | |
| utc_now_iso(), | |
| profile.last_updated.isoformat() if profile.last_updated else None, | |
| json.dumps(profile.playlists, separators=(",", ":")), | |
| ), | |
| ) | |
| def latest_snapshot(self, player_id: int): | |
| with self.conn() as c: | |
| return c.execute( | |
| "SELECT * FROM snapshots WHERE player_id=? ORDER BY observed_at DESC LIMIT 1", | |
| (player_id,), | |
| ).fetchone() | |
| def insert_matches(self, player_id: int, matches: list[dict[str, Any]]) -> list[dict[str, Any]]: | |
| newly_added: list[dict[str, Any]] = [] | |
| with self.conn() as c: | |
| for m in matches: | |
| ext = m.get("match_id") | |
| if not ext: | |
| continue | |
| cur = c.execute( | |
| """ | |
| INSERT OR IGNORE INTO matches( | |
| player_id,external_id,collected_at,playlist,result,rating,rating_delta,tier,division, | |
| is_grouped,matches_played,wins,goals,assists,saves,shots,mvps,first_seen_at,raw_json | |
| ) VALUES(?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?) | |
| """, | |
| ( | |
| player_id, ext, m.get("collected_at"), m.get("playlist"), m.get("result"), | |
| m.get("rating"), m.get("rating_delta"), m.get("tier"), m.get("division"), | |
| int(bool(m.get("is_grouped"))), m.get("matches_played"), m.get("wins"), | |
| m.get("goals"), m.get("assists"), m.get("saves"), m.get("shots"), m.get("mvps"), | |
| utc_now_iso(), json.dumps(m, separators=(",", ":")), | |
| ), | |
| ) | |
| if cur.rowcount: | |
| newly_added.append(m) | |
| return newly_added | |
| def recent_matches(self, player_id: int, limit: int = 10): | |
| with self.conn() as c: | |
| return c.execute( | |
| "SELECT * FROM matches WHERE player_id=? ORDER BY collected_at DESC LIMIT ?", | |
| (player_id, limit), | |
| ).fetchall() | |
| def latest_match(self, player_id: int): | |
| rows = self.recent_matches(player_id, 1) | |
| return rows[0] if rows else None | |
| def matches_since(self, player_id: int, since_iso: str): | |
| with self.conn() as c: | |
| return c.execute( | |
| """ | |
| SELECT * FROM matches | |
| WHERE player_id=? AND collected_at IS NOT NULL AND collected_at>=? | |
| ORDER BY collected_at ASC, id ASC | |
| """, | |
| (player_id, since_iso), | |
| ).fetchall() | |
| def export_players(self) -> list[dict[str, Any]]: | |
| with self.conn() as c: | |
| rows = c.execute("SELECT * FROM players ORDER BY guild_id,label COLLATE NOCASE").fetchall() | |
| return [dict(r) for r in rows] | |
| def export_matches(self, limit: int = 5000) -> list[dict[str, Any]]: | |
| with self.conn() as c: | |
| rows = c.execute( | |
| """ | |
| SELECT p.guild_id,p.platform,p.identifier,p.label,m.* | |
| FROM matches m JOIN players p ON p.id=m.player_id | |
| ORDER BY m.collected_at DESC, m.id DESC LIMIT ? | |
| """, | |
| (limit,), | |
| ).fetchall() | |
| out = [] | |
| for r in rows: | |
| d = dict(r) | |
| d.pop("id", None) | |
| d.pop("player_id", None) | |
| out.append(d) | |
| return out | |
| def restore_player(self, item: dict[str, Any]) -> Optional[int]: | |
| try: | |
| guild_id = int(item["guild_id"]) | |
| platform = str(item["platform"]) | |
| identifier = str(item["identifier"]) | |
| label = str(item.get("label") or identifier) | |
| channel_id = int(item.get("alert_channel_id") or 0) | |
| except (KeyError, TypeError, ValueError): | |
| return None | |
| with self.conn() as c: | |
| c.execute( | |
| """ | |
| INSERT INTO players( | |
| guild_id,label,platform,identifier,handle,tracker_player_id, | |
| alert_channel_id,enabled,created_at,alert_watermark_at | |
| ) VALUES(?,?,?,?,?,?,?,?,?,?) | |
| ON CONFLICT(guild_id,platform,identifier) DO UPDATE SET | |
| label=excluded.label, | |
| handle=COALESCE(excluded.handle,players.handle), | |
| tracker_player_id=COALESCE(excluded.tracker_player_id,players.tracker_player_id), | |
| alert_channel_id=CASE WHEN excluded.alert_channel_id>0 THEN excluded.alert_channel_id ELSE players.alert_channel_id END, | |
| enabled=excluded.enabled, | |
| alert_watermark_at=COALESCE(excluded.alert_watermark_at,players.alert_watermark_at) | |
| """, | |
| ( | |
| guild_id,label,platform,identifier,item.get("handle"),item.get("tracker_player_id"), | |
| channel_id,int(item.get("enabled",1)),item.get("created_at") or utc_now_iso(),item.get("alert_watermark_at"), | |
| ), | |
| ) | |
| row=c.execute( | |
| "SELECT id FROM players WHERE guild_id=? AND platform=? AND identifier=?", | |
| (guild_id,platform,identifier), | |
| ).fetchone() | |
| return int(row["id"]) if row else None | |
| def restore_match(self, player_id: int, item: dict[str, Any]) -> None: | |
| ext = item.get("external_id") or item.get("match_id") | |
| if not ext: | |
| return | |
| with self.conn() as c: | |
| c.execute( | |
| """ | |
| INSERT OR IGNORE INTO matches( | |
| player_id,external_id,collected_at,playlist,result,rating,rating_delta,tier,division, | |
| is_grouped,matches_played,wins,goals,assists,saves,shots,mvps,first_seen_at,raw_json | |
| ) VALUES(?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?) | |
| """, | |
| ( | |
| player_id,ext,item.get("collected_at"),item.get("playlist"),item.get("result"), | |
| item.get("rating"),item.get("rating_delta"),item.get("tier"),item.get("division"), | |
| int(bool(item.get("is_grouped"))),item.get("matches_played"),item.get("wins"), | |
| item.get("goals"),item.get("assists"),item.get("saves"),item.get("shots"),item.get("mvps"), | |
| item.get("first_seen_at") or utc_now_iso(),item.get("raw_json") or json.dumps(item,separators=(",",":")), | |
| ), | |
| ) | |
| class GitHubDataStore: | |
| API = "https://api.github.com" | |
| def __init__(self, token: str, repo: str, branch: str = "main"): | |
| self.token = token | |
| self.repo = repo | |
| self.branch = branch | |
| self.enabled = bool(token and repo) | |
| self.session = requests.Session() | |
| if self.enabled: | |
| self.session.headers.update({ | |
| "Authorization": f"Bearer {token}", | |
| "Accept": "application/vnd.github+json", | |
| "X-GitHub-Api-Version": "2022-11-28", | |
| "User-Agent": "rocket-watch-hf/1.0", | |
| }) | |
| self.last_sync_at: Optional[str] = None | |
| self.last_error: Optional[str] = None | |
| def _url(self, path: str) -> str: | |
| return f"{self.API}/repos/{self.repo}/contents/{path}" | |
| def read_json(self, path: str, default: dict[str, Any]) -> tuple[dict[str, Any], Optional[str]]: | |
| if not self.enabled: | |
| return default, None | |
| r = self.session.get(self._url(path), params={"ref": self.branch}, timeout=20) | |
| if r.status_code == 404: | |
| return default, None | |
| r.raise_for_status() | |
| payload = r.json() | |
| raw = base64.b64decode(payload.get("content", "")).decode("utf-8") | |
| return json.loads(raw), payload.get("sha") | |
| def write_json(self, path: str, data: dict[str, Any], message: str) -> None: | |
| if not self.enabled: | |
| return | |
| _, sha = self.read_json(path, {}) | |
| body = { | |
| "message": message, | |
| "content": base64.b64encode((json.dumps(data, indent=2, sort_keys=False) + "\n").encode()).decode(), | |
| "branch": self.branch, | |
| } | |
| if sha: | |
| body["sha"] = sha | |
| r = self.session.put(self._url(path), json=body, timeout=30) | |
| r.raise_for_status() | |
| def ensure_files(self) -> None: | |
| if not self.enabled: | |
| return | |
| defaults = { | |
| "datasets/users.json": {"schema_version": 1, "updated_at": None, "players": []}, | |
| "datasets/activity.json": {"schema_version": 1, "updated_at": None, "matches": []}, | |
| } | |
| for path, default in defaults.items(): | |
| data, sha = self.read_json(path, default) | |
| if sha is None: | |
| self.write_json(path, data, f"Create {path}") | |
| def restore_into(self, store: Store) -> dict[str, int]: | |
| if not self.enabled: | |
| return {"players": 0, "matches": 0} | |
| users, _ = self.read_json("datasets/users.json", {"players": []}) | |
| activity, _ = self.read_json("datasets/activity.json", {"matches": []}) | |
| key_to_id: dict[tuple[int,str,str], int] = {} | |
| restored_players = 0 | |
| for item in users.get("players") or []: | |
| pid = store.restore_player(item) | |
| if pid is not None: | |
| key_to_id[(int(item["guild_id"]), str(item["platform"]), str(item["identifier"]))] = pid | |
| restored_players += 1 | |
| restored_matches = 0 | |
| for item in activity.get("matches") or []: | |
| try: | |
| key = (int(item["guild_id"]), str(item["platform"]), str(item["identifier"])) | |
| except (KeyError, TypeError, ValueError): | |
| continue | |
| pid = key_to_id.get(key) | |
| if pid is None: | |
| row = store.get_player(key[0], key[2]) | |
| pid = int(row["id"]) if row else None | |
| if pid is not None: | |
| store.restore_match(pid, item) | |
| restored_matches += 1 | |
| self.last_sync_at = utc_now_iso() | |
| self.last_error = None | |
| return {"players": restored_players, "matches": restored_matches} | |
| def push_all(self, store: Store, reason: str = "Sync RL dev tracker datasets") -> None: | |
| if not self.enabled: | |
| return | |
| now = utc_now_iso() | |
| users = {"schema_version": 1, "updated_at": now, "players": store.export_players()} | |
| activity = {"schema_version": 1, "updated_at": now, "matches": store.export_matches(GITHUB_SYNC_ACTIVITY_LIMIT)} | |
| self.write_json("datasets/users.json", users, f"{reason}: users") | |
| self.write_json("datasets/activity.json", activity, f"{reason}: activity") | |
| self.last_sync_at = now | |
| self.last_error = None | |
| github_data = GitHubDataStore(GHT, GITHUB_DATA_REPO, GITHUB_DATA_BRANCH) | |
| store = Store(DB_PATH) | |
| tracker = TrackerClient() | |
| async def github_push(reason: str = "Sync RL dev tracker datasets") -> bool: | |
| if not github_data.enabled: | |
| return False | |
| try: | |
| await asyncio.to_thread(github_data.push_all, store, reason) | |
| return True | |
| except Exception as exc: | |
| github_data.last_error = f"{type(exc).__name__}: {exc}" | |
| log.warning("GitHub dataset sync failed: %s", exc) | |
| return False | |
| intents = discord.Intents.default() | |
| client = discord.Client(intents=intents) | |
| tree = app_commands.CommandTree(client) | |
| # Runtime state used by the health endpoint. | |
| runtime = { | |
| "discord_connected": False, | |
| "discord_user": None, | |
| "discord_last_error": None, | |
| "discord_last_attempt": None, | |
| "discord_last_connected": None, | |
| "discord_retry_seconds": None, | |
| "discord_stage": "idle", | |
| "discord_stage_since": None, | |
| } | |
| async def fetch_and_store(player_row, *, seed: bool = False) -> tuple[Any, list[dict[str, Any]]]: | |
| profile = await asyncio.to_thread(tracker.profile, player_row["platform"], player_row["identifier"]) | |
| sessions_payload = await asyncio.to_thread(tracker.sessions, player_row["platform"], player_row["identifier"]) | |
| matches = tracker.flatten_session_matches(sessions_payload) | |
| store.update_identity(player_row["id"], profile.handle, profile.player_id) | |
| store.add_snapshot(player_row["id"], profile) | |
| new_matches = store.insert_matches(player_row["id"], matches) | |
| # On initial add, establish a watermark at the newest historical match. Tracker can | |
| # reveal older/grouped rows later, and those must never be announced as live activity. | |
| if seed: | |
| dated = [m for m in matches if parse_dt(m.get("collected_at"))] | |
| if dated: | |
| newest = max(dated, key=lambda m: parse_dt(m.get("collected_at"))) | |
| store.set_alert_watermark(player_row["id"], newest.get("collected_at")) | |
| else: | |
| store.set_alert_watermark(player_row["id"], utc_now_iso()) | |
| return profile, [] | |
| return profile, new_matches | |
| def profile_embed(player_row, profile, latest_match=None) -> discord.Embed: | |
| e = discord.Embed( | |
| title=f"{profile.handle} — Rocket League", | |
| description=f"{profile.platform.upper()} • Season {profile.season or '?'}", | |
| color=discord.Color.blurple(), | |
| timestamp=datetime.now(timezone.utc), | |
| ) | |
| if profile.avatar_url: | |
| e.set_thumbnail(url=profile.avatar_url) | |
| if latest_match: | |
| e.add_field( | |
| name="Activity", | |
| value=f"**{activity_label(latest_match['collected_at'])}** • {discord_time(latest_match['collected_at'], 'R')}\n{discord_time(latest_match['collected_at'], 'f')}", | |
| inline=True, | |
| ) | |
| e.add_field(name="Last mode", value=latest_match["playlist"] or "Unknown", inline=True) | |
| if latest_match["rating"] is not None: | |
| delta = latest_match["rating_delta"] | |
| ds = f" ({delta:+d})" if delta is not None else "" | |
| e.add_field(name="Last MMR", value=f"{latest_match['rating']}{ds}", inline=True) | |
| ranked = [p for p in profile.playlists if p.get("rating") is not None] | |
| ranked.sort(key=lambda p: ("Ranked" not in (p.get("name") or ""), p.get("name") or "")) | |
| for p in ranked[:10]: | |
| rank = " ".join(x for x in [p.get("tier"), p.get("division")] if x) or "—" | |
| peak = f" • peak {p['peak_rating']}" if p.get("peak_rating") is not None else "" | |
| e.add_field( | |
| name=trunc(p.get("name") or "Playlist", 256), | |
| value=f"**{p['rating']} MMR** • {rank}{peak}", | |
| inline=False, | |
| ) | |
| e.set_footer(text="Tracker collection times can lag actual in-game activity.") | |
| return e | |
| def match_embed(player_row, match: dict[str, Any]) -> discord.Embed: | |
| rating = match.get("rating") | |
| delta = match.get("rating_delta") | |
| mmr = "Unknown" | |
| if rating is not None: | |
| mmr = str(rating) | |
| if delta is not None: | |
| mmr += f" ({delta:+d})" | |
| result = match.get("result") or "activity" | |
| e = discord.Embed( | |
| title=f"New RL activity — {player_row['label']}", | |
| description=f"**{match.get('playlist') or 'Unknown playlist'}** • {result}", | |
| color=discord.Color.green() if str(result).lower() == "victory" else discord.Color.orange(), | |
| timestamp=parse_dt(match.get("collected_at")) or datetime.now(timezone.utc), | |
| ) | |
| e.add_field(name="MMR", value=mmr, inline=True) | |
| rank = " ".join(x for x in [match.get("tier"), match.get("division")] if x) | |
| if rank: | |
| e.add_field(name="Rank", value=rank, inline=True) | |
| e.add_field( | |
| name="Observed", | |
| value=f"{discord_time(match.get('collected_at'), 'R')}\n{discord_time(match.get('collected_at'), 'f')}", | |
| inline=True, | |
| ) | |
| e.add_field(name="Matchmaking signal", value=queue_window_text(match.get("collected_at")), inline=False) | |
| stats = [] | |
| for k, label in [("goals", "G"), ("assists", "A"), ("saves", "S"), ("shots", "Shots")]: | |
| if match.get(k) is not None: | |
| stats.append(f"{label}: {match[k]}") | |
| if stats: | |
| e.add_field(name="Stats", value=" • ".join(stats), inline=False) | |
| e.set_footer(text="Activity is inferred from Tracker updates; it is not a direct online-status signal.") | |
| return e | |
| def build_activity_graph(matches, label: str, range_days: int, style: str = "bar") -> tuple[BytesIO, dict[str, Any], str]: | |
| valid = [] | |
| for m in matches: | |
| if m["is_grouped"]: | |
| continue | |
| dt = parse_dt(m["collected_at"]) | |
| if not dt: | |
| continue | |
| valid.append((m, dt)) | |
| if not valid: | |
| raise ValueError("No non-grouped matches in the selected time range.") | |
| hourly = [0] * 24 | |
| dow_hour = [[0 for _ in range(24)] for _ in range(7)] | |
| playlist_counter: Counter[str] = Counter() | |
| active_days = set() | |
| for m, dt in valid: | |
| hourly[dt.hour] += 1 | |
| dow_hour[dt.weekday()][dt.hour] += 1 | |
| playlist_counter[m["playlist"] or "Unknown"] += 1 | |
| active_days.add(dt.date().isoformat()) | |
| top_hour = max(range(24), key=lambda h: hourly[h]) | |
| top_hour_count = hourly[top_hour] | |
| top_day_index = max(range(7), key=lambda d: sum(dow_hour[d])) | |
| day_names = ["Mon", "Tue", "Wed", "Thu", "Fri", "Sat", "Sun"] | |
| top_day_count = sum(dow_hour[top_day_index]) | |
| avg_per_active_day = len(valid) / max(1, len(active_days)) | |
| top_playlist, top_playlist_count = playlist_counter.most_common(1)[0] | |
| fig = plt.figure(figsize=(10.5, 4.8 if style == "bar" else 5.8)) | |
| ax = fig.add_subplot(111) | |
| title = f"{label} activity by time of day — last {range_days} day(s)" | |
| if style == "heatmap": | |
| im = ax.imshow(dow_hour, aspect="auto", interpolation="nearest") | |
| ax.set_title(title + " (UTC)") | |
| ax.set_xlabel("Hour of day (UTC)") | |
| ax.set_ylabel("Day of week") | |
| ax.set_xticks(range(0, 24, 1)) | |
| ax.set_xticklabels([str(h) for h in range(24)], fontsize=8) | |
| ax.set_yticks(range(7)) | |
| ax.set_yticklabels(day_names) | |
| fig.colorbar(im, ax=ax, label="Observed matches") | |
| caption = "Heatmap uses UTC. Static images cannot auto-convert to each viewer's local timezone." | |
| else: | |
| ax.bar(list(range(24)), hourly) | |
| ax.set_title(title + " (UTC)") | |
| ax.set_xlabel("Hour of day (UTC)") | |
| ax.set_ylabel("Observed matches") | |
| ax.set_xticks(range(24)) | |
| ax.set_xlim(-0.6, 23.6) | |
| ax.grid(True, axis="y", alpha=0.3) | |
| caption = "Bar chart uses UTC. Static images cannot auto-convert to each viewer's local timezone." | |
| fig.tight_layout() | |
| buf = BytesIO() | |
| fig.savefig(buf, format="png", dpi=170, bbox_inches="tight") | |
| plt.close(fig) | |
| buf.seek(0) | |
| stats = { | |
| "match_count": len(valid), | |
| "active_days": len(active_days), | |
| "avg_per_active_day": avg_per_active_day, | |
| "top_hour": top_hour, | |
| "top_hour_count": top_hour_count, | |
| "top_day": day_names[top_day_index], | |
| "top_day_count": top_day_count, | |
| "top_playlist": top_playlist, | |
| "top_playlist_count": top_playlist_count, | |
| } | |
| return buf, stats, caption | |
| async def on_ready(): | |
| runtime["discord_connected"] = True | |
| runtime["discord_user"] = str(client.user) if client.user else None | |
| runtime["discord_last_connected"] = utc_now_iso() | |
| runtime["discord_last_error"] = None | |
| runtime["discord_retry_seconds"] = None | |
| log.info("Logged in as %s (%s)", client.user, client.user.id if client.user else "?") | |
| if not watcher.is_running(): | |
| watcher.change_interval(seconds=POLL_SECONDS) | |
| watcher.start() | |
| if github_data.enabled and not github_auto_sync.is_running(): | |
| github_auto_sync.change_interval(seconds=GITHUB_AUTO_SYNC_SECONDS) | |
| github_auto_sync.start() | |
| async def on_disconnect(): | |
| runtime["discord_connected"] = False | |
| async def help_cmd(interaction: discord.Interaction): | |
| e = discord.Embed( | |
| title="Dev Stalker — Help", | |
| description=( | |
| "Tracks public Rocket League Tracker activity for selected accounts. " | |
| "Tracker data can lag, so freshness is an inference rather than direct online/queue status." | |
| ), | |
| color=discord.Color.blurple(), | |
| ) | |
| e.add_field( | |
| name="Tracking", | |
| value=( | |
| "`/track_add` — add an Epic/Steam/Xbox/PSN account and alert this channel.\n" | |
| "`/track_remove` — stop tracking an account.\n" | |
| "`/track_list` — list accounts tracked in this server.\n" | |
| "`/rl_refresh` — force an immediate Tracker refresh for all tracked accounts." | |
| ), | |
| inline=False, | |
| ) | |
| e.add_field( | |
| name="Player info", | |
| value=( | |
| "`/rl_player` — current profile, ranks/MMR, and latest observed activity.\n" | |
| "`/rl_recent` — recent recorded matches/activity with local Discord timestamps.\n" | |
| "`/activity_graph` — graph the hours/day patterns for one tracked target or all tracked targets over a time range.\n" | |
| "`/watch_status` — compact playlist/MMR/freshness view for one target.\n" | |
| "`/watch_hot` — rank tracked targets by freshest observed activity." | |
| ), | |
| inline=False, | |
| ) | |
| e.add_field( | |
| name="GitHub persistence", | |
| value=( | |
| "GitHub backup is automatic. Adds/removes/new activity trigger immediate syncs, and a " | |
| f"background safety sync runs about every **{GITHUB_AUTO_SYNC_SECONDS // 60 or 1} minute(s)**.\n" | |
| "`/github_status` — show persistence health and last successful sync.\n" | |
| "`/github_sync` — manually force a backup if you suspect something went wrong." | |
| ), | |
| inline=False, | |
| ) | |
| e.add_field( | |
| name="Alert behavior", | |
| value=( | |
| "Historical matches are seeded silently when a player is first added. Grouped `Multiple` rows are " | |
| "kept in history but suppressed from live alerts. Discord timestamps render in each viewer's local timezone." | |
| ), | |
| inline=False, | |
| ) | |
| e.set_footer(text="Use /github_status if persistence ever looks unhealthy.") | |
| await interaction.response.send_message(embed=e, ephemeral=True) | |
| async def track_add(interaction: discord.Interaction, platform: str, identifier: str, label: Optional[str] = None): | |
| if not interaction.guild_id or not interaction.channel_id: | |
| await interaction.response.send_message("Use this command inside a server channel.", ephemeral=True) | |
| return | |
| platform = platform.strip().lower() | |
| aliases = {"xbox": "xbl", "playstation": "psn", "epicgames": "epic"} | |
| platform = aliases.get(platform, platform) | |
| if platform not in {"steam", "epic", "xbl", "psn"}: | |
| await interaction.response.send_message("Platform must be `steam`, `epic`, `xbl`, or `psn`.", ephemeral=True) | |
| return | |
| await interaction.response.defer(ephemeral=True) | |
| row_id = store.add_player(interaction.guild_id, label or identifier, platform, identifier.strip(), interaction.channel_id) | |
| row = store.get_player(interaction.guild_id, label or identifier) | |
| if not row: | |
| # If label changed on an existing record, fetch by identifier. | |
| row = store.get_player(interaction.guild_id, identifier.strip()) | |
| try: | |
| profile, _ = await fetch_and_store(row, seed=True) | |
| except TrackerError as exc: | |
| store.remove_player(interaction.guild_id, identifier.strip()) | |
| await interaction.followup.send(f"Could not load that Tracker profile: `{exc}`", ephemeral=True) | |
| return | |
| await github_push("Track add") | |
| await interaction.followup.send( | |
| f"Tracking **{profile.handle}** as **{label or profile.handle}**. New activity will be posted in <#{interaction.channel_id}>. Historical sessions were seeded without alerts.", | |
| ephemeral=True, | |
| ) | |
| async def track_remove(interaction: discord.Interaction, player: str): | |
| if not interaction.guild_id: | |
| await interaction.response.send_message("Use this in a server.", ephemeral=True) | |
| return | |
| ok = store.remove_player(interaction.guild_id, player.strip()) | |
| if ok: | |
| await github_push("Track remove") | |
| await interaction.response.send_message("Removed." if ok else "No matching tracked player.", ephemeral=True) | |
| async def track_list(interaction: discord.Interaction): | |
| if not interaction.guild_id: | |
| await interaction.response.send_message("Use this in a server.", ephemeral=True) | |
| return | |
| rows = store.players(interaction.guild_id) | |
| if not rows: | |
| await interaction.response.send_message("No players are being tracked yet.", ephemeral=True) | |
| return | |
| lines = [] | |
| for r in rows: | |
| last = store.latest_match(r["id"]) | |
| status = activity_label(last["collected_at"]) if last else "UNKNOWN" | |
| mode = last["playlist"] if last else "no session data" | |
| lines.append(f"• **{r['label']}** — `{r['platform']}` / `{r['identifier']}` — **{status}** — {mode}") | |
| await interaction.response.send_message("\n".join(lines), ephemeral=True) | |
| async def rl_player(interaction: discord.Interaction, player: str): | |
| if not interaction.guild_id: | |
| await interaction.response.send_message("Use this in a server.", ephemeral=True) | |
| return | |
| row = store.get_player(interaction.guild_id, player.strip()) | |
| if not row: | |
| await interaction.response.send_message("Player not found. Use `/track_list`.", ephemeral=True) | |
| return | |
| await interaction.response.defer() | |
| try: | |
| profile, _ = await fetch_and_store(row, seed=False) | |
| except TrackerError as exc: | |
| await interaction.followup.send(f"Tracker request failed: `{exc}`") | |
| return | |
| latest = store.latest_match(row["id"]) | |
| await interaction.followup.send(embed=profile_embed(row, profile, latest)) | |
| async def rl_recent(interaction: discord.Interaction, player: str, count: app_commands.Range[int, 1, 10] = 5): | |
| if not interaction.guild_id: | |
| await interaction.response.send_message("Use this in a server.", ephemeral=True) | |
| return | |
| row = store.get_player(interaction.guild_id, player.strip()) | |
| if not row: | |
| await interaction.response.send_message("Player not found. Use `/track_list`.", ephemeral=True) | |
| return | |
| matches = store.recent_matches(row["id"], count) | |
| if not matches: | |
| await interaction.response.send_message("No session records stored yet.") | |
| return | |
| lines = [] | |
| for m in matches: | |
| mmr = "?" if m["rating"] is None else str(m["rating"]) | |
| if m["rating_delta"] is not None: | |
| mmr += f" ({m['rating_delta']:+d})" | |
| lines.append( | |
| f"**{m['playlist'] or 'Unknown'}** — {m['result'] or 'activity'} — `{mmr} MMR` — " | |
| f"{discord_time(m['collected_at'], 'R')} ({discord_time(m['collected_at'], 't')})" | |
| ) | |
| e = discord.Embed(title=f"Recent activity — {row['label']}", description="\n".join(lines), color=discord.Color.blurple()) | |
| await interaction.response.send_message(embed=e) | |
| async def rl_refresh(interaction: discord.Interaction): | |
| if not interaction.guild_id: | |
| await interaction.response.send_message("Use this in a server.", ephemeral=True) | |
| return | |
| await interaction.response.defer(ephemeral=True) | |
| rows = store.players(interaction.guild_id) | |
| ok = 0 | |
| errors = [] | |
| for row in rows: | |
| try: | |
| await fetch_and_store(row, seed=False) | |
| ok += 1 | |
| except TrackerError as exc: | |
| errors.append(f"{row['label']}: {exc}") | |
| if ok: | |
| await github_push("Manual refresh") | |
| msg = f"Refreshed {ok}/{len(rows)} tracked players." | |
| if errors: | |
| msg += "\n" + "\n".join(f"• {trunc(e, 160)}" for e in errors[:5]) | |
| await interaction.followup.send(msg, ephemeral=True) | |
| async def watch_status(interaction: discord.Interaction, player: str): | |
| if not interaction.guild_id: | |
| await interaction.response.send_message("Use this in a server.", ephemeral=True) | |
| return | |
| row = store.get_player(interaction.guild_id, player.strip()) | |
| if not row: | |
| await interaction.response.send_message("Player not found. Use `/track_list`.", ephemeral=True) | |
| return | |
| latest = store.latest_match(row["id"]) | |
| if not latest: | |
| await interaction.response.send_message("No activity has been recorded for that player yet.") | |
| return | |
| rating = "?" if latest["rating"] is None else str(latest["rating"]) | |
| if latest["rating_delta"] is not None: | |
| rating += f" ({latest['rating_delta']:+d})" | |
| e = discord.Embed(title=f"Watch status — {row['label']}", color=discord.Color.blurple()) | |
| e.add_field(name="Playlist", value=latest["playlist"] or "Unknown", inline=True) | |
| e.add_field(name="MMR", value=rating, inline=True) | |
| e.add_field(name="Freshness", value=activity_label(latest["collected_at"]), inline=True) | |
| e.add_field(name="Observed", value=f"{discord_time(latest['collected_at'],'R')}\n{discord_time(latest['collected_at'],'f')}", inline=False) | |
| e.add_field(name="Matchmaking signal", value=queue_window_text(latest["collected_at"]), inline=False) | |
| e.set_footer(text="Tracker collection times can lag and do not prove the player is currently queued.") | |
| await interaction.response.send_message(embed=e) | |
| async def watch_hot(interaction: discord.Interaction): | |
| if not interaction.guild_id: | |
| await interaction.response.send_message("Use this in a server.", ephemeral=True) | |
| return | |
| entries=[] | |
| for row in store.players(interaction.guild_id): | |
| last=store.latest_match(row["id"]) | |
| if not last: | |
| continue | |
| dt=parse_dt(last["collected_at"]) | |
| if not dt: | |
| continue | |
| entries.append((dt,row,last)) | |
| if not entries: | |
| await interaction.response.send_message("No tracked activity is stored yet.", ephemeral=True) | |
| return | |
| entries.sort(key=lambda x:x[0], reverse=True) | |
| lines=[] | |
| for _,row,last in entries[:15]: | |
| mmr="?" if last["rating"] is None else str(last["rating"]) | |
| lines.append(f"**{row['label']}** — {last['playlist'] or 'Unknown'} — `{mmr} MMR` — **{activity_label(last['collected_at'])}** — {discord_time(last['collected_at'],'R')}") | |
| await interaction.response.send_message(embed=discord.Embed(title="Freshest tracked activity",description="\n".join(lines),color=discord.Color.blurple())) | |
| async def activity_graph( | |
| interaction: discord.Interaction, | |
| player: Optional[str] = None, | |
| range_days: app_commands.Range[int, 1, 365] = 30, | |
| style: app_commands.Choice[str] | None = None, | |
| ): | |
| if not interaction.guild_id: | |
| await interaction.response.send_message("Use this in a server.", ephemeral=True) | |
| return | |
| await interaction.response.defer() | |
| since_dt = datetime.now(timezone.utc) - timedelta(days=int(range_days)) | |
| selected_label = None | |
| matches = [] | |
| tracked_count = 0 | |
| if player and player.strip(): | |
| row = store.get_player(interaction.guild_id, player.strip()) | |
| if not row: | |
| await interaction.followup.send("Player not found. Use `/track_list`.") | |
| return | |
| selected_label = row["label"] | |
| matches = store.matches_since(row["id"], since_dt.isoformat()) | |
| tracked_count = 1 | |
| else: | |
| rows = store.players(interaction.guild_id) | |
| tracked_count = len(rows) | |
| if not rows: | |
| await interaction.followup.send("No players are being tracked yet.") | |
| return | |
| selected_label = f"All tracked players ({tracked_count})" | |
| for row in rows: | |
| matches.extend(store.matches_since(row["id"], since_dt.isoformat())) | |
| if not matches: | |
| await interaction.followup.send("No stored activity falls inside that time range yet.") | |
| return | |
| chosen_style = style.value if style else "bar" | |
| try: | |
| image_buf, stats, caption = build_activity_graph(matches, selected_label, int(range_days), chosen_style) | |
| except ValueError as exc: | |
| await interaction.followup.send(str(exc)) | |
| return | |
| safe_name = selected_label.replace(' ', '_').replace('(', '').replace(')', '').replace('/', '_') | |
| filename = f"activity_{safe_name}_{chosen_style}.png" | |
| file = discord.File(image_buf, filename=filename) | |
| embed = discord.Embed( | |
| title=f"Activity graph — {selected_label}", | |
| description=( | |
| f"Observed public Tracker activity over the last **{int(range_days)} day(s)**. " | |
| "Use this as a tendency signal, not exact online status." | |
| ), | |
| color=discord.Color.blurple(), | |
| ) | |
| embed.add_field(name="Scope", value=selected_label, inline=False) | |
| embed.add_field(name="Top hour (UTC)", value=f"`{stats['top_hour']:02d}:00–{(stats['top_hour'] + 1) % 24:02d}:00` • {stats['top_hour_count']} matches", inline=True) | |
| embed.add_field(name="Most active day", value=f"**{stats['top_day']}** • {stats['top_day_count']} matches", inline=True) | |
| embed.add_field(name="Top playlist", value=f"**{stats['top_playlist']}** • {stats['top_playlist_count']} matches", inline=True) | |
| embed.add_field(name="Observed matches", value=str(stats['match_count']), inline=True) | |
| embed.add_field(name="Active days", value=str(stats['active_days']), inline=True) | |
| embed.add_field(name="Avg / active day", value=f"{stats['avg_per_active_day']:.2f}", inline=True) | |
| embed.set_image(url=f"attachment://{filename}") | |
| embed.set_footer(text=caption) | |
| await interaction.followup.send(embed=embed, file=file) | |
| async def github_sync_cmd(interaction: discord.Interaction): | |
| await interaction.response.defer(ephemeral=True) | |
| if not github_data.enabled: | |
| await interaction.followup.send("GitHub persistence is disabled. Add `GHT` as a Hugging Face secret.", ephemeral=True) | |
| return | |
| ok=await github_push("Manual Discord sync") | |
| if ok: | |
| await interaction.followup.send(f"Synced datasets to `{GITHUB_DATA_REPO}`.", ephemeral=True) | |
| else: | |
| await interaction.followup.send(f"GitHub sync failed: `{trunc(github_data.last_error or 'unknown error',300)}`", ephemeral=True) | |
| async def github_status_cmd(interaction: discord.Interaction): | |
| state="enabled" if github_data.enabled else "disabled" | |
| msg=(f"GitHub persistence: **{state}**\nRepo: `{GITHUB_DATA_REPO}`\nBranch: `{GITHUB_DATA_BRANCH}`\n" | |
| f"Last sync: {discord_time(github_data.last_sync_at,'R') if github_data.last_sync_at else 'never'}") | |
| if github_data.last_error: | |
| msg += f"\nLast error: `{trunc(github_data.last_error,300)}`" | |
| await interaction.response.send_message(msg, ephemeral=True) | |
| async def github_auto_sync(): | |
| if not github_data.enabled: | |
| return | |
| ok = await github_push("Automatic safety sync") | |
| if ok: | |
| log.info("Automatic GitHub dataset sync completed") | |
| async def before_github_auto_sync(): | |
| await client.wait_until_ready() | |
| async def watcher(): | |
| rows = store.players() | |
| for row in rows: | |
| try: | |
| profile, new_matches = await fetch_and_store(row, seed=False) | |
| if not new_matches: | |
| continue | |
| watermark_raw = store.ensure_alert_watermark(row["id"]) | |
| watermark = parse_dt(watermark_raw) | |
| now = datetime.now(timezone.utc) | |
| filtered: list[dict[str, Any]] = [] | |
| for match in new_matches: | |
| collected = parse_dt(match.get("collected_at")) | |
| if not collected: | |
| continue | |
| # Grouped/aggregate Tracker rows such as "Multiple" can be backfilled later and | |
| # are not reliable enough for live matchmaking alerts. Keep them in history only. | |
| if match.get("is_grouped"): | |
| continue | |
| if watermark and collected <= watermark: | |
| continue | |
| if (now - collected).total_seconds() > ALERT_MAX_AGE_MINUTES * 60: | |
| continue | |
| filtered.append(match) | |
| if not filtered: | |
| continue | |
| filtered.sort(key=lambda m: m.get("collected_at") or "", reverse=True) | |
| channel = client.get_channel(row["alert_channel_id"]) | |
| if channel is None: | |
| try: | |
| channel = await client.fetch_channel(row["alert_channel_id"]) | |
| except discord.DiscordException: | |
| log.warning("Could not resolve channel %s", row["alert_channel_id"]) | |
| continue | |
| sent_any = False | |
| for match in filtered[:3]: | |
| try: | |
| await channel.send(embed=match_embed(row, match)) | |
| store.set_alert_watermark(row["id"], match.get("collected_at")) | |
| sent_any = True | |
| except discord.DiscordException as exc: | |
| log.warning("Failed sending alert for %s: %s", row["label"], exc) | |
| if sent_any: | |
| await github_push("New Tracker activity") | |
| except TrackerError as exc: | |
| log.warning("Tracker error for %s: %s", row["label"], exc) | |
| except Exception: | |
| log.exception("Unexpected watcher error for %s", row["label"]) | |
| await asyncio.sleep(1) | |
| async def before_watcher(): | |
| await client.wait_until_ready() | |
| # Hugging Face Docker Spaces expect an HTTP service. This tiny endpoint also gives | |
| # you a convenient uptime/health check while the Discord bot runs in the same process. | |
| health = FastAPI(title="Rocket Watch Health") | |
| def root(): | |
| return { | |
| "ok": True, | |
| "service": "rocket-watch-discord-bot", | |
| "discord_connected": runtime["discord_connected"], | |
| "discord_user": runtime["discord_user"], | |
| "discord_last_error": runtime["discord_last_error"], | |
| "discord_last_attempt": runtime["discord_last_attempt"], | |
| "discord_last_connected": runtime["discord_last_connected"], | |
| "discord_retry_seconds": runtime["discord_retry_seconds"], | |
| "discord_retry_max_seconds": max(10, int(os.getenv("DISCORD_RETRY_MAX_SECONDS", "30"))), | |
| "discord_stage": runtime["discord_stage"], | |
| "discord_stage_since": runtime["discord_stage_since"], | |
| "discord_relay_enabled": bool(DISCORD_RELAY_GATEWAY), | |
| "discord_relay_origin": DISCORD_RELAY_ORIGIN or None, | |
| "tracked_players": len(store.players()), | |
| "poll_seconds": POLL_SECONDS, | |
| "alert_max_age_minutes": ALERT_MAX_AGE_MINUTES, | |
| "github_persistence_enabled": github_data.enabled, | |
| "github_data_repo": GITHUB_DATA_REPO, | |
| "github_last_sync_at": github_data.last_sync_at, | |
| "github_auto_sync_seconds": GITHUB_AUTO_SYNC_SECONDS, | |
| "github_last_error": github_data.last_error, | |
| } | |
| def healthcheck(): | |
| return { | |
| "ok": True, | |
| "http": "online", | |
| "discord": "connected" if runtime["discord_connected"] else "disconnected", | |
| "discord_last_error": runtime["discord_last_error"], | |
| "discord_stage": runtime["discord_stage"], | |
| "discord_stage_since": runtime["discord_stage_since"], | |
| } | |
| def set_discord_stage(stage: str) -> None: | |
| runtime["discord_stage"] = stage | |
| runtime["discord_stage_since"] = utc_now_iso() | |
| async def debug_discord(): | |
| """Test both direct Discord and the configured Deno relay without a token.""" | |
| async def probe(label: str, url: str): | |
| started = time.monotonic() | |
| try: | |
| timeout = aiohttp.ClientTimeout(total=10, connect=5, sock_connect=5, sock_read=5) | |
| async with aiohttp.ClientSession(timeout=timeout) as session: | |
| async with session.get(url) as resp: | |
| body = await resp.text() | |
| return { | |
| "label": label, | |
| "url": url.replace(DISCORD_RELAY_SECRET, "***") if DISCORD_RELAY_SECRET else url, | |
| "status": resp.status, | |
| "body": body[:300], | |
| "elapsed_ms": round((time.monotonic() - started) * 1000), | |
| } | |
| except Exception as exc: | |
| return { | |
| "label": label, | |
| "url": url.replace(DISCORD_RELAY_SECRET, "***") if DISCORD_RELAY_SECRET else url, | |
| "error": f"{type(exc).__name__}: {exc}", | |
| "elapsed_ms": round((time.monotonic() - started) * 1000), | |
| } | |
| dns = None | |
| try: | |
| infos = await asyncio.wait_for( | |
| asyncio.get_running_loop().run_in_executor( | |
| None, lambda: socket.getaddrinfo("discord.com", 443, type=socket.SOCK_STREAM) | |
| ), | |
| timeout=5, | |
| ) | |
| dns = sorted({item[4][0] for item in infos})[:6] | |
| except Exception as exc: | |
| dns = f"{type(exc).__name__}: {exc}" | |
| result = { | |
| "discord_dns": dns, | |
| "direct": await probe("direct", "https://discord.com/api/v10/gateway"), | |
| "relay_enabled": bool(DISCORD_RELAY_GATEWAY), | |
| } | |
| if DISCORD_RELAY_GATEWAY: | |
| relay_url = f"{DISCORD_RELAY_ORIGIN}/p/{DISCORD_RELAY_SECRET}/api/v10/gateway" | |
| result["relay"] = await probe("deno-relay", relay_url) | |
| return result | |
| async def sync_commands(): | |
| """Sync slash commands after Discord is actually connected.""" | |
| try: | |
| if GUILD_ID: | |
| guild = discord.Object(id=int(GUILD_ID)) | |
| tree.copy_global_to(guild=guild) | |
| synced = await tree.sync(guild=guild) | |
| log.info("Synced %d slash commands to guild %s", len(synced), GUILD_ID) | |
| else: | |
| synced = await tree.sync() | |
| log.info("Synced %d global slash commands", len(synced)) | |
| except Exception: | |
| log.exception("Slash-command sync failed") | |
| def rebuild_discord_client() -> None: | |
| """Replace the Discord client with a brand-new HTTP/WebSocket stack. | |
| discord.py can close its internal aiohttp session after a failed login. | |
| Client.clear() resets cache/state but does not make that closed session usable | |
| again, so retries must use a fresh Client instance. | |
| """ | |
| global client, tree | |
| # Preserve the application-command objects already registered on the current | |
| # tree, then attach them to a fresh tree/client. | |
| commands = list(tree.get_commands()) | |
| new_client = discord.Client(intents=intents) | |
| new_tree = app_commands.CommandTree(new_client) | |
| for command in commands: | |
| new_tree.add_command(command) | |
| # Re-register event callbacks on the new Client. The callbacks intentionally | |
| # reference the module-level `client`, which is rebound below. | |
| new_client.event(on_ready) | |
| new_client.event(on_disconnect) | |
| client = new_client | |
| tree = new_tree | |
| async def discord_supervisor(): | |
| """Keep Discord failures isolated from the Hugging Face HTTP service. | |
| The Space listens only on $PORT (7860 by default). Discord uses outbound | |
| HTTPS/WSS and therefore does not need another local listening port. Every | |
| retry uses a completely fresh discord.Client so a failed aiohttp session | |
| cannot poison future attempts. | |
| """ | |
| delay = 3 | |
| # Keep recovery quick. discord.py's own reconnect loop can grow to very long | |
| # waits after repeated relay 503s, so this supervisor owns reconnects instead. | |
| max_delay = max(10, int(os.getenv("DISCORD_RETRY_MAX_SECONDS", "30"))) | |
| first_attempt = True | |
| while True: | |
| runtime["discord_last_attempt"] = utc_now_iso() | |
| runtime["discord_retry_seconds"] = None | |
| if not first_attempt: | |
| rebuild_discord_client() | |
| first_attempt = False | |
| connect_task = None | |
| try: | |
| log.info("Connecting to Discord with a fresh client...") | |
| set_discord_stage("rest_login") | |
| await asyncio.wait_for( | |
| client.login(DISCORD_BOT_TOKEN), | |
| timeout=DISCORD_LOGIN_TIMEOUT, | |
| ) | |
| log.info("Discord REST login succeeded as %s", client.user) | |
| set_discord_stage("command_sync") | |
| await asyncio.wait_for(sync_commands(), timeout=20) | |
| # client.connect() is intentionally long-running. Start it separately, | |
| # then require READY within a bounded amount of time. This distinguishes | |
| # a successful REST login from a Gateway/WebSocket failure. | |
| set_discord_stage("gateway_connect") | |
| # Important: do NOT let discord.py own the reconnect loop here. Its | |
| # exponential backoff can grow to many minutes if the relay returns | |
| # repeated 503s after a Deno Deploy eviction. A single gateway failure | |
| # should return control to this supervisor so it can rebuild the entire | |
| # Client/HTTP session and retry with a small capped delay. | |
| connect_task = asyncio.create_task(client.connect(reconnect=False), name="discord-gateway") | |
| await asyncio.wait_for(client.wait_until_ready(), timeout=DISCORD_READY_TIMEOUT) | |
| set_discord_stage("connected") | |
| log.info("Discord gateway is READY") | |
| delay = 3 | |
| # Stay attached until the gateway closes. With reconnect=False this | |
| # returns/raises immediately on a broken relay connection instead of | |
| # entering discord.py's long internal backoff loop. | |
| await connect_task | |
| runtime["discord_connected"] = False | |
| runtime["discord_last_error"] = "Discord connection closed" | |
| set_discord_stage("disconnected") | |
| delay = 3 | |
| except asyncio.CancelledError: | |
| try: | |
| if not client.is_closed(): | |
| await client.close() | |
| finally: | |
| raise | |
| except asyncio.TimeoutError: | |
| stage = runtime.get("discord_stage") or "unknown" | |
| timeout_used = DISCORD_LOGIN_TIMEOUT if stage == "rest_login" else DISCORD_READY_TIMEOUT | |
| runtime["discord_connected"] = False | |
| runtime["discord_last_error"] = f"Timeout while in {stage} after {timeout_used}s" | |
| log.error("Discord timed out during %s after %ss", stage, timeout_used) | |
| except discord.LoginFailure as exc: | |
| runtime["discord_connected"] = False | |
| runtime["discord_last_error"] = f"Login failure: {exc}" | |
| delay = 300 | |
| log.error("Discord rejected the bot token: %s", exc) | |
| except Exception as exc: | |
| runtime["discord_connected"] = False | |
| runtime["discord_last_error"] = f"{type(exc).__name__}: {exc}" | |
| log.exception("Discord connection attempt failed") | |
| # Best-effort cleanup. We never reuse this instance after a failure. | |
| if connect_task is not None and not connect_task.done(): | |
| connect_task.cancel() | |
| try: | |
| await connect_task | |
| except (asyncio.CancelledError, Exception): | |
| pass | |
| try: | |
| if not client.is_closed(): | |
| await client.close() | |
| except Exception: | |
| log.debug("Discord client cleanup failed", exc_info=True) | |
| runtime["discord_retry_seconds"] = delay | |
| set_discord_stage("retry_wait") | |
| log.warning("Discord unavailable; retrying with a fresh client in %ss", delay) | |
| await asyncio.sleep(delay) | |
| delay = min(delay * 2, max_delay) | |
| async def main(): | |
| if not DISCORD_BOT_TOKEN: | |
| raise RuntimeError( | |
| "DISCORD_BOT_TOKEN is not set. Add it as a Hugging Face Space secret or environment variable." | |
| ) | |
| if github_data.enabled: | |
| try: | |
| await asyncio.to_thread(github_data.ensure_files) | |
| restored = await asyncio.to_thread(github_data.restore_into, store) | |
| log.info("Restored %s players and %s activity records from GitHub", restored["players"], restored["matches"]) | |
| await github_push("Startup reconciliation") | |
| except Exception as exc: | |
| github_data.last_error = f"{type(exc).__name__}: {exc}" | |
| log.warning("GitHub startup restore failed: %s", exc) | |
| # Hugging Face exposes one HTTP port. Uvicorn owns that port; Discord runs as | |
| # an outbound background connection and never binds a local port. | |
| port = int(os.getenv("PORT", "7860")) | |
| config = uvicorn.Config( | |
| health, | |
| host="0.0.0.0", | |
| port=port, | |
| log_level=os.getenv("UVICORN_LOG_LEVEL", "info").lower(), | |
| lifespan="on", | |
| ) | |
| server = uvicorn.Server(config) | |
| discord_task = asyncio.create_task(discord_supervisor(), name="discord-supervisor") | |
| try: | |
| await server.serve() | |
| finally: | |
| discord_task.cancel() | |
| try: | |
| await discord_task | |
| except asyncio.CancelledError: | |
| pass | |
| if not client.is_closed(): | |
| await client.close() | |
| if __name__ == "__main__": | |
| asyncio.run(main()) | |