263 lines
8.9 KiB
Python
263 lines
8.9 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}')
|
|
|
|
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)
|
|
|
|
async def send_cancel(ctx):
|
|
embed = discord.Embed(
|
|
description="⚠️ No active operation to cancel",
|
|
color=discord.Color.orange()
|
|
)
|
|
await ctx.send(embed=embed, delete_after=5)
|
|
|
|
@bot.event
|
|
async def on_message(message):
|
|
if message.author == bot.user:
|
|
return
|
|
|
|
# Allow DMs and specified channels
|
|
if not isinstance(message.channel, discord.DMChannel) and message.channel.id not in ALLOWED_CHANNELS:
|
|
return
|
|
|
|
if not message.content.startswith('!mpbot'):
|
|
return
|
|
|
|
ctx = await bot.get_context(message)
|
|
content = message.content[len('!mpbot'):].strip()
|
|
|
|
if not content:
|
|
await send_start(ctx)
|
|
return
|
|
|
|
if content.lower() == 'help':
|
|
await send_start(ctx)
|
|
elif content.lower() == 'cancel':
|
|
await send_cancel(ctx)
|
|
elif UUID_PATTERN.match(content):
|
|
await process_uuid(ctx, content)
|
|
else:
|
|
await perform_search(ctx, content)
|
|
|
|
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 data:
|
|
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)
|
|
else:
|
|
embed = discord.Embed(
|
|
description="❌ No information found for this UUID.",
|
|
color=discord.Color.red()
|
|
)
|
|
await ctx.send(embed=embed, ephemeral=True)
|
|
|
|
except requests.exceptions.HTTPError as e:
|
|
if e.response.status_code == 403:
|
|
embed = discord.Embed(
|
|
description="⛔ Access denied for this UUID",
|
|
color=discord.Color.red()
|
|
)
|
|
else:
|
|
embed = discord.Embed(
|
|
description=f"🔧 API Error: {e.response.status_code}",
|
|
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) |