"""TSSBOT per-guild autolog preferences + guild team context (TEAMS.json). Mirrors SREBOT's PREFERENCES model, adapted for TSS. Per-guild file: ``STORAGE_VOL_PATH/PREFERENCES/-preferences.json`` — **shared with SREBOT** (both bots write to the same file; entries are told apart by ``Type``). A flat dict keyed by the watched ``team_id``:: { "": { "Type": "tss-team", "Name": "STaYA", "Logs": "<#channelid>" } } The ``Type`` value's ``tss``/``sre`` prefix marks the data source. (The store is generic — a future ``tss-player``/``tss-tournament`` kind could be added — but only ``tss-team`` is written today.) Channel values use the SREBOT encoding ``"<#ID>"`` (enabled) / ``"<#DISABLED-ID>"`` (disabled). TEAMS.json (set by ``/set-team``) records each guild's own team, mirroring SREBOT's SQUADRONS.json:: { "": { "TM_Name": "STaYA", "team_id": 12 } } """ from __future__ import annotations import json import logging import os import re from pathlib import Path from typing import Any, Iterator, Optional from .storage import STORAGE_DIR log = logging.getLogger("tssbot.preferences") PREFERENCES_DIR: Path = STORAGE_DIR / "PREFERENCES" TEAMS_JSON_PATH: Path = STORAGE_DIR / "TEAMS.json" FEATURES_DIR: Path = STORAGE_DIR / "FEATURES" _CHANNEL_ID_RE = re.compile(r"(\d{17,20})") # --------------------------------------------------------------------------- # Low-level JSON IO # --------------------------------------------------------------------------- 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(path: Path, data: Any) -> bool: try: path.parent.mkdir(parents=True, exist_ok=True) tmp = path.with_suffix(path.suffix + ".tmp") 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 # --------------------------------------------------------------------------- # Channel encoding # --------------------------------------------------------------------------- def channel_mention(channel_id: int | str) -> str: """Return the enabled channel-route encoding for a channel id.""" return f"<#{channel_id}>" def parse_channel(value: Any) -> tuple[Optional[int], bool]: """Return (channel_id, enabled) from a stored route value. Handles ``"<#ID>"``, ``"<#DISABLED-ID>"``, and bare digit strings. Returns ``(None, False)`` for anything unparseable. """ if value is None: return None, False text = str(value) match = _CHANNEL_ID_RE.search(text) if not match: return None, False enabled = "DISABLED" not in text.upper() return int(match.group(1)), enabled # --------------------------------------------------------------------------- # Per-guild preferences # --------------------------------------------------------------------------- _PREF_SUFFIX = "-preferences.json" def _guild_pref_path(guild_id: int | str) -> Path: return PREFERENCES_DIR / f"{guild_id}{_PREF_SUFFIX}" def load_guild_preferences(guild_id: int | str) -> dict[str, Any]: """Load one guild's preferences dict (empty dict if none).""" data = _read_json(_guild_pref_path(guild_id), {}) return data if isinstance(data, dict) else {} def save_guild_preferences(guild_id: int | str, prefs: dict[str, Any]) -> bool: """Persist one guild's preferences dict.""" return _write_json(_guild_pref_path(guild_id), prefs) def upsert_log_entry( guild_id: int | str, entity_id: int | str, type_: str, name: str, channel_id: int | str, ) -> bool: """Add/replace a ``tss-team``/``tss-player`` route for a guild.""" prefs = load_guild_preferences(guild_id) entry = prefs.setdefault(str(entity_id), {}) if not isinstance(entry, dict): entry = {} prefs[str(entity_id)] = entry entry["Type"] = type_ entry["Name"] = name entry["Logs"] = channel_mention(channel_id) return save_guild_preferences(guild_id, prefs) def remove_entry(guild_id: int | str, entity_id: int | str) -> bool: """Remove a watched entity from a guild's preferences. True if removed.""" prefs = load_guild_preferences(guild_id) if str(entity_id) in prefs: del prefs[str(entity_id)] return save_guild_preferences(guild_id, prefs) return False def iter_guild_preferences() -> Iterator[tuple[int, dict[str, Any]]]: """Yield (guild_id, prefs) for every guild with a preferences file. Reads the shared ``-preferences.json`` files; SREBOT entries in them are ignored by callers (the autolog matcher acts only on tss-* Types). """ if not PREFERENCES_DIR.is_dir(): return for path in PREFERENCES_DIR.glob(f"*{_PREF_SUFFIX}"): try: guild_id = int(path.name[: -len(_PREF_SUFFIX)]) except ValueError: continue data = _read_json(path, {}) if isinstance(data, dict) and data: yield guild_id, data # --------------------------------------------------------------------------- # TEAMS.json — each guild's own team # --------------------------------------------------------------------------- def load_teams() -> dict[str, Any]: data = _read_json(TEAMS_JSON_PATH, {}) return data if isinstance(data, dict) else {} def get_guild_team(guild_id: int | str) -> Optional[dict[str, Any]]: entry = load_teams().get(str(guild_id)) return entry if isinstance(entry, dict) else None def set_guild_team(guild_id: int | str, team_id: int, name: str) -> bool: teams = load_teams() teams[str(guild_id)] = {"TM_Name": name, "team_id": int(team_id)} return _write_json(TEAMS_JSON_PATH, teams) # --------------------------------------------------------------------------- # FEATURES.json — per-guild feature flags (shared with SREBOT; we use Language) # --------------------------------------------------------------------------- def _guild_features_path(guild_id: int | str) -> Path: return FEATURES_DIR / f"{guild_id}-features.json" def load_features(guild_id: int | str) -> dict[str, Any]: """Load a guild's feature flags (shared SREBOT/TSSBOT file).""" data = _read_json(_guild_features_path(guild_id), {}) return data if isinstance(data, dict) else {} def save_features(guild_id: int | str, features: dict[str, Any]) -> bool: """Persist a guild's feature flags.""" return _write_json(_guild_features_path(guild_id), features) def guild_language_column(guild_id: int | str) -> str: """Return the guild's LangTableReader column, e.g. ```` (the default).""" return load_features(guild_id).get("Language") or ""