415 lines
12 KiB
Python
415 lines
12 KiB
Python
"""
|
|
Module de gestion des informations serveur pour le bot Superviseur.
|
|
|
|
Ce module permet de stocker et charger des informations sur les serveurs Discord
|
|
de manière similaire au système de mémoire utilisateur.
|
|
|
|
Structure :
|
|
- Infos_serveurs/[nom_sanitisé]/[type].json
|
|
Exemple: Infos_serveurs/Omega_Kube/roles.json
|
|
Infos_serveurs/Omega_Kube/membres.json
|
|
Infos_serveurs/Omega_Kube/config.json
|
|
|
|
Usage :
|
|
from ia.infos_serveurs import save_server_info, load_server_info, list_server_files
|
|
|
|
# Sauvegarder des rôles
|
|
await save_server_info("Omega_Kube", "roles", roles_data)
|
|
|
|
# Charger des membres
|
|
membres = await load_server_info("Omega_Kube", "membres")
|
|
|
|
# Lister les fichiers d'un serveur
|
|
fichiers = list_server_files("Omega_Kube")
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
import json
|
|
import asyncio
|
|
import aiofiles
|
|
import logging
|
|
import re
|
|
from typing import Any, Dict, List, Optional
|
|
|
|
# Configuration
|
|
INFOS_DIR = os.environ.get("SUPERVISEUR_INFOS_DIR", "Infos_serveurs")
|
|
|
|
# Logger
|
|
logger = logging.getLogger('infos_serveurs')
|
|
|
|
# Métriques
|
|
metrics = {
|
|
"total_saves": 0,
|
|
"total_loads": 0,
|
|
"server_count": 0
|
|
}
|
|
|
|
|
|
def _sanitize_guild_name(name: str) -> str:
|
|
"""Sanitize guild name for use in filesystem path."""
|
|
# Remplacer les caractères invalides par _
|
|
sanitized = re.sub(r'[\/\\:*?"<>|\s]', '_', name)
|
|
return sanitized
|
|
|
|
|
|
def _get_server_dir(guild_id: str) -> str:
|
|
"""Get the directory path for a server."""
|
|
return os.path.join(INFOS_DIR, guild_id)
|
|
|
|
|
|
def _get_file_path(guild_id: str, info_type: str) -> str:
|
|
"""Get the file path for a specific info type."""
|
|
server_dir = _get_server_dir(guild_id)
|
|
os.makedirs(server_dir, exist_ok=True)
|
|
return os.path.join(server_dir, f"{info_type}.json")
|
|
|
|
|
|
def _ensure_infos_dir() -> None:
|
|
"""Ensure the root infos directory exists."""
|
|
os.makedirs(INFOS_DIR, exist_ok=True)
|
|
logger.debug(f"Répertoire infos serveur assuré: {INFOS_DIR}")
|
|
|
|
|
|
async def _atomic_write_async(path: str, data: Dict[str, Any]) -> None:
|
|
"""Atomic write using temporary file + replace."""
|
|
import tempfile
|
|
dirpath = os.path.dirname(path)
|
|
fd, tmp_path = tempfile.mkstemp(dir=dirpath)
|
|
try:
|
|
async with aiofiles.open(tmp_path, "w", encoding="utf-8") as f:
|
|
await f.write(json.dumps(data, ensure_ascii=False, indent=2))
|
|
await f.flush()
|
|
# Ensure durability at OS level
|
|
try:
|
|
with open(tmp_path, "rb") as f:
|
|
os.fsync(f.fileno())
|
|
except Exception:
|
|
pass
|
|
os.replace(tmp_path, path)
|
|
finally:
|
|
if os.path.exists(tmp_path):
|
|
try:
|
|
os.remove(tmp_path)
|
|
except Exception:
|
|
pass
|
|
|
|
|
|
async def save_server_info(
|
|
guild_id: str,
|
|
info_type: str,
|
|
data: Dict[str, Any],
|
|
persist_immediately: bool = True
|
|
) -> bool:
|
|
"""Sauvegarder des informations pour un serveur.
|
|
|
|
Args:
|
|
guild_id: Identifiant sanitarisé du serveur (ex: "Omega_Kube")
|
|
info_type: Type d'information (ex: "roles", "membres", "config")
|
|
data: Données à sauvegarder
|
|
persist_immediately: Si True, écrit immédiatement sur disque
|
|
|
|
Returns:
|
|
True si succès, False sinon
|
|
"""
|
|
try:
|
|
_ensure_infos_dir()
|
|
file_path = _get_file_path(guild_id, info_type)
|
|
|
|
if persist_immediately:
|
|
await _atomic_write_async(file_path, data)
|
|
else:
|
|
# Écriture différée
|
|
async with aiofiles.open(file_path, "w", encoding="utf-8") as f:
|
|
await f.write(json.dumps(data, ensure_ascii=False, indent=2))
|
|
|
|
metrics["total_saves"] += 1
|
|
logger.debug(f"Infos sauvegardées: {guild_id}/{info_type}.json")
|
|
return True
|
|
|
|
except Exception as e:
|
|
logger.error(f"Erreur lors de la sauvegarde {guild_id}/{info_type}: {e}")
|
|
return False
|
|
|
|
|
|
async def load_server_info(
|
|
guild_id: str,
|
|
info_type: str
|
|
) -> Optional[Dict[str, Any]]:
|
|
"""Charger des informations pour un serveur.
|
|
|
|
Args:
|
|
guild_id: Identifiant sanitarisé du serveur
|
|
info_type: Type d'information à charger
|
|
|
|
Returns:
|
|
Les données chargées ou None si erreur/inexistant
|
|
"""
|
|
try:
|
|
file_path = _get_file_path(guild_id, info_type)
|
|
|
|
if not os.path.exists(file_path):
|
|
logger.debug(f"Fichier inexistant: {file_path}")
|
|
return None
|
|
|
|
async with aiofiles.open(file_path, "r", encoding="utf-8") as f:
|
|
content = await f.read()
|
|
data = json.loads(content)
|
|
|
|
metrics["total_loads"] += 1
|
|
logger.debug(f"Infos chargées: {guild_id}/{info_type}.json")
|
|
return data
|
|
|
|
except json.JSONDecodeError as e:
|
|
logger.error(f"Erreur de parsing JSON pour {guild_id}/{info_type}: {e}")
|
|
return None
|
|
except Exception as e:
|
|
logger.error(f"Erreur lors du chargement {guild_id}/{info_type}: {e}")
|
|
return None
|
|
|
|
|
|
def list_server_files(guild_id: str) -> List[str]:
|
|
"""Lister les fichiers disponibles pour un serveur.
|
|
|
|
Args:
|
|
guild_id: Identifiant sanitarisé du serveur
|
|
|
|
Returns:
|
|
Liste des noms de fichiers (sans extension .json)
|
|
"""
|
|
try:
|
|
server_dir = _get_server_dir(guild_id)
|
|
if not os.path.exists(server_dir):
|
|
return []
|
|
|
|
files = []
|
|
for f in os.listdir(server_dir):
|
|
if f.endswith(".json"):
|
|
# Enlever l'extension .json
|
|
files.append(f[:-5])
|
|
|
|
return files
|
|
|
|
except Exception as e:
|
|
logger.error(f"Erreur lors du listage des fichiers pour {guild_id}: {e}")
|
|
return []
|
|
|
|
|
|
def list_all_servers() -> List[str]:
|
|
"""Lister tous les serveurs ayant des informations stockées.
|
|
|
|
Returns:
|
|
Liste des identifiants de serveurs
|
|
"""
|
|
try:
|
|
_ensure_infos_dir()
|
|
|
|
servers = []
|
|
for item in os.listdir(INFOS_DIR):
|
|
item_path = os.path.join(INFOS_DIR, item)
|
|
if os.path.isdir(item_path):
|
|
servers.append(item)
|
|
|
|
metrics["server_count"] = len(servers)
|
|
return servers
|
|
|
|
except Exception as e:
|
|
logger.error(f"Erreur lors du listage des serveurs: {e}")
|
|
return []
|
|
|
|
|
|
async def delete_server_info(guild_id: str, info_type: str) -> bool:
|
|
"""Supprimer un fichier d'information pour un serveur.
|
|
|
|
Args:
|
|
guild_id: Identifiant sanitarisé du serveur
|
|
info_type: Type d'information à supprimer
|
|
|
|
Returns:
|
|
True si succès, False sinon
|
|
"""
|
|
try:
|
|
file_path = _get_file_path(guild_id, info_type)
|
|
|
|
if os.path.exists(file_path):
|
|
os.remove(file_path)
|
|
logger.debug(f"Infos supprimées: {guild_id}/{info_type}.json")
|
|
return True
|
|
|
|
return False
|
|
|
|
except Exception as e:
|
|
logger.error(f"Erreur lors de la suppression {guild_id}/{info_type}: {e}")
|
|
return False
|
|
|
|
|
|
async def delete_all_server_info(guild_id: str) -> int:
|
|
"""Supprimer toutes les informations d'un serveur.
|
|
|
|
Args:
|
|
guild_id: Identifiant sanitarisé du serveur
|
|
|
|
Returns:
|
|
Nombre de fichiers supprimés
|
|
"""
|
|
try:
|
|
server_dir = _get_server_dir(guild_id)
|
|
|
|
if not os.path.exists(server_dir):
|
|
return 0
|
|
|
|
count = 0
|
|
for f in os.listdir(server_dir):
|
|
if f.endswith(".json"):
|
|
file_path = os.path.join(server_dir, f)
|
|
os.remove(file_path)
|
|
count += 1
|
|
|
|
logger.debug(f"Toutes les infos supprimées pour {guild_id}: {count} fichiers")
|
|
return count
|
|
|
|
except Exception as e:
|
|
logger.error(f"Erreur lors de la suppression totale pour {guild_id}: {e}")
|
|
return 0
|
|
|
|
|
|
def get_metrics() -> Dict[str, Any]:
|
|
"""Obtenir les métriques du module."""
|
|
return metrics.copy()
|
|
|
|
|
|
# Fonctions de commodité pour les cas d'usage courants
|
|
|
|
async def save_roles(guild_id: str, roles: List[Dict[str, Any]]) -> bool:
|
|
"""Sauvegarder la liste des rôles d'un serveur."""
|
|
data = {
|
|
"timestamp": str(__import__('datetime').datetime.now(__import__('datetime').timezone.utc).isoformat()),
|
|
"roles": roles
|
|
}
|
|
return await save_server_info(guild_id, "roles", data)
|
|
|
|
|
|
async def save_members(guild_id: str, members: List[Dict[str, Any]]) -> bool:
|
|
"""Sauvegarder la liste des membres d'un serveur."""
|
|
data = {
|
|
"timestamp": str(__import__('datetime').datetime.now(__import__('datetime').timezone.utc).isoformat()),
|
|
"members": members
|
|
}
|
|
return await save_server_info(guild_id, "membres", data)
|
|
|
|
|
|
async def save_config(guild_id: str, config: Dict[str, Any]) -> bool:
|
|
"""Sauvegarder la configuration d'un serveur."""
|
|
data = {
|
|
"timestamp": str(__import__('datetime').datetime.now(__import__('datetime').timezone.utc).isoformat()),
|
|
"config": config
|
|
}
|
|
return await save_server_info(guild_id, "config", data)
|
|
|
|
|
|
async def load_roles(guild_id: str) -> Optional[List[Dict[str, Any]]]:
|
|
"""Charger les rôles d'un serveur."""
|
|
data = await load_server_info(guild_id, "roles")
|
|
return data.get("roles") if data else None
|
|
|
|
|
|
async def load_members(guild_id: str) -> Optional[List[Dict[str, Any]]]:
|
|
"""Charger les membres d'un serveur."""
|
|
data = await load_server_info(guild_id, "membres")
|
|
return data.get("members") if data else None
|
|
|
|
|
|
async def load_config(guild_id: str) -> Optional[Dict[str, Any]]:
|
|
"""Charger la configuration d'un serveur."""
|
|
data = await load_server_info(guild_id, "config")
|
|
return data.get("config") if data else None
|
|
|
|
|
|
# Tests unitaires
|
|
def run_tests():
|
|
"""Exécuter des tests basiques."""
|
|
import tempfile
|
|
import shutil
|
|
import asyncio
|
|
|
|
test_dir = tempfile.mkdtemp()
|
|
global INFOS_DIR
|
|
original_dir = INFOS_DIR
|
|
INFOS_DIR = test_dir
|
|
|
|
try:
|
|
guild_id = "Test_Serveur"
|
|
|
|
# Test save/load
|
|
data = {"test": "data", "numbers": [1, 2, 3]}
|
|
result = asyncio.run(save_server_info(guild_id, "test_type", data))
|
|
assert result, "Save failed"
|
|
|
|
loaded = asyncio.run(load_server_info(guild_id, "test_type"))
|
|
assert loaded == data, "Load failed"
|
|
|
|
# Test list files
|
|
files = list_server_files(guild_id)
|
|
assert "test_type" in files, "List files failed"
|
|
|
|
# Test save_roles
|
|
roles = [{"name": "Admin", "id": 123}, {"name": "Member", "id": 456}]
|
|
result = asyncio.run(save_roles(guild_id, roles))
|
|
assert result, "Save roles failed"
|
|
|
|
loaded_roles = asyncio.run(load_roles(guild_id))
|
|
assert loaded_roles == roles, "Load roles failed"
|
|
|
|
# Test delete
|
|
result = asyncio.run(delete_server_info(guild_id, "test_type"))
|
|
assert result, "Delete failed"
|
|
|
|
loaded = asyncio.run(load_server_info(guild_id, "test_type"))
|
|
assert loaded is None, "Delete verification failed"
|
|
|
|
print("Tous les tests d'Infos_serveurs passent !")
|
|
|
|
finally:
|
|
shutil.rmtree(test_dir)
|
|
INFOS_DIR = original_dir
|
|
|
|
|
|
if __name__ == "__main__":
|
|
import asyncio
|
|
|
|
async def demo():
|
|
guild_id = "Omega_Kube"
|
|
|
|
# Exemple de sauvegarde de rôles
|
|
roles = [
|
|
{"name": "Admin", "id": 123456789, "permissions": "all"},
|
|
{"name": "Modérateur", "id": 987654321, "permissions": "kick,ban,mute"}
|
|
]
|
|
|
|
await save_roles(guild_id, roles)
|
|
print(f"Rôles sauvegardés pour {guild_id}")
|
|
|
|
# Exemple de sauvegarde de membres
|
|
members = [
|
|
{"name": "Utilisateur1", "id": 111111111, "roles": ["Member"]},
|
|
{"name": "Utilisateur2", "id": 222222222, "roles": ["Admin", "Member"]}
|
|
]
|
|
|
|
await save_members(guild_id, members)
|
|
print(f"Membres sauvegardés pour {guild_id}")
|
|
|
|
# Lister les fichiers
|
|
fichiers = list_server_files(guild_id)
|
|
print(f"Fichiers disponibles: {fichiers}")
|
|
|
|
# Charger les rôles
|
|
roles_charges = await load_roles(guild_id)
|
|
print(f"Rôles chargés: {roles_charges}")
|
|
|
|
# Métriques
|
|
print(f"Métriques: {get_metrics()}")
|
|
|
|
asyncio.run(demo())
|
|
run_tests()
|
|
|