392 lines
16 KiB
Python
392 lines
16 KiB
Python
"""
|
|
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 <id>")
|
|
|
|
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}
|