add role backup

This commit is contained in:
rambros 2026-02-27 21:41:35 +05:30
parent b9a89a4d2b
commit 5f9e8c9ed2
3 changed files with 186 additions and 99 deletions

View file

@ -30,11 +30,24 @@ class DiscordReader:
await self.client.login(self.token) await self.client.login(self.token)
# Use fetch methods specifically to bypass dependency on gateway cache # Use fetch methods specifically to bypass dependency on gateway cache
try:
logging.info(f"Fetching guild {self.server_id}...")
self.guild = await self.client.fetch_guild(self.server_id) self.guild = await self.client.fetch_guild(self.server_id)
logging.info(f"Successfully fetched guild: {self.guild.name}")
except discord.Forbidden:
logging.error(f"403 Forbidden: Missing Access to fetch guild {self.server_id}.")
raise
# Pre-fetch roles via API # Pre-fetch roles via API - Handle Forbidden gracefully
try:
roles = await self.guild.fetch_roles() roles = await self.guild.fetch_roles()
self.role_map = {r.id: r.name for r in roles} self.role_map = {r.id: r.name for r in roles}
except discord.Forbidden:
logging.warning("403 Forbidden: Missing Access to fetch roles. Continuing without role mapping.")
self.role_map = {}
except Exception as e:
logging.error(f"Failed to fetch roles: {e}")
self.role_map = {}
async def validate(self) -> Dict[str, Any]: async def validate(self) -> Dict[str, Any]:
"""Validates the token, server ID, intents, and permissions.""" """Validates the token, server ID, intents, and permissions."""
@ -119,6 +132,16 @@ class DiscordReader:
return [] return []
return await self.guild.fetch_stickers() return await self.guild.fetch_stickers()
async def get_members(self):
"""Returns all members in the server."""
if not self.guild:
return []
# Use a list to hold all members
members = []
async for member in self.guild.fetch_members(limit=None):
members.append(member)
return members
async def get_channels(self, category_id: int | None = None): async def get_channels(self, category_id: int | None = None):
"""Yields all non-category channels.""" """Yields all non-category channels."""
if not self.guild: if not self.guild:

View file

