clxud
224 lines
8.0 KiB
Python
224 lines
8.0 KiB
Python
"""Cross-bot shared JSON state: account links + blacklist.
|
|
|
|
Both SREBOT and TSSBOT put ``BOTS/SHARED`` on ``sys.path`` at startup, so this
|
|
module is importable by bare name (``from shared_store import ...``) from either
|
|
bot.
|
|
|
|
Two files are managed here:
|
|
|
|
* ``BLACKLIST.json`` — admin-curated, version-controlled, lives next to this
|
|
module in ``BOTS/SHARED``. Sections: ``discord_users`` (entries may be a plain
|
|
id, ``[id, comment, reason]``, or ``{"id": id, "comment": "...", "reason": "..."}``)
|
|
``squadrons`` (short tags, SRE-only — TSS teams are intentionally not
|
|
blacklistable), and ``guilds`` (Discord server ids, SRE-only — TSS does not
|
|
yet honour a guild blacklist). Read by both bots.
|
|
* ``PLAYERS.json`` — runtime-written Discord-user → WT-account links. Lives in
|
|
``$STORAGE_VOL_PATH`` (the storage volume both bots share, right next to
|
|
SQUADRONS.json), so SREBOT and TSSBOT read/write the same links.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import logging
|
|
import os
|
|
import threading
|
|
import time
|
|
from pathlib import Path
|
|
from typing import Any, Optional
|
|
|
|
log = logging.getLogger("shared_store")
|
|
|
|
_LOCK = threading.Lock()
|
|
|
|
# BLACKLIST.json lives alongside this module in BOTS/SHARED (version-controlled).
|
|
BLACKLIST_PATH: Path = Path(__file__).resolve().parent / "BLACKLIST.json"
|
|
|
|
PLAYERS_FILENAME = "PLAYERS.json"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Paths / IO
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def storage_dir() -> Path:
|
|
"""Return the shared storage volume root (where PLAYERS.json lives).
|
|
|
|
Both bots point ``STORAGE_VOL_PATH`` at the same volume, so PLAYERS.json
|
|
there is read/written by SREBOT and TSSBOT alike, next to SQUADRONS.json.
|
|
"""
|
|
raw = os.environ.get("STORAGE_VOL_PATH", "").strip()
|
|
if not raw:
|
|
raise RuntimeError("STORAGE_VOL_PATH must be set")
|
|
p = Path(raw).expanduser()
|
|
p.mkdir(parents=True, exist_ok=True)
|
|
return p
|
|
|
|
|
|
def _players_path() -> Path:
|
|
return storage_dir() / PLAYERS_FILENAME
|
|
|
|
|
|
def _read_json(path: Path, default: Any) -> Any:
|
|
try:
|
|
with open(path, "r", encoding="utf-8") as f:
|
|
return json.load(f)
|
|
except FileNotFoundError:
|
|
return default
|
|
except (json.JSONDecodeError, OSError) as e:
|
|
log.error("failed reading %s: %s", path, e)
|
|
return default
|
|
|
|
|
|
def _write_json_atomic(path: Path, data: Any) -> bool:
|
|
tmp = path.with_suffix(path.suffix + ".tmp")
|
|
try:
|
|
with open(tmp, "w", encoding="utf-8") as f:
|
|
json.dump(data, f, indent=4)
|
|
os.replace(tmp, path)
|
|
return True
|
|
except OSError as e:
|
|
log.error("failed writing %s: %s", path, e)
|
|
return False
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# PLAYERS.json — {discord_user_id: {"uid": str, "linked_unix": int}}
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def load_player_links() -> dict[str, dict[str, Any]]:
|
|
"""Return all Discord-user → WT-account links."""
|
|
data = _read_json(_players_path(), {})
|
|
return data if isinstance(data, dict) else {}
|
|
|
|
|
|
def get_linked_uid(discord_user_id: int | str) -> Optional[str]:
|
|
"""Return the WT UID linked to a Discord user, or None."""
|
|
entry = load_player_links().get(str(discord_user_id))
|
|
if isinstance(entry, dict):
|
|
uid = entry.get("uid")
|
|
return str(uid) if uid not in (None, "") else None
|
|
return None
|
|
|
|
|
|
def save_player_link(discord_user_id: int | str, uid: str) -> bool:
|
|
"""Link a Discord user to a WT UID (overwrites any existing link)."""
|
|
with _LOCK:
|
|
links = load_player_links()
|
|
links[str(discord_user_id)] = {"uid": str(uid), "linked_unix": int(time.time())}
|
|
return _write_json_atomic(_players_path(), links)
|
|
|
|
|
|
def remove_player_link(discord_user_id: int | str) -> bool:
|
|
"""Remove a Discord user's link. Returns True if a link was removed."""
|
|
with _LOCK:
|
|
links = load_player_links()
|
|
if str(discord_user_id) in links:
|
|
del links[str(discord_user_id)]
|
|
return _write_json_atomic(_players_path(), links)
|
|
return False
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# BLACKLIST.json — {"discord_users": [...], "squadrons": [...]}
|
|
# ---------------------------------------------------------------------------
|
|
|
|
_BLACKLIST_CACHE: Optional[dict[str, Any]] = None
|
|
_BLACKLIST_CACHE_TS: float = 0.0
|
|
_BLACKLIST_TTL = 60.0
|
|
|
|
|
|
def _default_blacklist() -> dict[str, list]:
|
|
return {"discord_users": [], "squadrons": [], "guilds": []}
|
|
|
|
|
|
def load_blacklist(force: bool = False) -> dict[str, Any]:
|
|
"""Return the blacklist dict, cached for a short TTL (reloadable with force)."""
|
|
global _BLACKLIST_CACHE, _BLACKLIST_CACHE_TS
|
|
now = time.time()
|
|
if not force and _BLACKLIST_CACHE is not None and (now - _BLACKLIST_CACHE_TS) < _BLACKLIST_TTL:
|
|
return _BLACKLIST_CACHE
|
|
data = _read_json(BLACKLIST_PATH, _default_blacklist())
|
|
if not isinstance(data, dict):
|
|
data = _default_blacklist()
|
|
data.setdefault("discord_users", [])
|
|
data.setdefault("squadrons", [])
|
|
data.setdefault("guilds", [])
|
|
_BLACKLIST_CACHE = data
|
|
_BLACKLIST_CACHE_TS = now
|
|
return data
|
|
|
|
|
|
def blacklisted_squadrons() -> list[str]:
|
|
"""Return blacklisted squadron short tags (SRE autolog matching)."""
|
|
return [str(s) for s in load_blacklist().get("squadrons", [])]
|
|
|
|
|
|
def blacklisted_guilds() -> list[int]:
|
|
"""Return blacklisted Discord guild (server) IDs.
|
|
|
|
Entries may be a plain id, ``[id, comment, reason]``, or a dict with
|
|
``id``/``guild_id`` plus optional ``comment``/``reason`` (mirrors the
|
|
``discord_users`` format). Comment/reason are for readability only and are
|
|
ignored here. Unparseable entries are skipped.
|
|
"""
|
|
out: list[int] = []
|
|
for entry in load_blacklist().get("guilds", []):
|
|
if isinstance(entry, dict):
|
|
raw = entry.get("id", entry.get("guild_id"))
|
|
elif isinstance(entry, (list, tuple)):
|
|
raw = entry[0] if entry else None
|
|
else:
|
|
raw = entry
|
|
if raw is None:
|
|
continue
|
|
try:
|
|
out.append(int(raw))
|
|
except (TypeError, ValueError):
|
|
continue
|
|
return out
|
|
|
|
|
|
def check_guild_blacklist(guild_id: int | str) -> tuple[bool, Optional[str]]:
|
|
"""Return (is_blacklisted, reason) for a Discord guild (server).
|
|
|
|
Entries follow the same flexible format as ``discord_users`` (plain id,
|
|
``[id, comment, reason]``, or a dict with ``id``/``guild_id`` plus optional
|
|
``reason``). SRE-only — see the module docstring. An empty/missing reason
|
|
comes back as ``None``.
|
|
"""
|
|
target = str(guild_id)
|
|
for entry in load_blacklist().get("guilds", []):
|
|
if isinstance(entry, dict):
|
|
blocked_id = entry.get("id", entry.get("guild_id"))
|
|
reason = entry.get("reason")
|
|
elif isinstance(entry, (list, tuple)):
|
|
blocked_id = entry[0] if entry else None
|
|
reason = entry[2] if len(entry) > 2 else None
|
|
else:
|
|
blocked_id, reason = entry, None
|
|
if str(blocked_id) == target:
|
|
return True, (reason or None)
|
|
return False, None
|
|
|
|
|
|
def check_user_blacklist(discord_user_id: int | str) -> tuple[bool, Optional[str]]:
|
|
"""Return (is_blacklisted, reason) for a Discord user.
|
|
|
|
Entries may be a plain id, ``[id, comment, reason]``, or a dict with
|
|
``id``/``discord_id`` plus optional ``reason``. The comment/name is ignored
|
|
by runtime checks and exists only to make BLACKLIST.json readable.
|
|
"""
|
|
target = str(discord_user_id)
|
|
for entry in load_blacklist().get("discord_users", []):
|
|
if isinstance(entry, dict):
|
|
blocked_id = entry.get("id", entry.get("discord_id"))
|
|
reason = entry.get("reason")
|
|
elif isinstance(entry, (list, tuple)):
|
|
blocked_id = entry[0]
|
|
reason = entry[2] if len(entry) > 2 else None
|
|
else:
|
|
blocked_id, reason = entry, None
|
|
if str(blocked_id) == target:
|
|
return True, reason
|
|
return False, None
|