""" Service d'export JSON pour le site web. Génère deux fichiers : - servers.json : tous les serveurs où le bot est présent - members.json : tous les membres (sans bots) d'un serveur cible """ import json import os import logging from datetime import datetime, timezone from pathlib import Path from typing import Optional import disnake logger = logging.getLogger("kuby.json_export") # ─── Chemins ──────────────────────────────────────────────────────────────── CONFIG_FILE = Path("data/json-export-config.json") HISTORY_FILE = Path("data/json-export-history.json") def _get_output_dir() -> Path: """Lit le dossier de sortie depuis la config (ou .env en fallback).""" if CONFIG_FILE.exists(): try: saved = json.loads(CONFIG_FILE.read_text(encoding="utf-8")) if saved.get("outputDir"): return Path(saved["outputDir"]) except Exception: pass return Path(os.getenv("JSON_OUTPUT_DIR", "data/website-json")) @property def OUTPUT_DIR() -> Path: return _get_output_dir() # Propriétés dynamiques — recalculées à chaque accès def get_output_dir() -> Path: return _get_output_dir() def get_servers_file() -> Path: return _get_output_dir() / "servers.json" def get_members_file() -> Path: return _get_output_dir() / "members.json" # Aliases statiques utilisés dans les imports existants (valeur initiale) OUTPUT_DIR = _get_output_dir() SERVERS_FILE = OUTPUT_DIR / "servers.json" MEMBERS_FILE = OUTPUT_DIR / "members.json" # ─── Config par défaut ─────────────────────────────────────────────────────── DEFAULT_CONFIG: dict = { # ID du serveur ciblé pour members.json (ne change jamais) "targetGuildId": os.getenv("TARGET_GUILD_ID", ""), # Heure de génération automatique quotidienne (format 24h) "autoHour": 3, "autoMinute": 0, # Dossier de sortie (vide = utilise JSON_OUTPUT_DIR du .env) "outputDir": os.getenv("JSON_OUTPUT_DIR", "data/website-json"), # Noms de rôles reconnus comme "fondateur" (insensible à la casse) "founderRoles": ["fondateur", "founder"], # Champs inclus dans servers.json "serverFields": { "id": True, "name": True, "icon": True, # URL pp du serveur "banner": True, # URL bannière "owner": True, # Propriétaire + rôle fondateur "memberCount": True, # Membres humains uniquement "description": True, "createdAt": True, "boostLevel": False, "boostCount": False, "verificationLevel": False, }, # Champs inclus dans members.json "memberFields": { "id": True, "username": True, "displayName": True, # Pseudo sur le serveur "globalName": True, # Nom d'affichage global "avatar": True, # URL avatar (serveur + global) "bio": True, # Description (toujours null, limite API Discord) "roles": True, # Tous les rôles avec couleur "joinedAt": True, # Date d'arrivée sur le serveur "createdAt": True, # Date de création du compte "badges": False, # Badges du compte "accentColor": False, # Couleur de profil "isBot": False, # Flag bot (les bots sont toujours exclus de la liste) }, } # ─── Helpers config ────────────────────────────────────────────────────────── def load_config() -> dict: CONFIG_FILE.parent.mkdir(parents=True, exist_ok=True) if not CONFIG_FILE.exists(): save_config(DEFAULT_CONFIG) return dict(DEFAULT_CONFIG) try: saved = json.loads(CONFIG_FILE.read_text(encoding="utf-8")) # Merge avec les valeurs par défaut pour les nouvelles clés merged = {**DEFAULT_CONFIG, **saved} merged["serverFields"] = {**DEFAULT_CONFIG["serverFields"], **saved.get("serverFields", {})} merged["memberFields"] = {**DEFAULT_CONFIG["memberFields"], **saved.get("memberFields", {})} merged["founderRoles"] = saved.get("founderRoles", DEFAULT_CONFIG["founderRoles"]) merged["outputDir"] = saved.get("outputDir", DEFAULT_CONFIG["outputDir"]) return merged except Exception as e: logger.error(f"Erreur lecture config : {e}") return dict(DEFAULT_CONFIG) def save_config(config: dict) -> None: CONFIG_FILE.parent.mkdir(parents=True, exist_ok=True) CONFIG_FILE.write_text(json.dumps(config, indent=2, ensure_ascii=False), encoding="utf-8") # ─── Historique des générations ────────────────────────────────────────────── MAX_HISTORY = 20 # nombre max d'entrées conservées def _load_history() -> list: if not HISTORY_FILE.exists(): return [] try: return json.loads(HISTORY_FILE.read_text(encoding="utf-8")) except Exception: return [] def _save_history(history: list) -> None: HISTORY_FILE.parent.mkdir(parents=True, exist_ok=True) HISTORY_FILE.write_text(json.dumps(history, indent=2, ensure_ascii=False), encoding="utf-8") def log_history(result: dict) -> None: """Ajoute une entrée dans l'historique après une génération.""" history = _load_history() entry = { "date": datetime.now(timezone.utc).isoformat(), "trigger": result.get("trigger", "auto"), # "auto" ou "manual" "serversCount": result["servers"]["count"] if result.get("servers") else None, "membersCount": result["members"]["count"] if result.get("members") else None, "guildName": result["members"]["guildName"] if result.get("members") else None, "errors": result.get("errors", []), "success": len(result.get("errors", [])) == 0, } history.insert(0, entry) _save_history(history[:MAX_HISTORY]) def get_history() -> list: return _load_history() # ─── Génération servers.json ───────────────────────────────────────────────── async def generate_servers_json(bot: disnake.Client) -> dict: config = load_config() fields = config["serverFields"] founder_keys = [r.lower() for r in config.get("founderRoles", ["fondateur", "founder"])] out_dir = Path(config["outputDir"]) servers_file = out_dir / "servers.json" servers = [] for guild in bot.guilds: try: guild = await bot.fetch_guild(guild.id, with_counts=False) except Exception: pass # on travaille avec la version cache entry = {} if fields.get("id"): entry["id"] = str(guild.id) if fields.get("name"): entry["name"] = guild.name if fields.get("icon"): entry["icon"] = str(guild.icon.url) if guild.icon else None if fields.get("banner"): entry["banner"] = str(guild.banner.url) if guild.banner else None if fields.get("description"): entry["description"] = guild.description if fields.get("createdAt"): entry["createdAt"] = guild.created_at.isoformat() if fields.get("boostLevel"): entry["boostLevel"] = guild.premium_tier if fields.get("boostCount"): entry["boostCount"] = guild.premium_subscription_count if fields.get("verificationLevel"): entry["verificationLevel"] = str(guild.verification_level) if fields.get("memberCount"): # Use cache if available (intents.members is True), otherwise fallback to count if guild.members: entry["memberCount"] = sum(1 for m in guild.members if not m.bot) else: entry["memberCount"] = guild.member_count if fields.get("owner"): founders = [] try: owner = await guild.fetch_member(guild.owner_id) founders.append({ "id": str(owner.id), "username": owner.name, "displayName": owner.display_name, "avatar": str(owner.display_avatar.url), "isOwner": True, }) except Exception: pass # Membres avec un rôle dont le nom correspond à founderRoles # Use cache if possible members_to_check = guild.members if guild.members else [] if not members_to_check: try: members_to_check = [m async for m in guild.fetch_members(limit=None)] except Exception: pass for member in members_to_check: if member.bot or member.id == guild.owner_id: continue for role in member.roles: if any(key in role.name.lower() for key in founder_keys): founders.append({ "id": str(member.id), "username": member.name, "displayName": member.display_name, "avatar": str(member.display_avatar.url), "isOwner": False, }) break entry["founders"] = founders servers.append(entry) output = { "generatedAt": datetime.now(timezone.utc).isoformat(), "count": len(servers), "servers": servers, } out_dir.mkdir(parents=True, exist_ok=True) servers_file.write_text(json.dumps(output, indent=2, ensure_ascii=False), encoding="utf-8") logger.info(f"servers.json généré → {servers_file} ({len(servers)} serveurs)") return output # ─── Génération members.json ───────────────────────────────────────────────── async def generate_members_json(bot: disnake.Client, guild_id: Optional[int] = None) -> dict: config = load_config() fields = config["memberFields"] target_id = guild_id or (int(config["targetGuildId"]) if config["targetGuildId"] else None) out_dir = Path(config["outputDir"]) members_file = out_dir / "members.json" if not target_id: raise ValueError("Aucun targetGuildId configuré. Utilise /json-config set-guild ") guild = bot.get_guild(target_id) or await bot.fetch_guild(target_id) if not guild: raise ValueError(f"Serveur introuvable : {target_id}") members_list = [] # Use cached members if available, or fetch if not source = guild.members if guild.members else [] if not source: async for m in guild.fetch_members(limit=None): if not m.bot: source.append(m) for member in source: if member.bot: continue entry = {} if fields.get("id"): entry["id"] = str(member.id) if fields.get("username"): entry["username"] = member.name if fields.get("displayName"): entry["displayName"] = member.display_name if fields.get("globalName"): entry["globalName"] = member.global_name or member.name if fields.get("avatar"): entry["avatar"] = str(member.display_avatar.url) entry["globalAvatar"] = str(member.avatar.url) if member.avatar else str(member.default_avatar.url) if fields.get("joinedAt"): entry["joinedAt"] = member.joined_at.isoformat() if member.joined_at else None if fields.get("createdAt"): entry["createdAt"] = member.created_at.isoformat() if fields.get("isBot"): entry["isBot"] = member.bot if fields.get("bio"): entry["bio"] = None # Non accessible via l'API Bot officielle if fields.get("accentColor"): entry["accentColor"] = None # Nécessite fetch user OAuth2 if fields.get("roles"): entry["roles"] = [ { "id": str(r.id), "name": r.name, "color": str(r.color), "position": r.position, "hoist": r.hoist, } for r in sorted(member.roles, key=lambda r: -r.position) if r.id != guild.id # exclut @everyone ] if fields.get("badges"): entry["badges"] = [flag.name for flag in member.public_flags.all()] members_list.append(entry) output = { "generatedAt": datetime.now(timezone.utc).isoformat(), "guildId": str(guild.id), "guildName": guild.name, "count": len(members_list), "members": members_list, } out_dir.mkdir(parents=True, exist_ok=True) members_file.write_text(json.dumps(output, indent=2, ensure_ascii=False), encoding="utf-8") logger.info(f"members.json généré → {members_file} ({len(members_list)} membres)") return output # ─── Génération complète ───────────────────────────────────────────────────── async def generate_all(bot: disnake.Client, trigger: str = "auto") -> dict: results = {"servers": None, "members": None, "errors": [], "trigger": trigger} try: results["servers"] = await generate_servers_json(bot) except Exception as e: results["errors"].append(f"servers: {e}") logger.error(f"Erreur génération servers.json : {e}") try: results["members"] = await generate_members_json(bot) except Exception as e: results["errors"].append(f"members: {e}") logger.error(f"Erreur génération members.json : {e}") log_history(results) return results # ─── Prévisualisation (sans écriture fichier) ──────────────────────────────── async def preview_servers(bot: disnake.Client, limit: int = 3) -> list: """Retourne un aperçu de N serveurs sans écrire le fichier.""" config = load_config() fields = config["serverFields"] founder_keys = [r.lower() for r in config.get("founderRoles", ["fondateur"])] results = [] for guild in list(bot.guilds)[:limit]: entry = { "id": str(guild.id), "name": guild.name, "icon": str(guild.icon.url) if guild.icon else None, } if fields.get("memberCount"): try: members = [m async for m in guild.fetch_members(limit=None)] entry["memberCount"] = sum(1 for m in members if not m.bot) except Exception: entry["memberCount"] = guild.member_count results.append(entry) return results async def preview_members(bot: disnake.Client, limit: int = 5) -> dict: """Retourne un aperçu de N membres sans écrire le fichier.""" config = load_config() target_id = int(config["targetGuildId"]) if config["targetGuildId"] else None if not target_id: raise ValueError("Aucun targetGuildId configuré.") guild = bot.get_guild(target_id) or await bot.fetch_guild(target_id) preview = [] count = 0 async for member in guild.fetch_members(limit=None): if member.bot: continue preview.append({ "id": str(member.id), "username": member.name, "displayName": member.display_name, "avatar": str(member.display_avatar.url), "roles": [r.name for r in member.roles if r.id != guild.id], }) count += 1 if count >= limit: break return {"guildName": guild.name, "totalFetched": limit, "members": preview}