@ -33,18 +33,34 @@ class DiscordExporter:
self.media_path.mkdir(exist_ok=True) self.media_path.mkdir(exist_ok=True)
logger.info(f"Export directory set to: {self.export_path}") logger.info(f"Export directory set to: {self.export_path}")
logger.info(f"Targeting server: {self.server_name} ({self.server_id})")
return metadata return metadata
async def export_metadata(self): async def export_metadata(self):
"""Saves server metadata to a JSON file.""" """Saves server metadata to a JSON file."""
metadata = await self.reader.get_server_metadata() metadata = await self.reader.get_server_metadata()
output_file = self.export_path / "server_metadata.json"
# Add relative paths to local assets
if self.reader.guild:
if self.reader.guild.icon:
ext = "gif" if self.reader.guild.icon.is_animated() else "png"
metadata["icon"] = f"server_media/server_icon.{ext}"
else:
metadata["icon"] = None
if self.reader.guild.banner:
ext = "gif" if self.reader.guild.banner.is_animated() else "png"
metadata["banner"] = f"server_media/server_banner.{ext}"
else:
metadata["banner"] = None
output_file = self.export_path / "server_profile.json"
with open(output_file, "w", encoding="utf-8") as f: with open(output_file, "w", encoding="utf-8") as f:
json.dump(metadata, f, indent=4, ensure_ascii=False) json.dump(metadata, f, indent=4, ensure_ascii=False)
return metadata return metadata
async def export_roles(self): async def export_roles(self):
"""Exports all roles to roles.json.""" """Exports all roles to server_roles.json."""
roles = await self.reader.get_roles() roles = await self.reader.get_roles()
role_data = [] role_data = []
for r in roles: for r in roles:
@ -58,43 +74,62 @@ class DiscordExporter:
"mentionable": r.mentionable "mentionable": r.mentionable
}) })
output_file = self.export_path / "roles.json" output_file = self.export_path / "server_roles.json"
with open(output_file, "w", encoding="utf-8") as f: with open(output_file, "w", encoding="utf-8") as f:
json.dump(role_data, f, indent=4, ensure_ascii=False) json.dump(role_data, f, indent=4, ensure_ascii=False)
return role_data return role_data
async def export_emojis_stickers(self): async def download_server_assets(self):
"""Exports all emojis and stickers, plus server icon/banner, to server_media folder.""" """Downloads server icon and banner to media folder."""
emojis = await self.reader.get_emojis()
stickers = await self.reader.get_stickers()
metadata = await self.reader.get_server_metadata() metadata = await self.reader.get_server_metadata()
# Download Server Icon # Download Server Icon
if metadata.get("icon_url"): if metadata.get("icon_url"):
try: try:
# We need a way to download from URL directly or use reader
# Since DiscordReader doesn't have a direct "download from URL" we might need one
# but for guild icon/banner we can use guild object if available in reader.
if self.reader.guild and self.reader.guild.icon: if self.reader.guild and self.reader.guild.icon:
logger.info(f"Downloading server icon: {self.reader.guild.icon.url}")
data = await self.reader.download_asset(self.reader.guild.icon) data = await self.reader.download_asset(self.reader.guild.icon)
ext = "gif" if self.reader.guild.icon.is_animated() else "png" ext = "gif" if self.reader.guild.icon.is_animated() else "png"
with open(self.media_path / f"server_icon.{ext}", "wb") as f: icon_path = self.media_path / f"server_icon.{ext}"
with open(icon_path, "wb") as f:
f.write(data) f.write(data)
logger.info(f"Saved server icon to {icon_path}")
else:
logger.warning("Icon URL found in metadata but guild icon asset is missing.")
except discord.Forbidden:
logger.error("403 Forbidden: Missing Access to download server icon.")
except Exception as e: except Exception as e:
logger.error(f"Failed to download server icon: {e}") logger.error(f"Failed to download server icon: {e}")
else:
logger.info("No server icon found to download.")
# Download Server Banner # Download Server Banner
if metadata.get("banner_url"): if metadata.get("banner_url"):
try: try:
if self.reader.guild and self.reader.guild.banner: if self.reader.guild and self.reader.guild.banner:
logger.info(f"Downloading server banner: {self.reader.guild.banner.url}")
data = await self.reader.download_asset(self.reader.guild.banner) data = await self.reader.download_asset(self.reader.guild.banner)
ext = "gif" if self.reader.guild.banner.is_animated() else "png" ext = "gif" if self.reader.guild.banner.is_animated() else "png"
with open(self.media_path / f"server_banner.{ext}", "wb") as f: banner_path = self.media_path / f"server_banner.{ext}"
with open(banner_path, "wb") as f:
f.write(data) f.write(data)
logger.info(f"Saved server banner to {banner_path}")
except discord.Forbidden:
logger.error("403 Forbidden: Missing Access to download server banner.")
except Exception as e: except Exception as e:
logger.error(f"Failed to download server banner: {e}") logger.error(f"Failed to download server banner: {e}")
else:
logger.info("No server banner found to download.")
async def export_customization(self):
"""Exports all emojis, stickers, members, plus server icon/banner, to server_media folder."""
# Ensure icon/banner are downloaded
await self.download_server_assets()
emojis = await self.reader.get_emojis()
stickers = await self.reader.get_stickers()
emoji_data = [] emoji_data = []
logger.info(f"Exporting {len(emojis)} emojis...")
for e in emojis: for e in emojis:
ext = "gif" if e.animated else "png" ext = "gif" if e.animated else "png"
filename = f"emoji_{e.name}_{e.id}.{ext}" filename = f"emoji_{e.name}_{e.id}.{ext}"
@ -109,10 +144,13 @@ class DiscordExporter:
"animated": e.animated, "animated": e.animated,
"filename": filename "filename": filename
}) })
except discord.Forbidden:
logger.error(f"403 Forbidden: Missing Access to download emoji {e.name}")
except Exception as ex: except Exception as ex:
logger.error(f"Failed to download emoji {e.name}: {ex}") logger.error(f"Failed to download emoji {e.name}: {ex}")
sticker_data = [] sticker_data = []
logger.info(f"Exporting {len(stickers)} stickers...")
for s in stickers: for s in stickers:
ext = "png" ext = "png"
if s.url: if s.url:
@ -131,15 +169,41 @@ class DiscordExporter:
"name": s.name, "name": s.name,
"filename": filename "filename": filename
}) })
except discord.Forbidden:
logger.error(f"403 Forbidden: Missing Access to download sticker {s.name}")
except Exception as ex: except Exception as ex:
logger.error(f"Failed to download sticker {s.name}: {ex}") logger.error(f"Failed to download sticker {s.name}: {ex}")
with open(self.export_path / "emojis.json", "w", encoding="utf-8") as f: member_data = []
json.dump(emoji_data, f, indent=4, ensure_ascii=False) logger.info("Attempting to export members (requires Server Members Intent)...")
with open(self.export_path / "stickers.json", "w", encoding="utf-8") as f: try:
json.dump(sticker_data, f, indent=4, ensure_ascii=False) members = await self.reader.get_members()
for m in members:
member_data.append({
"id": str(m.id),
"name": m.name,
"display_name": m.display_name,
"discriminator": m.discriminator,
"avatar_url": str(m.avatar.url) if m.avatar else None,
"bot": m.bot,
"roles": [str(r.id) for r in m.roles if not r.is_default()]
})
logger.info(f"Successfully exported {len(member_data)} members.")
except discord.Forbidden:
logger.warning("403 Forbidden: Missing Access to fetch members. Skipping members export. Ensure 'Server Members Intent' is enabled in Discord Dev Portal.")
except Exception as e:
logger.error(f"Failed to fetch members: {e}")
return len(emoji_data), len(sticker_data) customization = {
"emojis": emoji_data,
"stickers": sticker_data,
"members": member_data
}
with open(self.export_path / "customization.json", "w", encoding="utf-8") as f:
json.dump(customization, f, indent=4, ensure_ascii=False)
return len(emoji_data), len(sticker_data), len(member_data)
async def export_channels_structure(self): async def export_channels_structure(self):
"""Exports categories and channels hierarchy.""" """Exports categories and channels hierarchy."""

