|
|
@@ -0,0 +1,151 @@
|
|
|
+"""Registre des membres vus par le bot (noms, anciens pseudos, réglages de la mémoire), dans un fichier JSON."""
|
|
|
+
|
|
|
+from __future__ import annotations
|
|
|
+
|
|
|
+import asyncio
|
|
|
+import json
|
|
|
+import logging
|
|
|
+import os
|
|
|
+from datetime import datetime
|
|
|
+from pathlib import Path
|
|
|
+
|
|
|
+logger = logging.getLogger("chatbot.members")
|
|
|
+
|
|
|
+FILE_VERSION = 1
|
|
|
+# Noms Discord d'un membre : nom d'utilisateur (unique), nom d'affichage, pseudo sur le serveur.
|
|
|
+NAME_KEYS = ("username", "global_name", "nickname")
|
|
|
+# Anciens pseudos gardés par membre, du plus ancien au plus récent.
|
|
|
+MAX_FORMER_NAMES = 5
|
|
|
+
|
|
|
+
|
|
|
+class MemberRegistry:
|
|
|
+ """Garde, pour chaque membre vu (auteur d'un message lu par le bot, ou membre mentionné), ses noms
|
|
|
+ actuels et ses anciens pseudos, s'il a été prévenu de la mémoire et s'il l'a désactivée.
|
|
|
+
|
|
|
+ Le fichier n'est réécrit que si quelque chose change : pas d'écriture à chaque message."""
|
|
|
+
|
|
|
+ def __init__(self, file_path: Path):
|
|
|
+ self.file_path = file_path
|
|
|
+ self._data: dict[str, dict] = self._load()
|
|
|
+ self._save_lock = asyncio.Lock()
|
|
|
+
|
|
|
+ # ------------------------------------------------------------------ accès
|
|
|
+
|
|
|
+ def get(self, member_id: int | str) -> dict | None:
|
|
|
+ return self._data.get(str(member_id))
|
|
|
+
|
|
|
+ def members(self) -> dict[str, dict]:
|
|
|
+ """Tous les membres connus, par identifiant."""
|
|
|
+ return self._data
|
|
|
+
|
|
|
+ def memory_enabled(self, member_id: int | str) -> bool:
|
|
|
+ """La mémoire est activée par défaut, y compris pour un membre encore inconnu."""
|
|
|
+ entry = self.get(member_id)
|
|
|
+ return entry is None or entry["memory"]
|
|
|
+
|
|
|
+ async def see(self, member_id: int, username: str, global_name: str | None,
|
|
|
+ nickname: str | None = None, in_guild: bool = True) -> None:
|
|
|
+ """Enregistre un membre ou met ses noms à jour ; un nom remplacé rejoint les anciens pseudos.
|
|
|
+
|
|
|
+ in_guild=False (message privé) : le pseudo sur le serveur n'est pas connu, il reste inchangé."""
|
|
|
+ names = {"username": username, "global_name": global_name or None}
|
|
|
+ if in_guild:
|
|
|
+ names["nickname"] = nickname or None
|
|
|
+ key = str(member_id)
|
|
|
+ entry = self._data.get(key)
|
|
|
+ if entry is None:
|
|
|
+ self._data[key] = {
|
|
|
+ **{name_key: None for name_key in NAME_KEYS}, **names,
|
|
|
+ "former_names": [], "notified": False, "memory": True,
|
|
|
+ "first_seen": datetime.now().strftime("%Y-%m-%d"),
|
|
|
+ }
|
|
|
+ logger.info("Nouveau membre dans le registre : %s (%s)", display_name(self._data[key]), key)
|
|
|
+ await self._save()
|
|
|
+ return
|
|
|
+ old_names = [entry[name_key] for name_key in NAME_KEYS if entry[name_key]]
|
|
|
+ changed = False
|
|
|
+ for name_key, value in names.items():
|
|
|
+ if entry[name_key] != value:
|
|
|
+ entry[name_key] = value
|
|
|
+ changed = True
|
|
|
+ if not changed:
|
|
|
+ return
|
|
|
+ current = {entry[name_key] for name_key in NAME_KEYS if entry[name_key]}
|
|
|
+ former = [name for name in entry["former_names"] if name not in current]
|
|
|
+ for name in old_names:
|
|
|
+ if name not in current and name not in former:
|
|
|
+ former.append(name)
|
|
|
+ entry["former_names"] = former[-MAX_FORMER_NAMES:]
|
|
|
+ logger.info("Noms mis à jour dans le registre : %s (%s)", display_name(entry), key)
|
|
|
+ await self._save()
|
|
|
+
|
|
|
+ async def set_notified(self, member_id: int | str) -> None:
|
|
|
+ """Le membre a été prévenu de la mémoire : il ne le sera plus."""
|
|
|
+ if (entry := self.get(member_id)) is not None and not entry["notified"]:
|
|
|
+ entry["notified"] = True
|
|
|
+ await self._save()
|
|
|
+
|
|
|
+ async def set_memory(self, member_id: int | str, enabled: bool) -> None:
|
|
|
+ """Active ou désactive la mémoire pour ce membre."""
|
|
|
+ if (entry := self.get(member_id)) is not None and entry["memory"] != enabled:
|
|
|
+ entry["memory"] = enabled
|
|
|
+ logger.info("Mémoire %s pour %s (%s)", "activée" if enabled else "désactivée", display_name(entry), member_id)
|
|
|
+ await self._save()
|
|
|
+
|
|
|
+ # ------------------------------------------------------------ internes
|
|
|
+
|
|
|
+ def _load(self) -> dict[str, dict]:
|
|
|
+ if not self.file_path.exists():
|
|
|
+ return {}
|
|
|
+ try:
|
|
|
+ raw = json.loads(self.file_path.read_text(encoding="utf-8"))
|
|
|
+ if not isinstance(raw, dict) or not isinstance(raw.get("members"), dict):
|
|
|
+ raise ValueError("« members » absent ou n'est pas un dictionnaire")
|
|
|
+ except (ValueError, OSError) as exc: # JSON invalide, mauvais encodage ou structure inattendue
|
|
|
+ self._set_aside(exc)
|
|
|
+ return {}
|
|
|
+ return {key: entry for key, value in raw["members"].items() if (entry := self._sanitize(value))}
|
|
|
+
|
|
|
+ @staticmethod
|
|
|
+ def _sanitize(value) -> dict | None:
|
|
|
+ """Une entrée du fichier, complétée si besoin, ou None si elle est illisible."""
|
|
|
+ if not isinstance(value, dict) or not isinstance(value.get("username"), str):
|
|
|
+ return None
|
|
|
+ entry = {name_key: value.get(name_key) if isinstance(value.get(name_key), str) else None for name_key in NAME_KEYS}
|
|
|
+ former = value.get("former_names")
|
|
|
+ entry["former_names"] = [n for n in former if isinstance(n, str)] if isinstance(former, list) else []
|
|
|
+ entry["notified"] = value.get("notified") is True
|
|
|
+ entry["memory"] = value.get("memory") is not False
|
|
|
+ entry["first_seen"] = value.get("first_seen") if isinstance(value.get("first_seen"), str) else None
|
|
|
+ return entry
|
|
|
+
|
|
|
+ def _set_aside(self, exc: Exception) -> None:
|
|
|
+ """Met de côté un fichier illisible, sous un nom daté, comme pour l'historique."""
|
|
|
+ stamp = datetime.now().strftime("%Y-%m-%d_%H-%M-%S")
|
|
|
+ backup = self.file_path.with_name(f"{self.file_path.name}.corrompu-{stamp}")
|
|
|
+ logger.error("Registre des membres illisible (%s). Copie de secours : %s", exc, backup)
|
|
|
+ try:
|
|
|
+ os.replace(self.file_path, backup)
|
|
|
+ except OSError as err:
|
|
|
+ logger.error("Copie de secours impossible : %s", err)
|
|
|
+
|
|
|
+ async def _save(self) -> None:
|
|
|
+ payload = {"version": FILE_VERSION, "members": self._data}
|
|
|
+ text = json.dumps(payload, ensure_ascii=False, indent=2)
|
|
|
+ async with self._save_lock:
|
|
|
+ try:
|
|
|
+ await asyncio.to_thread(self._write_atomic, text)
|
|
|
+ except OSError as exc:
|
|
|
+ logger.error("Impossible de sauvegarder le registre des membres : %s", exc)
|
|
|
+
|
|
|
+ def _write_atomic(self, text: str) -> None:
|
|
|
+ # Écrit dans un fichier temporaire puis le renomme : jamais de fichier à moitié écrit.
|
|
|
+ self.file_path.parent.mkdir(parents=True, exist_ok=True)
|
|
|
+ tmp = self.file_path.with_suffix(self.file_path.suffix + ".tmp")
|
|
|
+ tmp.write_text(text, encoding="utf-8")
|
|
|
+ os.replace(tmp, self.file_path)
|
|
|
+
|
|
|
+
|
|
|
+def display_name(entry: dict) -> str:
|
|
|
+ """Le nom affiché sur le serveur : pseudo, sinon nom d'affichage, sinon nom d'utilisateur."""
|
|
|
+ return entry["nickname"] or entry["global_name"] or entry["username"]
|
Johnounet