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 13:27:13 +01:00
parent 664b3c2243
commit 0168321420
3 changed files with 218 additions and 386 deletions
Download Patch File Download Diff File Expand all files Collapse all files

11
Dockerfile
View File

@@ -3,16 +3,16 @@ FROM python:3.10-slim
WORKDIR /app
RUN apt-get update && apt-get install -y \
ffmpeg \
libffi-dev \
libnacl-dev \
libopus0 \
ffmpeg \
python3-dev \
&& apt-get clean \
&& rm -rf /var/lib/apt/lists/*
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
# Ensure yt-dlp is up-to-date so extractor fixes are applied
RUN pip install --no-cache-dir -U yt-dlp
COPY . .
@@ -24,10 +24,5 @@ ENV DB_PORT=$DB_PORT
ENV DB_NAME=$DB_NAME
ENV DB_USER=$DB_USER
ENV DB_PASSWORD=$DB_PASSWORD
ENV YTDL_COOKIES_FILE=$YTDL_COOKIES_FILE
ENV YTDL_COOKIES_B64=$YTDL_COOKIES_B64
ENV YTDL_COOKIES_FROM_BROWSER=$YTDL_COOKIES_FROM_BROWSER
ENV YTDL_UA=$YTDL_UA
ENV YTDL_YT_CLIENT=$YTDL_YT_CLIENT
CMD ["python", "app.py"]

591
app.py
View File

@@ -7,18 +7,8 @@ from dotenv import load_dotenv
import aiomysql
import json
from datetime import datetime
from typing import Optional, List, Dict
from concurrent.futures import ThreadPoolExecutor
import functools
import traceback
import base64
from pathlib import Path
# Optional: YouTube extraction
try:
import yt_dlp as ytdlp
except Exception:
ytdlp = None
from typing import Optional, List, Dict, Tuple
import yt_dlp
# Load environment variables
load_dotenv()
@@ -92,9 +82,6 @@ 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 +92,12 @@ 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(
@@ -119,12 +112,214 @@ async def on_ready():
except Exception as e:
print(f'Failed to sync commands: {e}')
# Pre-warm voice: ensure PyNaCl is importable
# =========================
# 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
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:
import nacl
print(f"🔊 Voice ready (PyNaCl {getattr(nacl, '__version__', 'unknown')})")
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()
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):
vc = await ensure_voice(ctx)
if vc:
await ctx.reply(f"✅ Joined {vc.channel.mention}")
@bot.hybrid_command(name='leave', description='Leave the voice channel')
async def cmd_leave(ctx):
if ctx.voice_client and ctx.voice_client.is_connected():
await ctx.voice_client.disconnect()
await ctx.reply("👋 Disconnected.")
else:
await ctx.reply("ℹ️ Not connected.")
@bot.hybrid_command(name='play', description='Play a YouTube URL or search term')
async def cmd_play(ctx, *, query: str):
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 ctx.reply(f"▶️ Queued: **{song.title}**")
await start_player_loop(ctx)
except Exception as e:
print(f"⚠️ Voice (PyNaCl) not available: {e}")
await ctx.reply(f"❌ Failed to add: {e}")
@bot.hybrid_command(name='skip', description='Skip the current track')
async def cmd_skip(ctx):
if ctx.voice_client and ctx.voice_client.is_playing():
ctx.voice_client.stop()
await ctx.reply("⏭️ Skipped.")
else:
await ctx.reply("ℹ️ Nothing is playing.")
@bot.hybrid_command(name='stop', description='Stop playback and clear the queue')
async def cmd_stop(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 ctx.reply("⏹️ Stopped and cleared queue.")
@bot.hybrid_command(name='pause', description='Pause playback')
async def cmd_pause(ctx):
if ctx.voice_client and ctx.voice_client.is_playing():
ctx.voice_client.pause()
await ctx.reply("⏸️ Paused.")
else:
await ctx.reply("ℹ️ Nothing is playing.")
@bot.hybrid_command(name='resume', description='Resume playback')
async def cmd_resume(ctx):
if ctx.voice_client and ctx.voice_client.is_paused():
ctx.voice_client.resume()
await ctx.reply("▶️ Resumed.")
else:
await ctx.reply("ℹ️ Nothing is paused.")
@bot.hybrid_command(name='np', description='Show the currently playing track')
async def cmd_nowplaying(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 ctx.reply(f"🎶 Now playing: **{state.current.title}**{dur}\n{state.current.webpage_url}")
else:
await ctx.reply("ℹ️ Nothing is playing.")
@bot.hybrid_command(name='queue', description='Show queued tracks')
async def cmd_queue(ctx):
state = get_guild_music(ctx.guild.id)
if state.queue.empty():
await ctx.reply("🗒️ 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 ctx.reply(embed=embed)
@bot.event
async def on_guild_join(guild):
@@ -318,364 +513,6 @@ 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] = {}
# --- yt-dlp cookies and options support ---
# You can supply YouTube cookies to bypass bot checks/age gates:
# - YTDL_COOKIES_FILE: Path to a Netscape-format cookies.txt inside the container (e.g., mounted secret)
# - YTDL_COOKIES_B64: Base64-encoded content of a Netscape-format cookies.txt; will be written to app directory
# - YTDL_COOKIES_FROM_BROWSER: e.g., "chrome:Default" (only works on hosts with a real browser profile, usually not in containers)
# - YTDL_UA: Custom User-Agent string
# - YTDL_YT_CLIENT: youtube player client hint (e.g., android, web) to work around some restrictions (default: android)
YTDL_COOKIEFILE: Optional[str] = None
YTDL_COOKIESFROMBROWSER: Optional[tuple] = None
try:
# Priority 1: cookies from base64 env
_cookies_b64 = os.getenv('YTDL_COOKIES_B64')
if _cookies_b64:
try:
decoded = base64.b64decode(_cookies_b64)
cookie_path = Path(__file__).with_name('cookies.txt')
cookie_path.write_bytes(decoded)
YTDL_COOKIEFILE = str(cookie_path)
print(f"🍪 Wrote YouTube cookies to {YTDL_COOKIEFILE} (from YTDL_COOKIES_B64)")
except Exception as e:
print(f"⚠️ Failed to decode/write YTDL_COOKIES_B64: {e}")
# Priority 2: cookies from a file path
if not YTDL_COOKIEFILE:
_cookies_file = os.getenv('YTDL_COOKIES_FILE')
if _cookies_file and os.path.exists(_cookies_file):
YTDL_COOKIEFILE = _cookies_file
print(f"🍪 Using YouTube cookies file: {YTDL_COOKIEFILE}")
elif _cookies_file:
print(f"⚠️ YTDL_COOKIES_FILE set but not found: {_cookies_file}")
# Optional: cookies from browser (rarely usable in containers)
_cookies_from_browser = os.getenv('YTDL_COOKIES_FROM_BROWSER')
if _cookies_from_browser:
parts = _cookies_from_browser.split(':')
browser = parts[0].strip().lower() if parts else None
profile = parts[1].strip() if len(parts) > 1 else None
if browser:
# (browser, profile, keyring, container)
YTDL_COOKIESFROMBROWSER = (browser, profile, None, None)
print(f"🍪 Will try cookies from browser: {browser} profile={profile or 'default'}")
except Exception as e:
print(f"⚠️ Cookie configuration error: {e}")
YTDL_OPTS = {
'format': 'bestaudio/best',
'noplaylist': True,
'quiet': True,
'default_search': 'ytsearch',
'skip_download': True,
}
def get_ytdl_opts() -> Dict:
"""Build yt-dlp options dynamically, injecting cookies and headers if configured."""
opts = dict(YTDL_OPTS)
# UA and extractor tweaks
ua = os.getenv('YTDL_UA') or (
'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 '
'(KHTML, like Gecko) Chrome/118.0 Safari/537.36'
)
yt_client = os.getenv('YTDL_YT_CLIENT', 'android')
opts['http_headers'] = {'User-Agent': ua}
# Extractor args can help avoid some player config checks
opts.setdefault('extractor_args', {})
opts['extractor_args'].setdefault('youtube', {})
# player_client hint (e.g., android)
opts['extractor_args']['youtube']['player_client'] = [yt_client]
# Use cookies if available
if YTDL_COOKIEFILE:
opts['cookiefile'] = YTDL_COOKIEFILE
elif YTDL_COOKIESFROMBROWSER:
opts['cookiesfrombrowser'] = YTDL_COOKIESFROMBROWSER
return opts
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
# Try extraction with multiple player_client hints if extraction fails due to player/nsig issues.
# Start with the configured client, then fall back to common alternatives.
preferred = os.getenv('YTDL_YT_CLIENT', 'android')
candidates = [preferred]
for c in ('web', 'tv', 'android_embedded', 'firetv'):
if c not in candidates:
candidates.append(c)
last_exc = None
for client_hint in candidates:
def _extract_with_client(client=client_hint):
opts = get_ytdl_opts()
# Override player_client for this attempt
try:
opts['extractor_args']['youtube']['player_client'] = [client]
except Exception:
opts.setdefault('extractor_args', {}).setdefault('youtube', {})['player_client'] = [client]
with ytdlp.YoutubeDL(opts) as ytdl:
return ytdl.extract_info(query, download=False)
try:
info = await loop.run_in_executor(executor, _extract_with_client)
if info is None:
continue
if 'entries' in info:
info = info['entries'][0]
return info
except Exception as e:
last_exc = e
msg = str(e).lower()
# If it's a cookies/sign-in issue, surface a helpful message and stop trying
if 'sign in to confirm' in msg or 'use --cookies' in msg or 'pass cookies' in msg:
print("❌ yt-dlp error: YouTube requires cookies to proceed.")
print(" Provide YTDL_COOKIES_FILE or YTDL_COOKIES_B64 (Netscape cookies.txt).")
traceback.print_exc()
return None
# If nsig/sabr or unsupported client warnings occurred, try next client hint
if 'nsig extraction failed' in msg or 'sabr' in msg or 'unsupported client' in msg:
print(f"⚠️ yt-dlp warning with client={client_hint}: {e}")
# continue to try other clients
continue
# Otherwise, log and stop trying
traceback.print_exc()
return None
# If we tried everything and failed, log last exception
if last_exc:
print(f"❌ All yt-dlp client attempts failed. Last error: {type(last_exc).__name__}: {last_exc}")
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:
# Give a hint if cookies likely required
hint = "If this is YouTube, the server IP may be challenged. Provide YTDL_COOKIES_FILE or YTDL_COOKIES_B64."
await ctx.reply(f"❌ Couldn't get audio from that query.\n{hint}")
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:

2
requirements.txt
View File

@@ -3,5 +3,5 @@ python-dotenv==1.0.0
aiohttp==3.9.1
aiomysql==0.2.0
PyMySQL==1.1.0
yt-dlp>=2024.04.09
yt-dlp==2025.1.26
PyNaCl==1.5.0