267 lines
9.2 KiB
Python
267 lines
9.2 KiB
Python
import os
|
|
import requests
|
|
import base64
|
|
import re
|
|
import json
|
|
from datetime import datetime
|
|
import discord
|
|
from discord.ext import commands
|
|
from discord.ui import Button, View
|
|
from dotenv import load_dotenv
|
|
|
|
# DISCORD SERVER + DM DIRECTUS BOT
|
|
|
|
load_dotenv()
|
|
|
|
# Configuration
|
|
API_BASE_URL = os.getenv("API_BASE_URL") # Configure in .env
|
|
BEARER_TOKEN = os.getenv("BEARER_TOKEN") # Configure in .env
|
|
DISCORD_TOKEN = os.getenv("DISCORD_TOKEN") # Configure in .env
|
|
ALLOWED_CHANNELS = [123456789] # Replace with your channel IDs
|
|
|
|
# UUID validation pattern
|
|
UUID_PATTERN = re.compile(r'^[0-9a-fA-F]{8}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{12}$')
|
|
|
|
def log_activity(message: str) -> None:
|
|
timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
|
|
print(f"[{timestamp}] {message}")
|
|
|
|
intents = discord.Intents.default()
|
|
intents.message_content = True
|
|
bot = commands.Bot(command_prefix='!', intents=intents)
|
|
|
|
class SearchResultsView(View):
|
|
def __init__(self, results):
|
|
super().__init__(timeout=60)
|
|
self.results = results
|
|
self.original_message = None
|
|
self.add_buttons()
|
|
|
|
def add_buttons(self):
|
|
for idx, item in enumerate(self.results[:25], 1):
|
|
self.add_item(ResultButton(item, idx))
|
|
|
|
class ResultButton(Button):
|
|
def __init__(self, item, index):
|
|
super().__init__(
|
|
label=f"{index}",
|
|
style=discord.ButtonStyle.primary,
|
|
custom_id=f"result_{item['uuid']}"
|
|
)
|
|
self.item = item
|
|
|
|
async def callback(self, interaction: discord.Interaction):
|
|
if self.view.original_message:
|
|
try:
|
|
await self.view.original_message.delete()
|
|
except (discord.NotFound, discord.Forbidden):
|
|
pass
|
|
await self.show_addon_embed(interaction)
|
|
|
|
async def show_addon_embed(self, interaction: discord.Interaction):
|
|
data = self.item
|
|
download_hashes = data.get('download_hash', '')
|
|
decoded_urls = []
|
|
|
|
if download_hashes:
|
|
for i, hash_part in enumerate(download_hashes.split(','), 1):
|
|
try:
|
|
decoded = base64.b64decode(hash_part.strip()).decode('utf-8')
|
|
decoded_urls.append(f"{i}. {decoded}")
|
|
except:
|
|
decoded_urls.append(f"{i}. Invalid hash")
|
|
|
|
embed = discord.Embed(
|
|
title=data.get('name', 'N/A'),
|
|
color=discord.Color.blue()
|
|
)
|
|
embed.add_field(name="UUID", value=data.get('uuid', 'N/A'), inline=False)
|
|
embed.add_field(name="Creator", value=data.get('creator', 'N/A'), inline=True)
|
|
embed.add_field(name="Version", value=data.get('version', 'N/A'), inline=True)
|
|
|
|
if decoded_urls:
|
|
embed.add_field(
|
|
name="Download URLs",
|
|
value="\n".join(decoded_urls),
|
|
inline=False
|
|
)
|
|
|
|
await interaction.response.send_message(embed=embed, ephemeral=True)
|
|
|
|
@bot.event
|
|
async def on_ready():
|
|
log_activity(f'Logged in as {bot.user.name}')
|
|
try:
|
|
synced = await bot.tree.sync()
|
|
log_activity(f"Synced {len(synced)} commands")
|
|
except Exception as e:
|
|
log_activity(f"Command sync error: {str(e)}")
|
|
|
|
async def validate_channel(ctx):
|
|
if isinstance(ctx.channel, discord.DMChannel) or ctx.channel.id in ALLOWED_CHANNELS:
|
|
return True
|
|
embed = discord.Embed(
|
|
description="❌ This command is not available in this channel.",
|
|
color=discord.Color.red()
|
|
)
|
|
await ctx.send(embed=embed, ephemeral=True, delete_after=5)
|
|
return False
|
|
|
|
@bot.hybrid_command(name="mpbot", description="Search addons or lookup by UUID")
|
|
async def mpbot(ctx: commands.Context, *, query: str):
|
|
"""Main command handler for addon searches"""
|
|
if not await validate_channel(ctx):
|
|
return
|
|
|
|
# Delete original message if in guild channel
|
|
if ctx.guild and ctx.interaction is None:
|
|
try:
|
|
await ctx.message.delete()
|
|
except (discord.NotFound, discord.Forbidden):
|
|
pass
|
|
|
|
query = query.strip().lower()
|
|
|
|
if query == "help":
|
|
await send_start(ctx)
|
|
elif query == "cancel":
|
|
await send_cancel(ctx)
|
|
elif UUID_PATTERN.match(query):
|
|
await process_uuid(ctx, query)
|
|
else:
|
|
await perform_search(ctx, query)
|
|
|
|
async def send_start(ctx):
|
|
embed = discord.Embed(
|
|
title="Addon Search Bot",
|
|
description="I can help you find addons!\n\n"
|
|
"🔍 **Search**: `/mpbot [search terms]`\n"
|
|
"🔎 **UUID Lookup**: `/mpbot [uuid]`\n"
|
|
"🆘 **Help**: `/mpbot help`\n"
|
|
"❌ **Cancel**: `/mpbot cancel`",
|
|
color=discord.Color.green()
|
|
)
|
|
await ctx.send(embed=embed, ephemeral=True)
|
|
|
|
async def send_cancel(ctx):
|
|
embed = discord.Embed(
|
|
description="⚠️ No active operation to cancel",
|
|
color=discord.Color.orange()
|
|
)
|
|
await ctx.send(embed=embed, ephemeral=True, delete_after=5)
|
|
|
|
async def perform_search(ctx, query):
|
|
try:
|
|
headers = {"Authorization": f"Bearer {BEARER_TOKEN}"}
|
|
search_params = {
|
|
"filter": json.dumps({
|
|
"_or": [
|
|
{"name": {"_icontains": query}},
|
|
{"creator": {"_icontains": query}}
|
|
]
|
|
}),
|
|
"fields": "uuid,name,creator,version,download_hash",
|
|
"limit": "10"
|
|
}
|
|
|
|
response = requests.get(API_BASE_URL, headers=headers, params=search_params)
|
|
response.raise_for_status()
|
|
|
|
search_data = response.json().get('data', [])
|
|
|
|
if not search_data:
|
|
embed = discord.Embed(
|
|
description="🔍 No results found for your query.",
|
|
color=discord.Color.red()
|
|
)
|
|
await ctx.send(embed=embed, ephemeral=True)
|
|
return
|
|
|
|
results_list = [
|
|
f"{idx}. **{item.get('name', 'N/A')}** by {item.get('creator', 'N/A')}"
|
|
for idx, item in enumerate(search_data, 1)
|
|
]
|
|
|
|
embed = discord.Embed(
|
|
title=f"Search Results for '{query}'",
|
|
description="\n".join(results_list) + "\n\nClick a button below to view details:",
|
|
color=discord.Color.blue()
|
|
)
|
|
|
|
view = SearchResultsView(search_data)
|
|
sent_message = await ctx.send(embed=embed, view=view)
|
|
view.original_message = sent_message
|
|
|
|
except requests.exceptions.HTTPError as e:
|
|
embed = discord.Embed(
|
|
title="API Error",
|
|
description=f"Error code: {e.response.status_code}",
|
|
color=discord.Color.red()
|
|
)
|
|
await ctx.send(embed=embed, ephemeral=True)
|
|
except Exception as e:
|
|
log_activity(f"Search error: {str(e)}")
|
|
embed = discord.Embed(
|
|
title="Error",
|
|
description="⚠️ Error processing search",
|
|
color=discord.Color.red()
|
|
)
|
|
await ctx.send(embed=embed, ephemeral=True)
|
|
|
|
async def process_uuid(ctx, uuid):
|
|
try:
|
|
headers = {"Authorization": f"Bearer {BEARER_TOKEN}"}
|
|
response = requests.get(f"{API_BASE_URL}{uuid}", headers=headers)
|
|
response.raise_for_status()
|
|
|
|
data = response.json().get('data', {})
|
|
if not data:
|
|
embed = discord.Embed(
|
|
description="❌ No information found for this UUID.",
|
|
color=discord.Color.red()
|
|
)
|
|
await ctx.send(embed=embed, ephemeral=True)
|
|
return
|
|
|
|
download_hashes = data.get('download_hash', '')
|
|
decoded_urls = []
|
|
|
|
if download_hashes:
|
|
for i, hash_part in enumerate(download_hashes.split(','), 1):
|
|
try:
|
|
decoded = base64.b64decode(hash_part.strip()).decode('utf-8')
|
|
decoded_urls.append(f"{i}. {decoded}")
|
|
except:
|
|
decoded_urls.append(f"{i}. Invalid hash")
|
|
|
|
embed = discord.Embed(
|
|
title=data.get('name', 'N/A'),
|
|
color=discord.Color.blue()
|
|
)
|
|
embed.add_field(name="UUID", value=data.get('uuid', 'N/A'), inline=False)
|
|
embed.add_field(name="Creator", value=data.get('creator', 'N/A'), inline=True)
|
|
embed.add_field(name="Version", value=data.get('version', 'N/A'), inline=True)
|
|
|
|
if decoded_urls:
|
|
embed.add_field(
|
|
name="Download URLs",
|
|
value="\n".join(decoded_urls),
|
|
inline=False
|
|
)
|
|
|
|
await ctx.send(embed=embed, ephemeral=True)
|
|
|
|
except requests.exceptions.HTTPError as e:
|
|
description = f"🔧 API Error: {e.response.status_code}" if e.response.status_code != 403 else "⛔ Access denied for this UUID"
|
|
embed = discord.Embed(description=description, color=discord.Color.red())
|
|
await ctx.send(embed=embed, ephemeral=True)
|
|
except Exception as e:
|
|
log_activity(f"UUID processing error: {str(e)}")
|
|
embed = discord.Embed(
|
|
description="⚠️ Error fetching UUID details",
|
|
color=discord.Color.red()
|
|
)
|
|
await ctx.send(embed=embed, ephemeral=True)
|
|
|
|
if __name__ == '__main__':
|
|
bot.run(DISCORD_TOKEN) |