diff --git a/openshorts/telegram_bot.py b/openshorts/telegram_bot.py index 3f88712..a9a15d0 100644 --- a/openshorts/telegram_bot.py +++ b/openshorts/telegram_bot.py @@ -1,9 +1,18 @@ """ Telegram bot for viral clip processing. -Receives YouTube URLs, processes via main.py, sends progress updates. +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 @@ -17,187 +26,360 @@ 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" -VENV_PYTHON = "/home/ren/viral/venv/bin/python3" -COOKIES = "/home/ren/yt/cookies.txt" -SCRIPT_DIR = Path(__file__).parent -queue = asyncio.Queue() +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): - if update.effective_chat.id != CHAT_ID: + 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( - "🎬 Mandame un link de YouTube y lo proceso.\n" - "Los clips aparecen en Nextcloud automáticamente." + "🎬 *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): - if update.effective_chat.id != CHAT_ID: + 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}") + 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"⏳ En cola: {qsize}") + await update.message.reply_text(f"⏳ {qsize} videos en cola, iniciando pronto...") else: - await update.message.reply_text("💤 Sin actividad.") + await update.message.reply_text("💤 Sin actividad. Mandame un link de YouTube.") async def handle_message(update: Update, context: ContextTypes.DEFAULT_TYPE): - if update.effective_chat.id != CHAT_ID: + 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 - text = update.message.text.strip() + if not re.search(r"(youtube\.com|youtu\.be)", text): - await update.message.reply_text("❌ Mandame un link de YouTube.") + log.info(" Not a YouTube URL, ignoring.") + await update.message.reply_text("❌ Mandame un link de YouTube (youtube.com o youtu.be).") 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("🚀 Iniciando procesamiento...") + await update.message.reply_text("🚀 URL recibida, iniciando procesamiento...") else: - await update.message.reply_text(f"⏳ Encolado ({qsize + 1}º en cola).") - - -async def get_video_title(url: str) -> str: - try: - r = await asyncio.to_thread( - subprocess.run, - ["yt-dlp", "--print", "title", url], - capture_output=True, text=True, timeout=30, + 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", ) - return r.stdout.strip() - except subprocess.TimeoutExpired: - print("[DEBUG] yt-dlp title timed out after 30s") - return "video" - except FileNotFoundError: - print("[DEBUG] yt-dlp command not found. Trying venv yt-dlp...") - try: - r = await asyncio.to_thread( - subprocess.run, - [VENV_PYTHON, "-m", "yt_dlp", "--cookies", COOKIES, "--print", "title", url], - capture_output=True, text=True, timeout=30, - ) - return r.stdout.strip() or "video" - except Exception as e2: - print(f"[DEBUG] venv yt-dlp also failed: {e2}") - return "video" + + +# ─── 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"ydl = yt_dlp.YoutubeDL({{'quiet': True, 'no_warnings': True, 'cookiefile': '{YT_COOKIES}'}}); " + 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: - print(f"[DEBUG] yt-dlp title failed: {e}") - return "video" + 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 procesando: {e}") - except Exception: - pass + 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()}") +# ─── Main video processor ───────────────────────────────────────────────────── async def process_video(bot, url: str): - print(f"[DEBUG] process_video: url={url[:80]}") + log.info(f"[process] ===== Starting process_video =====") + log.info(f"[process] URL: {url}") + # 1. Get title title = await get_video_title(url) - print(f"[DEBUG] title={title}") + log.info(f"[process] Video title: {title!r}") - safe = re.sub(r'[<>:"/\\|?*#\s]+', "_", title)[:80] + safe = re.sub(r'[<>:"/\\|?*#\s]+', "_", title)[:80] or "video" folder = f"{safe}_{datetime.now():%Y-%m-%d}" target = f"{NEXTCLOUD_BASE}/{folder}" os.makedirs(target, exist_ok=True) - print(f"[DEBUG] NEXTCLOUD_TARGET={target}") + log.info(f"[process] Nextcloud target folder: {target}") - msg = await bot.send_message(CHAT_ID, "🚀 Descargando...") + # 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 env = os.environ.copy() env["NEXTCLOUD_TARGET"] = target - temp_dir = tempfile.mkdtemp(prefix="viral_") + env["PYTHONUNBUFFERED"] = "1" + + # 4. Build command — main.py MUST run with the viral venv Python + cmd = [ + VENV_PYTHON, "-u", MAIN_PY, + "-u", url, + "-o", target, + ] + 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: - cmd = [ - str(SCRIPT_DIR / "main.py"), - "-u", url, - "-o", temp_dir, - ] - print(f"[DEBUG] Spawning: {' '.join(cmd)}") - proc = await asyncio.create_subprocess_exec( - VENV_PYTHON, - "-u", *cmd, + *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").strip() + line = raw.decode("utf-8", errors="replace").rstrip() if not line: continue - print(f"[DEBUG] stdout: {line[:100]}") - text = None - if "Transcribing" in line: - text = "🎙️ Transcribiendo..." - elif "Calling local LLM" in line: - text = "🤖 Analizando clips virales..." + # Log every line from the subprocess + log.debug(f"[main.py] {line[:200]}") + + # Parse progress markers from main.py output + if "Downloading video from YouTube" 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)) - text = f"🔥 Encontrados {total_clips} clips" + 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)) - text = f"🎬 Clip {cur_clip}/?: reencuadrando + subs..." - elif "Normalizing" in line and cur_clip > 0: - text = f"📱 Clip {cur_clip}/{max(total_clips, cur_clip)}: normalizando..." - elif "Copying to Nextcloud" in line and cur_clip > 0: - text = f"📤 Clip {cur_clip}/{max(total_clips, cur_clip)}: subiendo a Nextcloud..." + 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: - text = None + log.info(f"[process] Pipeline complete: {line}") - if text and text != last_text: - last_text = text - try: - await msg.edit_text(text) - except Exception: - pass + rc = await proc.wait() + log.info(f"[process] main.py exited with return code: {rc}") - await proc.wait() - print(f"[DEBUG] main.py exit code: {proc.returncode}") + 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 - finally: - shutil.rmtree(temp_dir, ignore_errors=True) - - try: - await msg.delete() - except Exception: - pass + # 5. Final completion message + if rc == 0: + total_done = len(completed_clips) + log.info(f"[process] SUCCESS — {total_done} clips uploaded to 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 ''}* subidos a Nextcloud:\n" + f"`{folder}/`\n\n" + f"Abrilos en tu Nextcloud 🎉", + parse_mode="Markdown", + ) + else: + 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`" + ) +# ─── 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) @@ -205,8 +387,11 @@ def main(): .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)