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-27 22:35:59 +01:00
parent 18bb4de57e
commit d8067c8769
3 changed files with 274 additions and 4 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
Dockerfile
View File

@@ -3,8 +3,8 @@ FROM python:3.10-slim
WORKDIR /app
RUN apt-get update && apt-get install -y \
ffmpeg \
libffi-dev \
libnacl-dev \
python3-dev \
&& apt-get clean \
&& rm -rf /var/lib/apt/lists/*

270
app.py
View File

@@ -8,6 +8,15 @@ import aiomysql
import json
from datetime import datetime
from typing import Optional, List, Dict
from concurrent.futures import ThreadPoolExecutor
import functools
import traceback
# Optional: YouTube extraction
try:
import yt_dlp as ytdlp
except Exception:
ytdlp = None
# Load environment variables
load_dotenv()
@@ -81,6 +90,9 @@ intents.members = True
bot = commands.Bot(command_prefix='!', intents=intents)
# Thread pool for blocking operations like yt-dlp extraction
executor = ThreadPoolExecutor(max_workers=4)
@bot.event
async def on_ready():
"""Event triggered when the bot is ready"""
@@ -105,6 +117,13 @@ async def on_ready():
except Exception as e:
print(f'Failed to sync commands: {e}')
# Pre-warm voice: ensure PyNaCl is importable
try:
import nacl
print(f"🔊 Voice ready (PyNaCl {getattr(nacl, '__version__', 'unknown')})")
except Exception as e:
print(f"⚠️ Voice (PyNaCl) not available: {e}")
@bot.event
async def on_guild_join(guild):
"""Event triggered when the bot joins a server"""
@@ -297,6 +316,257 @@ def get_team_emoji(ctx: commands.Context, team_name: str) -> str:
return custom
return TEAM_EMOTE_OVERRIDES.get("default", "🎖️")
# =========================
# Music / YouTube playback
# =========================
# Global music state per guild
MUSIC_STATE: Dict[int, Dict] = {}
YTDL_OPTS = {
'format': 'bestaudio/best',
'noplaylist': True,
'quiet': True,
'default_search': 'ytsearch',
'skip_download': True,
}
FFMPEG_BEFORE = "-reconnect 1 -reconnect_streamed 1 -reconnect_delay_max 5"
FFMPEG_OPTS = {
'before_options': FFMPEG_BEFORE,
'options': '-vn'
}
def _get_state(guild: discord.Guild) -> Dict:
st = MUSIC_STATE.get(guild.id)
if not st:
st = MUSIC_STATE[guild.id] = {
'queue': [],
'now': None,
'volume': 0.5, # 50%
}
return st
async def _ensure_connected(ctx: commands.Context) -> Optional[discord.VoiceClient]:
if not ctx.guild:
await ctx.reply("❌ This command can only be used in a server.")
return None
if not ctx.author.voice or not ctx.author.voice.channel:
await ctx.reply("❌ You must be connected to a voice channel.")
return None
channel = ctx.author.voice.channel
if ctx.voice_client and ctx.voice_client.channel == channel:
return ctx.voice_client
if ctx.voice_client and ctx.voice_client.channel != channel:
try:
await ctx.voice_client.move_to(channel)
return ctx.voice_client
except Exception as e:
await ctx.reply(f"❌ Couldn't move to your channel: {e}")
return None
try:
vc = await channel.connect()
return vc
except Exception as e:
await ctx.reply(f"❌ Couldn't join voice: {e}")
return None
async def _ytdlp_extract(loop: asyncio.AbstractEventLoop, query: str) -> Optional[Dict]:
if not ytdlp:
return None
def _extract():
with ytdlp.YoutubeDL(YTDL_OPTS) as ytdl:
return ytdl.extract_info(query, download=False)
try:
info = await loop.run_in_executor(executor, _extract)
if info is None:
return None
if 'entries' in info:
info = info['entries'][0]
return info
except Exception:
traceback.print_exc()
return None
async def _create_audio_source(loop: asyncio.AbstractEventLoop, search: str, volume: float):
# Accept either URL or search text; prepend ytsearch1: if not a URL
if not (search.startswith('http://') or search.startswith('https://')):
search = f"ytsearch1:{search}"
info = await _ytdlp_extract(loop, search)
if not info:
return None
url = info.get('url')
webpage_url = info.get('webpage_url') or info.get('original_url') or url
title = info.get('title') or 'Unknown title'
uploader = info.get('uploader') or ''
thumb = None
if isinstance(info.get('thumbnails'), list) and info['thumbnails']:
thumb = info['thumbnails'][-1].get('url')
# Use PCM + volume transformer for runtime volume control
audio = discord.PCMVolumeTransformer(discord.FFmpegPCMAudio(url, **FFMPEG_OPTS), volume=max(0.0, min(2.0, volume)))
return {
'source': audio,
'title': title,
'webpage_url': webpage_url,
'uploader': uploader,
'thumbnail': thumb,
'stream_url': url,
}
def _start_next(guild: discord.Guild, ctx_channel: Optional[discord.abc.Messageable] = None):
state = _get_state(guild)
vc = guild.voice_client
if not vc:
state['now'] = None
state['queue'].clear()
return
if vc.is_playing() or vc.is_paused():
return
if not state['queue']:
state['now'] = None
return
item = state['queue'].pop(0)
state['now'] = item
def _after(error):
if error:
print(f"Audio error: {error}")
# Schedule next on event loop thread-safely
bot.loop.call_soon_threadsafe(_start_next, guild, None)
try:
vc.play(item['source'], after=_after)
except Exception as e:
print(f"Failed to start playback: {e}")
bot.loop.call_soon_threadsafe(_start_next, guild, None)
return
# Optionally announce now playing
if ctx_channel:
try:
embed = discord.Embed(title="🎶 Now Playing", description=f"[{item['title']}]({item['webpage_url']})", color=discord.Color.purple())
if item.get('thumbnail'):
embed.set_thumbnail(url=item['thumbnail'])
ctx_task = ctx_channel.send(embed=embed)
asyncio.create_task(ctx_task)
except Exception:
pass
# ------------- Music Commands -------------
@bot.hybrid_command(name='join', description='Join your voice channel')
async def join(ctx: commands.Context):
vc = await _ensure_connected(ctx)
if vc:
await ctx.reply(f"✅ Joined {vc.channel.mention}")
@bot.hybrid_command(name='leave', description='Leave the voice channel and clear the queue')
async def leave(ctx: commands.Context):
if not ctx.voice_client:
await ctx.reply("❌ I'm not in a voice channel.")
return
state = _get_state(ctx.guild)
state['queue'].clear()
state['now'] = None
try:
await ctx.voice_client.disconnect()
await ctx.reply("👋 Left the voice channel and cleared the queue.")
except Exception as e:
await ctx.reply(f"❌ Couldn't leave: {e}")
@bot.hybrid_command(name='play', description='Play a YouTube URL or search query')
async def play(ctx: commands.Context, *, query: str):
vc = await _ensure_connected(ctx)
if not vc:
return
loop = asyncio.get_running_loop()
state = _get_state(ctx.guild)
await ctx.reply("🔎 Searching…")
item = await _create_audio_source(loop, query, state['volume'])
if not item:
await ctx.reply("❌ Couldn't get audio from that query.")
return
state['queue'].append(item)
if not vc.is_playing() and not vc.is_paused() and not state['now']:
_start_next(ctx.guild, ctx.channel)
else:
await ctx.reply(f"➕ Queued: [{item['title']}]({item['webpage_url']})")
@bot.hybrid_command(name='skip', description='Skip the current track')
async def skip(ctx: commands.Context):
if not ctx.voice_client or not ctx.voice_client.is_playing():
await ctx.reply("❌ Nothing is playing.")
return
ctx.voice_client.stop()
await ctx.reply("⏭️ Skipped.")
@bot.hybrid_command(name='stop', description='Stop playback and clear the queue')
async def stop(ctx: commands.Context):
state = _get_state(ctx.guild)
state['queue'].clear()
state['now'] = None
if ctx.voice_client and (ctx.voice_client.is_playing() or ctx.voice_client.is_paused()):
ctx.voice_client.stop()
await ctx.reply("⏹️ Stopped and cleared the queue.")
@bot.hybrid_command(name='pause', description='Pause playback')
async def pause(ctx: commands.Context):
if not ctx.voice_client or not ctx.voice_client.is_playing():
await ctx.reply("❌ Nothing is playing.")
return
ctx.voice_client.pause()
await ctx.reply("⏸️ Paused.")
@bot.hybrid_command(name='resume', description='Resume playback')
async def resume(ctx: commands.Context):
if not ctx.voice_client or not ctx.voice_client.is_paused():
await ctx.reply("❌ Nothing to resume.")
return
ctx.voice_client.resume()
await ctx.reply("▶️ Resumed.")
@bot.hybrid_command(name='queue', description='Show the queue')
async def queue_cmd(ctx: commands.Context):
state = _get_state(ctx.guild)
now = state.get('now')
q = state.get('queue', [])
if not now and not q:
await ctx.reply("🗒️ Queue is empty.")
return
desc = ""
if now:
desc += f"Now: [{now['title']}]({now['webpage_url']})\n\n"
if q:
for i, it in enumerate(q[:10], 1):
desc += f"{i}. [{it['title']}]({it['webpage_url']})\n"
if len(q) > 10:
desc += f"…and {len(q)-10} more"
embed = discord.Embed(title="📜 Queue", description=desc, color=discord.Color.blurple())
await ctx.reply(embed=embed)
@bot.hybrid_command(name='np', description='Show the current track')
async def now_playing(ctx: commands.Context):
state = _get_state(ctx.guild)
now = state.get('now')
if not now:
await ctx.reply("❌ Nothing is playing.")
return
embed = discord.Embed(title="🎶 Now Playing", description=f"[{now['title']}]({now['webpage_url']})", color=discord.Color.purple())
if now.get('thumbnail'):
embed.set_thumbnail(url=now['thumbnail'])
await ctx.reply(embed=embed)
@bot.hybrid_command(name='volume', description='Set the player volume (0-200%)')
async def volume(ctx: commands.Context, percent: int):
if percent < 0 or percent > 200:
await ctx.reply("❌ Volume must be between 0 and 200.")
return
state = _get_state(ctx.guild)
vol = percent / 100.0
state['volume'] = vol
# Update current source if playing
now = state.get('now')
if now and now['source'] and isinstance(now['source'], discord.PCMVolumeTransformer):
now['source'].volume = vol
await ctx.reply(f"🔊 Volume set to {percent}%.")
def _flag_from_iso2(code: str) -> Optional[str]:
"""Return unicode flag from 2-letter ISO code (e.g., 'DE' -> 🇩🇪)."""
if not code or len(code) != 2:

6
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
PyMySQL==1.1.0
yt-dlp>=2024.04.09
PyNaCl==1.5.0