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

modified: .gitignore

Browse Source 
modified:   Dockerfile
	modified:   README.md
	modified:   app.py
This commit is contained in:
SimolZimol
2025-10-28 01:57:59 +01:00
parent c253767497
commit 9d91d7120b
4 changed files with 49 additions and 245 deletions
Download Patch File Download Diff File Expand all files Collapse all files

224
app.py
View File

@@ -11,8 +11,6 @@ from typing import Optional, List, Dict
from concurrent.futures import ThreadPoolExecutor
import functools
import traceback
import aiohttp
import time
import base64
from pathlib import Path
@@ -374,40 +372,33 @@ except Exception as e:
print(f"⚠️ Cookie configuration error: {e}")
YTDL_OPTS = {
'format': 'bestaudio[ext=m4a]/bestaudio/best',
'format': 'bestaudio/best',
'noplaylist': True,
'quiet': True,
'default_search': 'ytsearch',
'skip_download': True,
}
def get_ytdl_opts(client_hint: Optional[str] = None, use_cookies: bool = True) -> Dict:
"""Build yt-dlp options dynamically, injecting cookies and headers if configured.
client_hint: preferred YouTube client (e.g., 'android', 'web', 'ios', 'tv').
"""
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_env = os.getenv('YTDL_YT_CLIENT', 'android')
yt_client = client_hint or yt_client_env
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
# player_client hint (e.g., android)
opts['extractor_args']['youtube']['player_client'] = [yt_client]
# Use cookies if available
if use_cookies:
if YTDL_COOKIEFILE:
opts['cookiefile'] = YTDL_COOKIEFILE
elif YTDL_COOKIESFROMBROWSER:
opts['cookiesfrombrowser'] = YTDL_COOKIESFROMBROWSER
# Optional force IPv4 (set YTDL_FORCE_IPV4=1)
if os.getenv('YTDL_FORCE_IPV4'):
opts['force_ipv4'] = True
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"
@@ -453,81 +444,24 @@ async def _ensure_connected(ctx: commands.Context) -> Optional[discord.VoiceClie
async def _ytdlp_extract(loop: asyncio.AbstractEventLoop, query: str) -> Optional[Dict]:
if not ytdlp:
return None
# Try multiple YouTube client profiles to bypass some restrictions
env_client = os.getenv('YTDL_YT_CLIENT', 'android')
cookies_present = bool(YTDL_COOKIEFILE or YTDL_COOKIESFROMBROWSER)
# Helper to try a sequence of clients with/without cookies
async def try_clients(clients: List[str], use_cookies: bool) -> Optional[Dict]:
nonlocal query
last_err: Optional[Exception] = None
for client in clients:
def _extract_with_client():
with ytdlp.YoutubeDL(get_ytdl_opts(client, use_cookies=use_cookies)) 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_err = e
msg = str(e)
print(f"⚠️ yt-dlp attempt failed (cookies={'on' if use_cookies else 'off'}) client='{client}': {msg[:200]}")
retriable_markers = (
'Sign in to confirm', 'Use --cookies', 'pass cookies', 'HTTP Error 403',
'Signature extraction failed', 'Requested format is not available',
'Only images are available', 'missing a url', 'SABR streaming'
)
if any(m in msg for m in retriable_markers):
continue
else:
break
if last_err:
return None
return None
# Build ordered lists
# When cookies are present, android/ios are skipped by yt-dlp; so try web/tv first with cookies.
# If that fails (e.g., SABR), try without cookies to enable android extraction.
cookie_clients: List[str] = []
for c in [env_client, 'web', 'tv']:
if c in ('web', 'tv') and c not in cookie_clients:
cookie_clients.append(c)
nocookie_clients: List[str] = []
for c in [env_client, 'android', 'ios', 'web', 'tv']:
if c not in nocookie_clients:
nocookie_clients.append(c)
last_error: Optional[Exception] = None
# Pass 1: with cookies (if provided)
if cookies_present:
info = await try_clients(cookie_clients, use_cookies=True)
if info:
return info
# Pass 2: without cookies to allow android/ios
info = await try_clients(nocookie_clients, use_cookies=False)
if info:
return info
else:
# No cookies: try standard nocookie list
info = await try_clients(nocookie_clients, use_cookies=False)
if info:
return info
# All attempts failed
# At this point, all attempts failed; print last traceback if any
def _extract():
with ytdlp.YoutubeDL(get_ytdl_opts()) as ytdl:
return ytdl.extract_info(query, download=False)
try:
if cookies_present:
print("❌ yt-dlp: All attempts failed with and without cookies. Consider trying another client or refreshing cookies.")
else:
print("❌ yt-dlp: All attempts failed without cookies. Consider providing cookies or changing YTDL_YT_CLIENT.")
except Exception:
pass
return None
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 as e:
# Make YouTube cookie issues clearer in logs
msg = str(e)
if 'Sign in to confirm you’re not a bot' 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
async def _create_audio_source(loop: asyncio.AbstractEventLoop, search: str, volume: float):
# Accept either URL or search text; prepend ytsearch1: if not a URL
@@ -622,11 +556,8 @@ async def play(ctx: commands.Context, *, query: str):
await ctx.reply("🔎 Searching…")
item = await _create_audio_source(loop, query, state['volume'])
if not item:
# Give a hint if cookies or client change likely required
hint = (
"If this is YouTube, try YTDL_YT_CLIENT=android (or web) and/or provide cookies via "
"YTDL_COOKIES_FILE or YTDL_COOKIES_B64."
)
# 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)
@@ -713,83 +644,6 @@ async def volume(ctx: commands.Context, percent: int):
now['source'].volume = vol
await ctx.reply(f"🔊 Volume set to {percent}%.")
# ------------- Connectivity / Web Test -------------
@bot.hybrid_command(name='webtest', description='Test outbound web access and YouTube extraction')
async def webtest(ctx: commands.Context, *, query: Optional[str] = None):
"""Check basic web reachability and try a lightweight yt-dlp extraction.
- Tests HTTP GET to a few endpoints
- Tries yt-dlp search or URL extraction (if yt-dlp is available)
"""
urls = [
("Google 204", "https://www.google.com/generate_204"),
("YouTube", "https://www.youtube.com"),
("Discord API", "https://discord.com/api/v10/gateway"),
]
results = []
timeout = aiohttp.ClientTimeout(total=8)
try:
async with aiohttp.ClientSession(timeout=timeout) as session:
for name, url in urls:
t0 = time.monotonic()
status = None
err = None
try:
async with session.get(url, allow_redirects=True) as resp:
status = resp.status
except Exception as e:
err = str(e)
dt = (time.monotonic() - t0) * 1000
results.append((name, url, status, err, dt))
except Exception as e:
results.append(("session", "<session>", None, f"Session error: {e}", 0.0))
# Try yt-dlp extraction if available
ytdlp_ok = False
ytdlp_msg = ""
chosen = query or "ytsearch1:never gonna give you up"
try:
loop = asyncio.get_running_loop()
info = await _ytdlp_extract(loop, chosen)
if info:
ytdlp_ok = True
ytdlp_msg = f"Success: {info.get('title', 'unknown title')}"
else:
ytdlp_msg = "No info returned (possibly cookies required or blocked)."
except Exception as e:
ytdlp_msg = f"Error: {e}"
# Cookies configured?
cookies_file = os.getenv('YTDL_COOKIES_FILE')
cookies_b64 = bool(os.getenv('YTDL_COOKIES_B64'))
from_browser = os.getenv('YTDL_COOKIES_FROM_BROWSER')
desc_lines = []
for name, url, status, err, dt in results:
if status is not None:
desc_lines.append(f"• {name}: {status} ({dt:.0f} ms)")
else:
desc_lines.append(f"• {name}: ❌ {err or 'unknown error'}")
desc = "\n".join(desc_lines)
embed = discord.Embed(title="🌐 Web Test", description=desc, color=discord.Color.teal())
embed.add_field(name="yt-dlp", value=("✅ " + ytdlp_msg) if ytdlp_ok else ("❌ " + ytdlp_msg), inline=False)
cookie_status = []
if cookies_file and os.path.exists(cookies_file):
cookie_status.append(f"file: {cookies_file}")
elif cookies_file:
cookie_status.append(f"file missing: {cookies_file}")
if cookies_b64:
cookie_status.append("b64: provided")
if from_browser:
cookie_status.append(f"from-browser: {from_browser}")
if not cookie_status:
cookie_status.append("none")
embed.add_field(name="yt-dlp cookies", value=", ".join(cookie_status), inline=False)
await ctx.reply(embed=embed)
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:
@@ -1855,23 +1709,11 @@ async def on_command_error(ctx, error):
import traceback
traceback.print_exc()
# Send detailed error to user if owner; be safe for slash interactions
try:
content_owner = f"❌ **Error Details (Owner only):**\n```python\n{type(error).__name__}: {str(error)[:1800]}\n```"
content_user = "❌ An unknown error occurred!"
if hasattr(ctx, 'interaction') and ctx.interaction is not None:
if not ctx.interaction.response.is_done():
await ctx.interaction.response.send_message(content_owner if ctx.author.id == OWNER_ID else content_user, ephemeral=ctx.author.id != OWNER_ID)
else:
await ctx.interaction.followup.send(content_owner if ctx.author.id == OWNER_ID else content_user, ephemeral=ctx.author.id != OWNER_ID)
else:
await ctx.send(content_owner if ctx.author.id == OWNER_ID else content_user)
except Exception as send_err:
# Final fallback: try reply, then ignore
try:
await ctx.reply("❌ Error occurred, and sending details failed.")
except Exception:
pass
# Send detailed error to user if owner
if ctx.author.id == OWNER_ID:
await ctx.send(f"❌ **Error Details (Owner only):**\n```python\n{type(error).__name__}: {str(error)[:1800]}\n```")
else:
await ctx.send("❌ An unknown error occurred!")
async def main():
"""Main function to start the bot"""