""" Telegram bot for viral clip processing. Receives YouTube URLs, processes via main.py (using the viral venv), sends progress updates. Fixes applied vs previous version: - main.py is now launched with VENV_PYTHON (/home/ren/viral/venv/bin/python3) - get_video_title uses /home/ren/yt/yt-dl with the yt venv, not bare yt-dlp - Comprehensive logging to stdout (journalctl) at every step - Telegram progress messages show exact clip counts once known - Error details are sent back to the Telegram chat - Completion message lists all clips with their Nextcloud folder """ import asyncio import logging import os import re import subprocess import sys import shutil import tempfile from datetime import datetime from pathlib import Path from dotenv import load_dotenv from telegram import Update from telegram.ext import Application, CommandHandler, MessageHandler, filters, ContextTypes # ─── Logging ─────────────────────────────────────────────────────────────────── logging.basicConfig( level=logging.DEBUG, format="%(asctime)s [%(levelname)s] %(name)s: %(message)s", handlers=[logging.StreamHandler(sys.stdout)], ) log = logging.getLogger("viral_bot") # Also set python-telegram-bot logger to INFO to see polling events logging.getLogger("telegram").setLevel(logging.INFO) logging.getLogger("httpx").setLevel(logging.WARNING) # ─── Config ──────────────────────────────────────────────────────────────────── load_dotenv() BOT_TOKEN = os.environ["TELEGRAM_BOT_TOKEN"] CHAT_ID = int(os.environ["TELEGRAM_CHAT_ID"]) NEXTCLOUD_BASE = "/home/ren/nextcloud/html/data/renato97/files/viral_clips" # The venv that has all pipeline deps (scenedetect, mediapipe, etc.) VENV_PYTHON = "/home/ren/viral/venv/bin/python3" # The yt-dl wrapper script and its own venv (has yt-dlp-ejs + Node.js solver) YT_DL_SCRIPT = "/home/ren/yt/yt-dl" YT_DL_PYTHON = "/home/ren/yt/venv/bin/python" YT_COOKIES = "/home/ren/yt/cookies.txt" SCRIPT_DIR = Path(__file__).parent MAIN_PY = str(SCRIPT_DIR / "main.py") log.info("=" * 60) log.info("Viral Telegram Bot starting up") log.info(f" CHAT_ID={CHAT_ID}") log.info(f" VENV_PYTHON={VENV_PYTHON}") log.info(f" YT_DL_PYTHON={YT_DL_PYTHON}") log.info(f" YT_DL_SCRIPT={YT_DL_SCRIPT}") log.info(f" YT_COOKIES={YT_COOKIES} (exists={os.path.exists(YT_COOKIES)})") log.info(f" MAIN_PY={MAIN_PY} (exists={os.path.exists(MAIN_PY)})") log.info(f" NEXTCLOUD_BASE={NEXTCLOUD_BASE}") log.info("=" * 60) # ─── State ───────────────────────────────────────────────────────────────────── queue: asyncio.Queue = asyncio.Queue() current_job: str | None = None # ─── Handlers ────────────────────────────────────────────────────────────────── async def start(update: Update, context: ContextTypes.DEFAULT_TYPE): chat_id = update.effective_chat.id log.info(f"/start from chat_id={chat_id}") if chat_id != CHAT_ID: log.warning(f" Ignoring unauthorized chat_id={chat_id}") return await update.message.reply_text( "🎬 *Viral Clip Bot*\n\n" "Mandame un link de YouTube y lo proceso automáticamente:\n" "• Descarga en máxima calidad\n" "• Detecta clips virales con IA\n" "• Recorte vertical 9:16 con seguimiento de cara\n" "• Subtítulos animados con highlight rosa\n" "• Clips suben solos a Nextcloud\n\n" "Comandos:\n" "• /status — qué está procesando\n" "• /help — esta ayuda", parse_mode="Markdown", ) async def help_cmd(update: Update, context: ContextTypes.DEFAULT_TYPE): await start(update, context) async def status(update: Update, context: ContextTypes.DEFAULT_TYPE): chat_id = update.effective_chat.id log.info(f"/status from chat_id={chat_id}") if chat_id != CHAT_ID: return qsize = queue.qsize() if current_job: name = current_job[:80] await update.message.reply_text( f"▶️ *Procesando:* `{name}`\n⏳ En cola: {qsize} más", parse_mode="Markdown", ) elif qsize > 0: await update.message.reply_text(f"⏳ {qsize} videos en cola, iniciando pronto...") else: await update.message.reply_text("💤 Sin actividad. Mandame un link de YouTube, Twitch o Kick.") async def handle_message(update: Update, context: ContextTypes.DEFAULT_TYPE): chat_id = update.effective_chat.id text = update.message.text.strip() if update.message.text else "" log.info(f"Message from chat_id={chat_id}: {text[:100]}") if chat_id != CHAT_ID: log.warning(f" Ignoring unauthorized chat_id={chat_id}") return if not re.search(r"(youtube\.com|youtu\.be|twitch\.tv|kick\.com)", text): log.info(" Not a YouTube, Twitch or Kick URL, ignoring.") await update.message.reply_text("❌ Mandame un link de YouTube, Twitch o Kick.") return qsize = queue.qsize() log.info(f" Enqueueing URL. Current queue size: {qsize}, current_job={current_job}") await queue.put(text) if qsize == 0 and not current_job: await update.message.reply_text("🚀 URL recibida, iniciando procesamiento...") else: await update.message.reply_text( f"⏳ URL encolada — posición {qsize + 1} en cola.\n" f"Actualmente procesando: `{(current_job or '')[:60]}`", parse_mode="Markdown", ) # ─── Title fetching ──────────────────────────────────────────────────────────── async def get_video_title(url: str) -> str: """Fetch the video title using yt-dlp (info-only, no download).""" log.info(f"[title] Fetching title for: {url[:80]}") # Use the yt venv's yt-dlp to extract info only (fast, no download) try: result = await asyncio.to_thread( subprocess.run, [ YT_DL_PYTHON, "-c", f"import yt_dlp, sys; " f"opts = {{" f" 'quiet': True, 'no_warnings': True," f" 'cookiesfrombrowser': ('chrome',)," f" 'js_runtimes': {{'node': {{}}}}," f" 'extractor_args': {{'youtube': {{'player_client': ['tv', 'web']}}}}," f" 'format': 'worst[height<=360]+worstaudio/worst[height<=360]'," f"}}; " f"ydl = yt_dlp.YoutubeDL(opts); " f"info = ydl.extract_info(sys.argv[1], download=False); " f"print(info.get('title', '') if info else '')", url, ], capture_output=True, text=True, timeout=45, ) log.debug(f"[title] stdout: {result.stdout[:200]}") log.debug(f"[title] stderr: {result.stderr[:200]}") title = result.stdout.strip() if title: log.info(f"[title] Got title: {title!r}") return title else: log.warning(f"[title] No title returned (rc={result.returncode})") except Exception as e: log.warning(f"[title] Exception fetching title: {e}") log.warning("[title] Could not get title, using 'video'") return "video" # ─── Queue processor ────────────────────────────────────────────────────────── async def process_queue(bot): global current_job log.info("[queue] Queue processor started.") while True: log.debug("[queue] Waiting for next item in queue...") url = await queue.get() log.info(f"[queue] Dequeued URL: {url[:80]}") current_job = url try: await process_video(bot, url) except Exception as e: log.error(f"[queue] Unhandled exception for url={url[:60]}: {e}", exc_info=True) try: await bot.send_message( CHAT_ID, f"❌ *Error inesperado* procesando:\n`{url[:80]}`\n\n`{str(e)[:400]}`", parse_mode="Markdown", ) except Exception as send_err: log.error(f"[queue] Failed to send error message: {send_err}") finally: current_job = None queue.task_done() log.info(f"[queue] Job done. Queue size now: {queue.qsize()}") def get_video_id(url: str) -> str: """Extract unique video ID from YouTube, Twitch, or Kick URL.""" # 1. YouTube yt_match = re.search(r"(?:v=|\/shorts\/|\/embed\/|\/v\/|youtu\.be\/)([a-zA-Z0-9_-]{11})", url) if yt_match: return yt_match.group(1) # 2. Twitch VODs twitch_vod_match = re.search(r"twitch\.tv\/videos\/(\d+)", url) if twitch_vod_match: return f"twitch_vod_{twitch_vod_match.group(1)}" # 3. Twitch Clips twitch_clip_match = re.search(r"(?:clips\.twitch\.tv\/|twitch\.tv\/\w+\/clip\/)([a-zA-Z0-9_-]+)", url) if twitch_clip_match: return f"twitch_clip_{twitch_clip_match.group(1)}" # 4. Kick VODs (contains UUID) kick_vod_match = re.search(r"kick\.com\/(?:video\/|(?:\w+\/videos\/))([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})", url) if kick_vod_match: return f"kick_vod_{kick_vod_match.group(1)}" # 5. Kick Clips kick_clip_match = re.search(r"kick\.com\/\w+\/clips\/([a-zA-Z0-9_-]+)", url) if kick_clip_match: return f"kick_clip_{kick_clip_match.group(1)}" # Fallback to md5 hash of URL to ensure uniqueness import hashlib url_hash = hashlib.md5(url.encode('utf-8')).hexdigest()[:10] return f"video_{url_hash}" # ─── Main video processor ───────────────────────────────────────────────────── async def process_video(bot, url: str): log.info(f"[process] ===== Starting process_video =====") log.info(f"[process] URL: {url}") # 1. Extract video ID and Get title yt_id = get_video_id(url) title = await get_video_title(url) log.info(f"[process] Video ID: {yt_id}, Title: {title!r}") # Use title if available, else fall back to YouTube ID so folders are always unique safe_title = re.sub(r'[<>:"/\\|?*#\s]+', "_", title)[:60].strip("_") if title and title != "video" else "" title_folder = f"{safe_title}_{yt_id}" if safe_title else yt_id date_folder = datetime.now().strftime("%Y-%m-%d") # Target folder: viral_clips/YYYY-MM-DD/video_title_ytid/ folder = f"{date_folder}/{title_folder}" target = f"{NEXTCLOUD_BASE}/{folder}" os.makedirs(target, exist_ok=True) log.info(f"[process] Nextcloud target folder: {target}") # 2. Send initial Telegram message msg = await bot.send_message( CHAT_ID, f"🎬 *Procesando:* `{title[:60] or url[:60]}`\n\n" "🔄 Iniciando pipeline...", parse_mode="Markdown", ) log.info(f"[process] Sent initial Telegram message (id={msg.message_id})") # 3. Set up environment for main.py # Use a temp dir for intermediate files — main.py will copy finished clips # to NEXTCLOUD_TARGET and run docker files:scan automatically. temp_output = tempfile.mkdtemp(prefix="viral_bot_") env = os.environ.copy() env["NEXTCLOUD_TARGET"] = target env["PYTHONUNBUFFERED"] = "1" # CRITICAL: Remove PYTHONPATH so the system protobuf doesn't conflict with # the venv's protobuf used by mediapipe → fixes 'SymbolDatabase.GetPrototype' crash env.pop("PYTHONPATH", None) log.info(f"[process] Temp output dir: {temp_output}") log.info(f"[process] PYTHONPATH cleared from subprocess env (mediapipe safety)") # 4. Build command — main.py MUST run with the viral venv Python cmd = [ VENV_PYTHON, "-u", MAIN_PY, "-u", url, "-o", temp_output, ] log.info(f"[process] Spawning command: {' '.join(cmd)}") total_clips = 0 cur_clip = 0 last_text = "" completed_clips = [] async def edit_msg(new_text: str): nonlocal last_text if new_text == last_text: return last_text = new_text try: await msg.edit_text(new_text, parse_mode="Markdown") log.debug(f"[process] Telegram msg updated: {new_text[:80]}") except Exception as e: log.warning(f"[process] Failed to edit Telegram message: {e}") try: proc = await asyncio.create_subprocess_exec( *cmd, env=env, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.STDOUT, ) log.info(f"[process] Subprocess started, PID={proc.pid}") async for raw in proc.stdout: line = raw.decode("utf-8", errors="replace").rstrip() if not line: continue # Log every line from the subprocess log.debug(f"[main.py] {line[:200]}") # Parse progress markers from main.py output if "Downloading video" in line: await edit_msg( f"🎬 *{title[:50] or 'Video'}*\n\n" f"📥 Descargando..." ) elif "Transcribing video" in line: await edit_msg( f"🎬 *{title[:50] or 'Video'}*\n\n" f"🎙️ Transcribiendo audio..." ) elif "Calling local LLM" in line or "Calling LLM" in line: await edit_msg( f"🎬 *{title[:50] or 'Video'}*\n\n" f"🤖 Analizando clips virales con IA..." ) elif m := re.search(r"Found\s+(\d+)\s+viral", line, re.IGNORECASE): total_clips = int(m.group(1)) log.info(f"[process] Detected {total_clips} viral clips") await edit_msg( f"🎬 *{title[:50] or 'Video'}*\n\n" f"🔥 Encontrados *{total_clips} clips virales*\n" f"Iniciando recorte y subtítulos..." ) elif m := re.search(r"\[Clip\s+(\d+)\]\s+Starting", line): cur_clip = int(m.group(1)) total_str = f"/{total_clips}" if total_clips else "" log.info(f"[process] Processing clip {cur_clip}{total_str}") await edit_msg( f"🎬 *{title[:50] or 'Video'}*\n\n" f"✂️ Clip {cur_clip}{total_str}: recortando en vertical + tracking de cara..." ) elif re.search(r"\[Clip\s+\d+\]\s+Burning subtitles", line): total_str = f"/{total_clips}" if total_clips else "" await edit_msg( f"🎬 *{title[:50] or 'Video'}*\n\n" f"🔥 Clip {cur_clip}{total_str}: quemando subtítulos..." ) elif m := re.search(r"\[Clip\s+(\d+)\]\s+Copying to Nextcloud", line): n = int(m.group(1)) total_str = f"/{total_clips}" if total_clips else "" log.info(f"[process] Clip {n} uploading to Nextcloud") await edit_msg( f"🎬 *{title[:50] or 'Video'}*\n\n" f"📤 Clip {n}{total_str}: subiendo a Nextcloud..." ) elif m := re.search(r"\[Clip\s+(\d+)\]\s+Nextcloud sync complete", line): n = int(m.group(1)) completed_clips.append(n) log.info(f"[process] Clip {n} synced to Nextcloud ✅") total_str = f"/{total_clips}" if total_clips else "" await edit_msg( f"🎬 *{title[:50] or 'Video'}*\n\n" f"✅ Clip {n}{total_str} en Nextcloud!\n" f"Completados: {len(completed_clips)}{total_str}" ) elif "Total execution time" in line: log.info(f"[process] Pipeline complete: {line}") rc = await proc.wait() log.info(f"[process] main.py exited with return code: {rc}") except Exception as exc: log.error(f"[process] Exception while running main.py: {exc}", exc_info=True) await bot.send_message( CHAT_ID, f"❌ *Error ejecutando main.py:*\n`{str(exc)[:400]}`", parse_mode="Markdown", ) raise # 5. Cleanup temp dir try: shutil.rmtree(temp_output, ignore_errors=True) log.info(f"[process] Cleaned up temp dir: {temp_output}") except Exception as e: log.warning(f"[process] Failed to clean temp dir: {e}") # 6. Always force a Nextcloud scan so files appear regardless of what main.py did log.info("[process] Forcing Nextcloud files:scan...") try: scan_result = await asyncio.to_thread( subprocess.run, ["docker", "exec", "-u", "www-data", "nextcloud", "php", "occ", "files:scan", "renato97"], capture_output=True, text=True, timeout=120, ) log.info(f"[process] Nextcloud scan done (rc={scan_result.returncode})") log.debug(f"[process] scan stdout: {scan_result.stdout[-300:]}") except Exception as e: log.warning(f"[process] Nextcloud scan failed: {e}") # 7. Count clips — use log-tracked list as primary (exact count for THIS run), # fall back to disk count only if parser got 0 (e.g. regex didn't match) total_done = len(completed_clips) if total_done == 0: try: disk_clips = [f for f in os.listdir(target) if f.endswith(".mp4") and not f.startswith("temp_")] total_done = len(disk_clips) log.info(f"[process] Fallback disk count: {total_done} clips in {target}") except Exception as e: log.warning(f"[process] Could not count disk clips: {e}") log.info(f"[process] Final clip count for THIS run: {total_done} (from completed_clips={len(completed_clips)})") # 8. Final completion message if rc == 0 and total_done > 0: log.info(f"[process] SUCCESS — {total_done} clips in Nextcloud") try: await msg.delete() except Exception: pass await bot.send_message( CHAT_ID, f"✅ *¡Listo!* `{title[:60] or url[:60]}`\n\n" f"📱 *{total_done} clip{'s' if total_done != 1 else ''}* en Nextcloud:\n" f"`{folder}/`\n\n" f"Abrilos en tu Nextcloud 🎉", parse_mode="Markdown", ) elif rc != 0: log.error(f"[process] FAILED — main.py returned code {rc}") await edit_msg( f"❌ *Falló el procesamiento* (código {rc})\n" f"`{url[:80]}`\n\n" f"Revisá los logs: `sudo journalctl -u viral-telegram.service -n 100`" ) else: log.warning("[process] rc=0 but no final clips found in Nextcloud folder") await edit_msg( f"⚠️ *Pipeline completado pero no se encontraron clips*\n" f"Revisá la carpeta `{folder}/` en Nextcloud manualmente." ) # ─── App lifecycle ───────────────────────────────────────────────────────────── async def post_init(app: Application): log.info("[init] post_init called — creating queue processor task") asyncio.create_task(process_queue(app.bot)) def main(): log.info("[main] Building Telegram Application...") app = ( Application.builder() .token(BOT_TOKEN) .post_init(post_init) .build() ) app.add_handler(CommandHandler("start", start)) app.add_handler(CommandHandler("help", help_cmd)) app.add_handler(CommandHandler("status", status)) app.add_handler(MessageHandler(filters.TEXT & ~filters.COMMAND, handle_message)) log.info("[main] 🤖 Viral Telegram Bot iniciado. Polling...") print("🤖 Viral Telegram Bot iniciado.") app.run_polling(allowed_updates=Update.ALL_TYPES) if __name__ == "__main__": main()