Initial commit: Discord automation tools
This commit is contained in:
240
scripts/backup_server.py
Executable file
240
scripts/backup_server.py
Executable file
@@ -0,0 +1,240 @@
|
||||
# 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())
|
||||
Reference in New Issue
Block a user