View file

@ -1,5 +1,6 @@
import sys import sys
import asyncio import asyncio
import discord
import logging import logging
import time import time
from rich.console import Console from rich.console import Console
@ -77,94 +78,90 @@ class ExodusCLI:
console.print(f"[bold cyan]Source Server:[/bold cyan] {d_display}") console.print(f"[bold cyan]Source Server:[/bold cyan] {d_display}")
console.print("\n[bold]Main Menu[/bold]") console.print("\n[bold]Main Menu[/bold]")
console.print("(1) Setup Export Folder & Metadata") console.print("(1) Backup Basic Server Profile")
console.print("(2) Export Roles") console.print("(2) Backup User Names & Avatars, other customization")
console.print("(3) Export Emojis & Stickers") console.print("(3) Backup Message")
console.print("(4) Export Channels Structure") console.print("(4) Update & Sync Backup")
console.print("(5) Export All Message History (Long operation)")
console.print("(C) Configuration") console.print("(C) Configuration")
console.print("(Q) Exit") console.print("(Q) Exit")
choice = Prompt.ask("\nSelect an option", choices=["1", "2", "3", "4", "5", "C", "Q"], default="Q", show_choices=False).upper() choice = Prompt.ask("\nSelect an option", choices=["1", "2", "3", "4", "C", "Q"], default="Q", show_choices=False).upper()
if choice == "1": if choice == "1":
await self.setup_export() await self.backup_server_profile()
elif choice == "2": elif choice == "2":
await self.export_roles() await self.backup_customization()
elif choice == "3": elif choice == "3":
await self.export_assets() await self.backup_messages()
elif choice == "4": elif choice == "4":
await self.export_structure() await self.sync_backup()
elif choice == "5":
await self.export_messages()
elif choice == "C": elif choice == "C":
await self.edit_configuration() await self.edit_configuration()
elif choice == "Q": elif choice == "Q":
await self.engine.close_connections() await self.engine.close_connections()
break break
async def setup_export(self): async def backup_server_profile(self):
"""Option 1: Backup name, banner, logo and roles (with permissions)"""
if not self.tokens_valid: return if not self.tokens_valid: return
try: try:
await self.engine.discord_reader.start() await self.engine.discord_reader.start()
with console.status("[yellow]Setting up export directory...[/yellow]"): # setup folder
meta = await self.exporter.setup() meta = await self.exporter.setup()
# Export metadata & assets
with console.status("[yellow]Backing up server profile & assets...[/yellow]"):
await self.exporter.export_metadata() await self.exporter.export_metadata()
console.print(f"[bold green]Export directory ready: {self.exporter.export_path}[/bold green]") await self.exporter.download_server_assets()
console.print(f"Server: [bold]{meta['name']}[/bold] ({meta['id']})")
# export roles separately to handle 403 gracefully
try:
with console.status("[yellow]Backing up roles...[/yellow]"):
await self.exporter.export_roles()
except discord.Forbidden:
console.print("[yellow]Warning: 403 Forbidden. Could not backup roles (Missing Access).[/yellow]")
console.print(f"[bold green]Basic Server Profile backed up to: {self.exporter.export_path}[/bold green]")
except discord.Forbidden as e:
console.print(f"[bold red]Backup failed: {e}[/bold red]")
console.print("[yellow]Hint: This is a 'Missing Access' error. Check if your bot has 'Guilds' permissions enabled.[/yellow]")
except Exception as e: except Exception as e:
console.print(f"[bold red]Setup failed: {e}[/bold red]") console.print(f"[bold red]Backup failed: {e}[/bold red]")
finally: finally:
await self.engine.close_connections() await self.engine.close_connections()
async def export_roles(self): async def backup_customization(self):
"""Option 2: Backup User Names & Avatars, Emojis, Stickers, etc."""
if not self.tokens_valid: return if not self.tokens_valid: return
try: try:
await self.engine.discord_reader.start() await self.engine.discord_reader.start()
await self.exporter.setup() await self.exporter.setup()
with console.status("[yellow]Exporting roles...[/yellow]"): with console.status("[yellow]Backing up customization (Members, Emojis & Stickers)...[/yellow]"):
roles = await self.exporter.export_roles() e_count, s_count, m_count = await self.exporter.export_customization()
console.print(f"[bold green]Exported {len(roles)} roles to roles.json[/bold green]") console.print(f"[bold green]Backup complete: {m_count} members, {e_count} emojis and {s_count} stickers saved to customization.json[/bold green]")
except discord.Forbidden as e:
console.print(f"[bold red]Customization backup failed: {e}[/bold red]")
console.print("[yellow]Hint: Missing Access. Ensure 'Server Members Intent' is enabled in the Discord Developer Portal.[/yellow]")
except Exception as e: except Exception as e:
console.print(f"[bold red]Role export failed: {e}[/bold red]") console.print(f"[bold red]Customization backup failed: {e}[/bold red]")
finally: finally:
await self.engine.close_connections() await self.engine.close_connections()
async def export_assets(self): async def backup_messages(self):
"""Option 3: Backup Channels structure and all message history."""
if not self.tokens_valid: return if not self.tokens_valid: return
try: try:
await self.engine.discord_reader.start() await self.engine.discord_reader.start()
await self.exporter.setup() await self.exporter.setup()
with console.status("[yellow]Exporting emojis and stickers...[/yellow]"):
e_count, s_count = await self.exporter.export_emojis_stickers()
console.print(f"[bold green]Exported {e_count} emojis and {s_count} stickers.[/bold green]")
except Exception as e:
console.print(f"[bold red]Asset export failed: {e}[/bold red]")
finally:
await self.engine.close_connections()
async def export_structure(self): # 1. Export structure first
if not self.tokens_valid: return
try:
await self.engine.discord_reader.start()
await self.exporter.setup()
with console.status("[yellow]Exporting channel structure...[/yellow]"): with console.status("[yellow]Exporting channel structure...[/yellow]"):
struct = await self.exporter.export_channels_structure() await self.exporter.export_channels_structure()
console.print(f"[bold green]Exported channel hierarchy to channels_structure.json[/bold green]")
except Exception as e:
console.print(f"[bold red]Structure export failed: {e}[/bold red]")
finally:
await self.engine.close_connections()
async def export_messages(self): # 2. Export messages
if not self.tokens_valid: return
try:
await self.engine.discord_reader.start()
await self.exporter.setup()
channels = await self.engine.discord_reader.get_channels() channels = await self.engine.discord_reader.get_channels()
console.print(f"\n[yellow]Found {len(channels)} channels to backup.[/yellow]")
console.print(f"\n[yellow]Found {len(channels)} channels to export.[/yellow]") if not Confirm.ask("Start message backup? This may take a while.", default=True):
if not Confirm.ask("Start message export? This may take a while.", default=True):
return return
with Progress( with Progress(
@ -174,26 +171,29 @@ class ExodusCLI:
TaskProgressColumn(), TaskProgressColumn(),
console=console console=console
) as progress: ) as progress:
overall_task = progress.add_task("[cyan]Exporting Channels...", total=len(channels)) overall_task = progress.add_task("[cyan]Exporting Channels...", total=len(channels))
for chan in channels: for chan in channels:
if chan.type not in [discord.ChannelType.text, discord.ChannelType.news, discord.ChannelType.voice]: if chan.type not in [discord.ChannelType.text, discord.ChannelType.news, discord.ChannelType.voice]:
progress.advance(overall_task) progress.advance(overall_task)
continue continue
progress.update(overall_task, description=f"[cyan]Backing up: {chan.name}")
progress.update(overall_task, description=f"[cyan]Exporting: {chan.name}")
await self.exporter.export_channel_messages(chan.id) await self.exporter.export_channel_messages(chan.id)
# Also export threads for this channel
await self.exporter.export_threads(chan.id) await self.exporter.export_threads(chan.id)
progress.advance(overall_task) progress.advance(overall_task)
console.print("[bold green]Message export complete![/bold green]") console.print("[bold green]Message backup complete![/bold green]")
except Exception as e: except Exception as e:
console.print(f"[bold red]Message export failed: {e}[/bold red]") console.print(f"[bold red]Message backup failed: {e}[/bold red]")
finally: finally:
await self.engine.close_connections() await self.engine.close_connections()
async def sync_backup(self):
"""Option 4: Update & Sync Backup (Runs a full pass but overwrites where needed)."""
console.print("[yellow]Syncing backup... (Re-validating all data)[/yellow]")
await self.backup_server_profile()
await self.backup_customization()
await self.backup_messages()
async def edit_configuration(self): async def edit_configuration(self):
# reuse or implement simplified version of edit_configuration from app.py # reuse or implement simplified version of edit_configuration from app.py
console.print("\n[bold]Configuration Editor[/bold]") console.print("\n[bold]Configuration Editor[/bold]")