chatbot-maisonnette/bot/members.py

"""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"]