240 lines
9.6 KiB
Python
Executable File
240 lines
9.6 KiB
Python
Executable File
# discord_tools/scripts/server_backup.py
|
|
|
|
import os
|
|
import sys
|
|
import json
|
|
import asyncio
|
|
import aiohttp
|
|
from datetime import datetime
|
|
import requests
|
|
|
|
# Add the parent directory to the Python path
|
|
script_dir = os.path.dirname(os.path.abspath(__file__))
|
|
project_root = os.path.dirname(os.path.dirname(script_dir))
|
|
sys.path.insert(0, project_root)
|
|
|
|
from discord_tools.config.settings import DISCORD_TOKEN, DATA_DIR
|
|
from discord_tools.utils.api_utils import make_discord_request
|
|
|
|
class DiscordServerBackup:
|
|
def __init__(self, guild_id):
|
|
self.guild_id = guild_id
|
|
self.backup_data = {}
|
|
self.backup_folder = os.path.join(DATA_DIR, f"server_backup_{guild_id}_{datetime.now().strftime('%Y%m%d_%H%M%S')}")
|
|
self.create_folder_structure()
|
|
|
|
def create_folder_structure(self):
|
|
folders = [
|
|
self.backup_folder,
|
|
os.path.join(self.backup_folder, "channels"),
|
|
os.path.join(self.backup_folder, "images"),
|
|
os.path.join(self.backup_folder, "images", "emojis"),
|
|
os.path.join(self.backup_folder, "images", "stickers"),
|
|
]
|
|
for folder in folders:
|
|
os.makedirs(folder, exist_ok=True)
|
|
|
|
def download_file(self, url, filepath):
|
|
response = requests.get(url)
|
|
if response.status_code == 200:
|
|
with open(filepath, 'wb') as f:
|
|
f.write(response.content)
|
|
print(f"Downloaded {os.path.basename(filepath)}")
|
|
else:
|
|
print(f"Failed to download {os.path.basename(filepath)}")
|
|
|
|
async def fetch_guild_info(self):
|
|
endpoint = f'/guilds/{self.guild_id}'
|
|
response = make_discord_request('GET', endpoint)
|
|
if response:
|
|
guild_info = response.json()
|
|
self.save_json("guild_info", guild_info)
|
|
print("Guild information fetched successfully.")
|
|
|
|
# Download icon
|
|
if 'icon' in guild_info:
|
|
icon_url = f"https://cdn.discordapp.com/icons/{self.guild_id}/{guild_info['icon']}.png"
|
|
self.download_file(icon_url, os.path.join(self.backup_folder, "images", "guild_icon.png"))
|
|
|
|
# Download banner
|
|
if 'banner' in guild_info:
|
|
banner_url = f"https://cdn.discordapp.com/banners/{self.guild_id}/{guild_info['banner']}.png"
|
|
self.download_file(banner_url, os.path.join(self.backup_folder, "images", "guild_banner.png"))
|
|
else:
|
|
print("Failed to fetch guild information.")
|
|
|
|
async def fetch_channels(self):
|
|
endpoint = f'/guilds/{self.guild_id}/channels'
|
|
response = make_discord_request('GET', endpoint)
|
|
if response:
|
|
channels = response.json()
|
|
self.save_json("channels", channels)
|
|
|
|
# Create channel structure
|
|
channel_structure = self.create_channel_structure(channels)
|
|
self.save_json("channel_structure", channel_structure)
|
|
|
|
# Fetch channel-specific settings
|
|
await self.fetch_channel_settings(channels)
|
|
|
|
print("Channel information and structure fetched successfully.")
|
|
else:
|
|
print("Failed to fetch channel information.")
|
|
|
|
def create_channel_structure(self, channels):
|
|
structure = {}
|
|
categories = {c['id']: c for c in channels if c['type'] == 4}
|
|
|
|
for channel in channels:
|
|
if channel['type'] != 4: # Not a category
|
|
parent_id = channel.get('parent_id')
|
|
if parent_id:
|
|
if parent_id not in structure:
|
|
structure[parent_id] = []
|
|
structure[parent_id].append(channel)
|
|
else:
|
|
if 'no_category' not in structure:
|
|
structure['no_category'] = []
|
|
structure['no_category'].append(channel)
|
|
|
|
return structure
|
|
|
|
async def fetch_channel_settings(self, channels):
|
|
for channel in channels:
|
|
endpoint = f'/channels/{channel["id"]}'
|
|
response = make_discord_request('GET', endpoint)
|
|
if response:
|
|
channel_data = response.json()
|
|
self.save_json(f"channels/{channel['id']}_settings", channel_data)
|
|
else:
|
|
print(f"Failed to fetch settings for channel {channel['name']}")
|
|
|
|
async def fetch_roles(self):
|
|
endpoint = f'/guilds/{self.guild_id}/roles'
|
|
response = make_discord_request('GET', endpoint)
|
|
if response:
|
|
roles = response.json()
|
|
self.save_json("roles", roles)
|
|
print("Role information fetched successfully.")
|
|
else:
|
|
print("Failed to fetch role information.")
|
|
|
|
async def fetch_emojis(self):
|
|
endpoint = f'/guilds/{self.guild_id}/emojis'
|
|
response = make_discord_request('GET', endpoint)
|
|
if response:
|
|
emojis = response.json()
|
|
self.save_json("emojis", emojis)
|
|
print("Emoji information fetched successfully.")
|
|
|
|
# Download custom emojis
|
|
for emoji in emojis:
|
|
emoji_url = f"https://cdn.discordapp.com/emojis/{emoji['id']}.png"
|
|
self.download_file(emoji_url, os.path.join(self.backup_folder, "images", "emojis", f"{emoji['name']}.png"))
|
|
else:
|
|
print("Failed to fetch emoji information.")
|
|
|
|
async def fetch_stickers(self):
|
|
endpoint = f'/guilds/{self.guild_id}/stickers'
|
|
response = make_discord_request('GET', endpoint)
|
|
if response:
|
|
stickers = response.json()
|
|
self.save_json("stickers", stickers)
|
|
print("Sticker information fetched successfully.")
|
|
|
|
# Download custom stickers
|
|
for sticker in stickers:
|
|
sticker_url = f"https://cdn.discordapp.com/stickers/{sticker['id']}.png"
|
|
self.download_file(sticker_url, os.path.join(self.backup_folder, "images", "stickers", f"{sticker['name']}.png"))
|
|
else:
|
|
print("Failed to fetch sticker information.")
|
|
|
|
async def fetch_invites(self):
|
|
endpoint = f'/guilds/{self.guild_id}/invites'
|
|
response = make_discord_request('GET', endpoint)
|
|
if response:
|
|
invites = response.json()
|
|
self.save_json("invites", invites)
|
|
print("Server invites fetched successfully.")
|
|
else:
|
|
print("Failed to fetch server invites.")
|
|
|
|
async def fetch_messages(self, channel_id, limit=None):
|
|
messages = []
|
|
last_message_id = None
|
|
while True:
|
|
endpoint = f'/channels/{channel_id}/messages?limit=100'
|
|
if last_message_id:
|
|
endpoint += f'&before={last_message_id}'
|
|
response = make_discord_request('GET', endpoint)
|
|
if response and response.status_code == 200:
|
|
new_messages = response.json()
|
|
if not new_messages:
|
|
break
|
|
messages.extend(new_messages)
|
|
last_message_id = new_messages[-1]['id']
|
|
if limit and len(messages) >= limit:
|
|
messages = messages[:limit]
|
|
break
|
|
else:
|
|
print(f"Failed to fetch messages for channel {channel_id}")
|
|
break
|
|
return messages
|
|
|
|
async def backup_messages(self):
|
|
channels_data = self.load_json("channels")
|
|
if not channels_data:
|
|
print("Channel information not available. Skipping message backup.")
|
|
return
|
|
|
|
text_channels = [channel for channel in channels_data if channel['type'] == 0]
|
|
|
|
for channel in text_channels:
|
|
print(f"Backing up messages from channel: {channel['name']}")
|
|
messages = await self.fetch_messages(channel['id'])
|
|
self.save_json(f"channels/{channel['id']}_messages", messages)
|
|
print(f"Backed up {len(messages)} messages from {channel['name']}")
|
|
|
|
async def create_backup(self, include_messages=False):
|
|
tasks = [
|
|
self.fetch_guild_info(),
|
|
self.fetch_channels(),
|
|
self.fetch_roles(),
|
|
self.fetch_emojis(),
|
|
self.fetch_stickers(),
|
|
self.fetch_invites(),
|
|
]
|
|
await asyncio.gather(*tasks)
|
|
|
|
if include_messages:
|
|
await self.backup_messages()
|
|
|
|
def save_json(self, filename, data):
|
|
filepath = os.path.join(self.backup_folder, f"{filename}.json")
|
|
with open(filepath, 'w', encoding='utf-8') as f:
|
|
json.dump(data, f, ensure_ascii=False, indent=4)
|
|
|
|
def load_json(self, filename):
|
|
filepath = os.path.join(self.backup_folder, f"{filename}.json")
|
|
try:
|
|
with open(filepath, 'r', encoding='utf-8') as f:
|
|
return json.load(f)
|
|
except FileNotFoundError:
|
|
print(f"File not found: {filepath}")
|
|
return None
|
|
|
|
async def main():
|
|
guild_id = input("Enter the Discord server ID to backup: ").strip()
|
|
|
|
if not guild_id.isdigit():
|
|
print("Invalid server ID. Please enter a numeric ID.")
|
|
return
|
|
|
|
include_messages = input("Do you want to backup all messages? (y/n): ").strip().lower() == 'y'
|
|
|
|
backup = DiscordServerBackup(guild_id)
|
|
await backup.create_backup(include_messages)
|
|
print(f"Backup completed. Files saved in {backup.backup_folder}")
|
|
|
|
if __name__ == "__main__":
|
|
asyncio.run(main()) |