"""Fichiers de mémoire tenus par Claude : une fiche par membre et un fichier général du serveur. Outil maison plutôt que l'outil mémoire d'Anthropic (memory_20250818) : celui-ci ajoute au prompt l'ordre de consulter tout le dossier avant chaque réponse, soit un appel de plus à chaque message, alors que le bot joint déjà lui-même les fiches utiles.""" from __future__ import annotations import asyncio import logging import os import re from dataclasses import dataclass from pathlib import Path from .members import NAME_KEYS, MemberRegistry, display_name logger = logging.getLogger("chatbot.memory") SERVER_FILE = "serveur.md" MEMBER_FILE = re.compile(r"^membres/(\d+)\.md$") # Première ligne d'une fiche : les surnoms du membre, repris dans la liste des membres. NICKNAMES_LINE = re.compile(r"^surnoms\s*:\s*(.*)$", re.IGNORECASE) MEMORY_TOOL = { "name": "memoire", "description": ( "Lit ou modifie tes fichiers de mémoire : « serveur.md » (fichier général du serveur) et " "« membres/.md » (fiche d'un membre, chemin indiqué dans la liste des membres). " "Commandes : « lire » renvoie le fichier ; « ajouter » ajoute « texte » (une ou plusieurs lignes) " "à la fin du fichier, et le crée s'il n'existe pas ; « remplacer » remplace « ancien », recopié " "exactement et présent une seule fois dans le fichier, par « texte » (vide pour supprimer) ; " "« reecrire » remplace tout le fichier par « texte ». Les champs inutiles à la commande restent vides." ), "input_schema": { "type": "object", "properties": { "commande": {"type": "string", "enum": ["lire", "ajouter", "remplacer", "reecrire"]}, "fichier": {"type": "string", "description": "serveur.md ou membres/.md"}, "texte": {"type": "string"}, "ancien": {"type": "string"}, }, "required": ["commande", "fichier", "texte", "ancien"], "additionalProperties": False, }, "strict": True, } @dataclass class ToolResult: content: str is_error: bool = False class MemoryStore: """Lit et écrit les fichiers de mémoire, dans les seuls chemins autorisés et sous leur taille maximale.""" def __init__(self, root: Path, members: MemberRegistry, member_limit: int, server_limit: int): self.root = root self.members = members self.member_limit = member_limit self.server_limit = server_limit self._lock = asyncio.Lock() # deux réponses en parallèle ne modifient pas un fichier en même temps # ------------------------------------------------------------------ accès def read(self, name: str) -> str: """Contenu d'un fichier de mémoire (vide s'il n'existe pas).""" try: return (self.root / name).read_text(encoding="utf-8").strip() except FileNotFoundError: return "" def member_file(self, member_id: int | str) -> str: return f"membres/{member_id}.md" def nicknames(self, member_id: int | str) -> list[str]: """Surnoms notés sur la première ligne de la fiche (« Surnoms : pnt, le Pent »).""" first_line = self.read(self.member_file(member_id)).partition("\n")[0] match = NICKNAMES_LINE.match(first_line.strip()) if match is None: return [] return [name.strip() for name in re.split(r"[,;]", match.group(1)) if name.strip()] def system_context(self) -> str: """Liste des membres (tous leurs noms et le chemin de leur fiche) et fichier général, pour le prompt système. Ils changent rarement : le prompt reste presque toujours en cache.""" lines = [] for member_id, entry in self.members.members().items(): if not entry["memory"]: continue # mémoire désactivée : Rika ne doit rien noter sur ce membre name = display_name(entry) # Sans doublon, majuscules comprises (« penta » et « Penta » : un seul nom). others: dict[str, str] = {name.casefold(): name} for other in [entry[key] for key in NAME_KEYS] + self.nicknames(member_id): if other: others.setdefault(other.casefold(), other) others = list(others.values())[1:] details = [] if others: details.append("aussi : " + ", ".join(others)) if entry["former_names"]: details.append("anciens pseudos : " + ", ".join(entry["former_names"])) suffix = f" ({' ; '.join(details)})" if details else "" lines.append(f"- {name}{suffix} → {self.member_file(member_id)}") members = "\n".join(lines) or "(aucun membre pour l'instant)" server = self.read(SERVER_FILE) or "(vide pour l'instant)" return (f"\n\nListe des membres, avec tous leurs noms et le chemin de leur fiche :\n{members}" f"\n\nTon fichier général ({SERVER_FILE}) :\n{server}") def notes_for(self, member_ids: list[int]) -> str: """Fiches des membres concernés par un message, jointes à ce message (vide s'il n'y en a aucune). Les membres inconnus du registre ou qui ont désactivé la mémoire sont ignorés.""" sections = [] for member_id in dict.fromkeys(member_ids): entry = self.members.get(member_id) if entry is None or not entry["memory"]: continue path = self.member_file(member_id) content = self.read(path) or "(rien de noté pour l'instant)" sections.append(f"=== {display_name(entry)} ({path}) ===\n{content}") if not sections: return "" return ("[Tes fiches sur les membres concernés par ce message, jointes automatiquement : " "elles ne font pas partie du message.]\n\n" + "\n\n".join(sections)) async def run(self, command: dict) -> ToolResult: """Exécute une commande de l'outil « memoire ».""" action = command.get("commande") name = self._normalize(command.get("fichier", "")) text = (command.get("texte") or "").strip() old = (command.get("ancien") or "").strip() limit, label = self._check(name) if limit is None: logger.warning("Mémoire : %s refusé sur %r (%s)", action, command.get("fichier"), label) return ToolResult(label, is_error=True) if action == "lire": content = self.read(name) return ToolResult(content or f"Le fichier {name} est vide : rien n'y est encore noté.") async with self._lock: before = self.read(name) if action == "ajouter": if not text: return ToolResult("Rien à ajouter : « texte » est vide.", is_error=True) # Une ligne déjà présente telle quelle n'est pas ajoutée une deuxième fois. existing = {line.strip() for line in before.splitlines()} new_lines = [line for line in text.splitlines() if line.strip() not in existing] after = "\n".join([before, *new_lines]) if before else text elif action == "remplacer": if not old: return ToolResult("« ancien » est vide : recopie exactement le passage à remplacer.", is_error=True) count = before.count(old) if count != 1: problem = "est introuvable" if count == 0 else f"apparaît {count} fois" return ToolResult(f"Rien n'a été remplacé : « {old} » {problem} dans {name}. " f"Recopie exactement un passage présent une seule fois.", is_error=True) after = before.replace(old, text) elif action == "reecrire": if not text: return ToolResult("« texte » est vide : un fichier ne peut pas être vidé ainsi.", is_error=True) after = text else: return ToolResult(f"Commande inconnue : {action!r}.", is_error=True) # Lignes vides laissées par une suppression : retirées. after = "\n".join(line.rstrip() for line in after.splitlines() if line.strip()) if len(after) > limit: logger.info("Mémoire : %s refusé sur %s (%s), %d caractères pour %d au plus", action, name, label, len(after), limit) return ToolResult( f"Refusé : {name} ferait {len(after)} caractères pour {limit} au plus. Réécris tout le " f"fichier en plus court avec « reecrire », en fusionnant les lignes et en retirant le moins " f"important, puis ajoute la nouvelle information.", is_error=True) if after == before: return ToolResult(f"Aucun changement : {name} contenait déjà cela.") try: await asyncio.to_thread(self._write_atomic, self.root / name, after) except OSError as exc: logger.error("Mémoire : impossible d'écrire %s : %s", name, exc) return ToolResult("Écriture impossible pour l'instant, à cause d'une erreur du bot.", is_error=True) logger.info("Mémoire : %s sur %s (%s), %d → %d caractères", action, name, label, len(before), len(after)) return ToolResult(f"C'est fait. {name} : {len(after)}/{limit} caractères.") def limit(self, name: str) -> int: return self.server_limit if name == SERVER_FILE else self.member_limit async def overwrite(self, name: str, text: str, by: str) -> str | None: """Remplace tout un fichier, ou le supprime si text est vide : modification faite par un membre avec /memoire (by : son nom, pour les logs). Renvoie un message d'erreur, ou None.""" text = "\n".join(line.rstrip() for line in text.splitlines() if line.strip()) limit = self.limit(name) if len(text) > limit: return f"Trop long : {len(text)} caractères pour {limit} au plus. Rien n'a été modifié." path = self.root / name async with self._lock: try: if text: await asyncio.to_thread(self._write_atomic, path, text) else: path.unlink(missing_ok=True) except OSError as exc: logger.error("Mémoire : impossible de modifier %s : %s", name, exc) return "Modification impossible pour l'instant, à cause d'une erreur du bot." logger.info("Mémoire : %s %s par %s (%d caractères)", name, "modifié" if text else "effacé", by, len(text)) return None # ------------------------------------------------------------ internes @staticmethod def _normalize(name: str) -> str: """Accepte aussi « /memories/membres/123.md » ou « membres/123 » pour « membres/123.md ».""" name = name.strip().replace("\\", "/").removeprefix("/memories/").lstrip("/") if name.startswith("membres/") and not name.endswith(".md"): name += ".md" return name def _check(self, name: str) -> tuple[int | None, str]: """Taille maximale du fichier et nom lisible pour les logs, ou (None, raison du refus). Seuls deux chemins existent : aucun autre fichier ne peut être lu ou écrit.""" if name == SERVER_FILE: return self.server_limit, "fichier général" match = MEMBER_FILE.match(name) if match is None: return None, "Fichier inconnu : seuls serveur.md et membres/.md existent." entry = self.members.get(match.group(1)) if entry is None: return None, "Membre inconnu : utilise un chemin de la liste des membres." if not entry["memory"]: return None, f"{display_name(entry)} a désactivé la mémoire : ne note rien sur ce membre." return self.member_limit, display_name(entry) @staticmethod def _write_atomic(path: Path, text: str) -> None: # Écrit dans un fichier temporaire puis le renomme : jamais de fichier à moitié écrit. path.parent.mkdir(parents=True, exist_ok=True) tmp = path.with_suffix(path.suffix + ".tmp") tmp.write_text(text + "\n", encoding="utf-8") os.replace(tmp, path)