Simon/hoi4botdc
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

modified: Dockerfile

Browse Source 
modified:   app.py
	modified:   requirements.txt
This commit is contained in:
SimolZimol
2025-10-28 15:04:02 +01:00
parent caec9b326a
commit 6bc6984876
3 changed files with 9 additions and 305 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
Dockerfile
View File

@@ -5,8 +5,6 @@ WORKDIR /app
RUN apt-get update && apt-get install -y \
libffi-dev \
libnacl-dev \
libopus0 \
ffmpeg \
python3-dev \
&& apt-get clean \
&& rm -rf /var/lib/apt/lists/*

306
app.py
View File

@@ -7,8 +7,7 @@ from dotenv import load_dotenv
import aiomysql
import json
from datetime import datetime
from typing import Optional, List, Dict, Tuple
import yt_dlp
from typing import Optional, List, Dict
# Load environment variables
load_dotenv()
@@ -79,7 +78,6 @@ intents = discord.Intents.default()
intents.message_content = True
intents.guilds = True
intents.members = True
intents.voice_states = True
bot = commands.Bot(command_prefix='!', intents=intents)
@@ -93,12 +91,6 @@ async def on_ready():
# Initialize database
await init_database()
# Preload opus if available (voice)
try:
if not discord.opus.is_loaded():
discord.opus.load_opus('libopus')
except Exception:
pass
# Set bot status
await bot.change_presence(
@@ -113,267 +105,6 @@ async def on_ready():
except Exception as e:
print(f'Failed to sync commands: {e}')
# =========================
# YouTube/Voice Player
# =========================
COOKIE_FILE = os.getenv('YT_COOKIES_PATH', '/app/cookie.txt')
class Song:
def __init__(self, title: str, url: str, webpage_url: str, duration: Optional[int]):
self.title = title
self.url = url
self.webpage_url = webpage_url
self.duration = duration
class GuildMusic:
def __init__(self):
self.queue: asyncio.Queue[Song] = asyncio.Queue()
self.current: Optional[Song] = None
self.play_next = asyncio.Event()
self.player_task: Optional[asyncio.Task] = None
music_states: Dict[int, GuildMusic] = {}
def get_guild_music(guild_id: int) -> GuildMusic:
st = music_states.get(guild_id)
if not st:
st = GuildMusic()
music_states[guild_id] = st
return st
async def maybe_defer(ctx: commands.Context):
"""Defer interaction responses to avoid 'interaction failed' on long ops (>3s)."""
try:
inter = getattr(ctx, 'interaction', None)
if inter and not inter.response.is_done():
await inter.response.defer(thinking=True)
except Exception:
pass
def ytdlp_opts() -> dict:
opts = {
'format': 'bestaudio/best',
'noplaylist': True,
'quiet': True,
'no_warnings': True,
'default_search': 'auto',
'nocheckcertificate': True,
'source_address': '0.0.0.0',
}
try:
if os.path.exists(COOKIE_FILE):
opts['cookiefile'] = COOKIE_FILE
except Exception:
pass
return opts
async def yt_extract(query: str) -> Song:
loop = asyncio.get_event_loop()
def _extract() -> Tuple[str, str, str, Optional[int]]:
with yt_dlp.YoutubeDL(ytdlp_opts()) as ydl:
info = ydl.extract_info(query, download=False)
if 'entries' in info:
info = info['entries'][0]
title = info.get('title')
url = info.get('url')
webpage_url = info.get('webpage_url') or info.get('original_url') or query
duration = info.get('duration')
return title, url, webpage_url, duration
title, url, webpage_url, duration = await loop.run_in_executor(None, _extract)
return Song(title, url, webpage_url, duration)
FFMPEG_BEFORE_OPTS = '-reconnect 1 -reconnect_streamed 1 -reconnect_delay_max 5'
FFMPEG_OPTS = '-vn'
async def ensure_voice(ctx: commands.Context) -> Optional[discord.VoiceClient]:
if not ctx.author or not getattr(ctx.author, 'voice', None) or not ctx.author.voice:
await ctx.reply("❌ You need to be in a voice channel.")
return None
channel = ctx.author.voice.channel
if not channel:
await ctx.reply("❌ Can't find your voice channel.")
return None
if ctx.voice_client is None:
try:
return await channel.connect(timeout=30, reconnect=True, self_deaf=True)
except Exception as e:
await ctx.reply(f"❌ Failed to join: {e}")
return None
elif ctx.voice_client.channel != channel:
try:
await ctx.voice_client.move_to(channel)
except Exception as e:
await ctx.reply(f"❌ Failed to move: {e}")
return None
return ctx.voice_client
async def start_player_loop(ctx: commands.Context):
guild = ctx.guild
if not guild:
return
state = get_guild_music(guild.id)
if state.player_task and not state.player_task.done():
return
async def runner():
while True:
state.play_next.clear()
try:
song = await asyncio.wait_for(state.queue.get(), timeout=300)
except asyncio.TimeoutError:
# idle for 5 minutes, disconnect
if ctx.voice_client and ctx.voice_client.is_connected():
await ctx.voice_client.disconnect()
break
state.current = song
source = discord.FFmpegPCMAudio(song.url, before_options=FFMPEG_BEFORE_OPTS, options=FFMPEG_OPTS)
ctx.voice_client.play(source, after=lambda e: state.play_next.set())
await state.play_next.wait()
state.current = None
state.player_task = asyncio.create_task(runner())
@bot.hybrid_command(name='join', description='Join your voice channel')
async def cmd_join(ctx):
await maybe_defer(ctx)
vc = await ensure_voice(ctx)
if vc:
await _safe_send(ctx, f"✅ Joined {vc.channel.mention}")
@bot.hybrid_command(name='leave', description='Leave the voice channel')
async def cmd_leave(ctx):
await maybe_defer(ctx)
if ctx.voice_client and ctx.voice_client.is_connected():
await ctx.voice_client.disconnect()
await _safe_send(ctx, "👋 Disconnected.")
else:
await _safe_send(ctx, "ℹ️ Not connected.")
@bot.hybrid_command(name='play', description='Play a YouTube URL or search term')
async def cmd_play(ctx, *, query: str):
await maybe_defer(ctx)
vc = await ensure_voice(ctx)
if not vc:
return
try:
song = await yt_extract(query)
state = get_guild_music(ctx.guild.id)
await state.queue.put(song)
await _safe_send(ctx, f"▶️ Queued: **{song.title}**")
await start_player_loop(ctx)
except Exception as e:
await _safe_send(ctx, f"❌ Failed to add: {e}")
@bot.hybrid_command(name='skip', description='Skip the current track')
async def cmd_skip(ctx):
await maybe_defer(ctx)
if ctx.voice_client and ctx.voice_client.is_playing():
ctx.voice_client.stop()
await _safe_send(ctx, "⏭️ Skipped.")
else:
await _safe_send(ctx, "ℹ️ Nothing is playing.")
@bot.hybrid_command(name='stop', description='Stop playback and clear the queue')
async def cmd_stop(ctx):
await maybe_defer(ctx)
state = get_guild_music(ctx.guild.id)
# Clear queue
try:
while True:
state.queue.get_nowait()
except asyncio.QueueEmpty:
pass
if ctx.voice_client and ctx.voice_client.is_playing():
ctx.voice_client.stop()
await _safe_send(ctx, "⏹️ Stopped and cleared queue.")
@bot.hybrid_command(name='pause', description='Pause playback')
async def cmd_pause(ctx):
await maybe_defer(ctx)
if ctx.voice_client and ctx.voice_client.is_playing():
ctx.voice_client.pause()
await _safe_send(ctx, "⏸️ Paused.")
else:
await _safe_send(ctx, "ℹ️ Nothing is playing.")
@bot.hybrid_command(name='resume', description='Resume playback')
async def cmd_resume(ctx):
await maybe_defer(ctx)
if ctx.voice_client and ctx.voice_client.is_paused():
ctx.voice_client.resume()
await _safe_send(ctx, "▶️ Resumed.")
else:
await _safe_send(ctx, "ℹ️ Nothing is paused.")
@bot.hybrid_command(name='np', description='Show the currently playing track')
async def cmd_nowplaying(ctx):
await maybe_defer(ctx)
state = get_guild_music(ctx.guild.id)
if state.current:
dur = f" ({state.current.duration//60}:{state.current.duration%60:02d})" if state.current.duration else ""
await _safe_send(ctx, f"🎶 Now playing: **{state.current.title}**{dur}\n{state.current.webpage_url}")
else:
await _safe_send(ctx, "ℹ️ Nothing is playing.")
@bot.hybrid_command(name='queue', description='Show queued tracks')
async def cmd_queue(ctx):
await maybe_defer(ctx)
state = get_guild_music(ctx.guild.id)
if state.queue.empty():
await _safe_send(ctx, "🗒️ Queue is empty.")
return
# Snapshot queue without draining
items: List[Song] = []
try:
while True:
items.append(state.queue.get_nowait())
except asyncio.QueueEmpty:
pass
# Put them back
for s in items:
await state.queue.put(s)
desc = "\n".join([f"{i+1}. {s.title}" for i, s in enumerate(items[:10])])
more = state.queue.qsize() - len(items[:10])
if more > 0:
desc += f"\n...and {more} more"
embed = discord.Embed(title="🎼 Queue", description=desc, color=discord.Color.purple())
await _safe_send(ctx, embed=embed)
@bot.hybrid_command(name='voicediag', description='Diagnose voice playback environment')
async def cmd_voicediag(ctx):
await maybe_defer(ctx)
details = []
details.append(f"discord.py: {discord.__version__}")
# PyNaCl check
try:
import nacl
details.append(f"PyNaCl: {getattr(nacl, '__version__', 'present')}")
except Exception as e:
details.append(f"PyNaCl: missing ({e})")
# Opus check
try:
loaded = discord.opus.is_loaded()
details.append(f"Opus loaded: {loaded}")
except Exception as e:
details.append(f"Opus check failed: {e}")
# FFmpeg check
ff = "unknown"
try:
proc = await asyncio.create_subprocess_exec('ffmpeg', '-version', stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.STDOUT)
out, _ = await proc.communicate()
ff = out.decode(errors='ignore').splitlines()[0][:120]
except FileNotFoundError:
ff = "not found"
except Exception as e:
ff = f"error: {e}"
details.append(f"ffmpeg: {ff}")
# Cookie file
details.append(f"cookie.txt present: {os.path.exists(COOKIE_FILE)} at {COOKIE_FILE}")
embed = discord.Embed(title='Voice Diagnostics', description='\n'.join(details), color=discord.Color.dark_grey())
await _safe_send(ctx, embed=embed)
@bot.event
async def on_guild_join(guild):
"""Event triggered when the bot joins a server"""
@@ -1617,14 +1348,14 @@ async def hoi4leaderboard(ctx, game_type: Optional[str] = "standard", limit: Opt
async def on_command_error(ctx, error):
"""Handles command errors"""
if isinstance(error, commands.CheckFailure):
await _safe_send(ctx, "❌ You don't have permission to use this command!")
await ctx.send("❌ You don't have permission to use this command!")
elif isinstance(error, commands.CommandNotFound):
# Silently ignore command not found errors
pass
elif isinstance(error, commands.MissingRequiredArgument):
await _safe_send(ctx, f"❌ Missing arguments! Command: `{ctx.command}`")
await ctx.send(f"❌ Missing arguments! Command: `{ctx.command}`")
elif isinstance(error, commands.BadArgument):
await _safe_send(ctx, "❌ Invalid argument!")
await ctx.send("❌ Invalid argument!")
else:
# Log detailed error information
print(f"❌ Unknown error in command '{ctx.command}': {type(error).__name__}: {error}")
@@ -1633,34 +1364,9 @@ async def on_command_error(ctx, error):
# Send detailed error to user if owner
if ctx.author.id == OWNER_ID:
await _safe_send(ctx, f"❌ **Error Details (Owner only):**\n```python\n{type(error).__name__}: {str(error)[:1800]}\n```")
await ctx.send(f"❌ **Error Details (Owner only):**\n```python\n{type(error).__name__}: {str(error)[:1800]}\n```")
else:
await _safe_send(ctx, "❌ An unknown error occurred!")
async def _safe_send(ctx: commands.Context, content: str = None, **kwargs):
"""Send a message safely for both message and slash contexts, even if the interaction timed out."""
try:
if getattr(ctx, 'interaction', None):
# If we have an interaction, try normal response first
interaction = ctx.interaction
if not interaction.response.is_done():
await interaction.response.send_message(content=content, **kwargs)
return
# Otherwise use followup
await interaction.followup.send(content=content, **kwargs)
return
# Fallback to classic send
await ctx.send(content=content, **kwargs)
except discord.NotFound:
# Interaction unknown/expired, try channel.send
try:
channel = getattr(ctx, 'channel', None)
if channel:
await channel.send(content=content, **kwargs)
except Exception:
pass
except Exception:
pass
await ctx.send("❌ An unknown error occurred!")
async def main():
"""Main function to start the bot"""

4
requirements.txt
View File

@@ -1,7 +1,7 @@
discord.py==2.3.2
python-dotenv==1.0.0
aiohttp==3.9.1
asyncpg==0.29.0
psycopg2-binary==2.9.9
aiomysql==0.2.0
PyMySQL==1.1.0
yt-dlp==2025.1.26
PyNaCl==1.5.0