viral/openshorts/telegram_bot.py

"""
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.")


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)", text):
        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("🚀 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"ydl = yt_dlp.YoutubeDL({{'quiet': True, 'no_warnings': True, 'cookiesfrombrowser': ('chrome',)}}); "
                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()}")


# ─── Main video processor ─────────────────────────────────────────────────────
async def process_video(bot, url: str):
    log.info(f"[process] ===== Starting process_video =====")
    log.info(f"[process] URL: {url}")

    # 1. Get title
    title = await get_video_title(url)
    log.info(f"[process] Video title: {title!r}")

    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)
    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 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))
                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 final clips in target folder
    try:
        final_clips = [f for f in os.listdir(target) if f.endswith(".mp4") and not f.startswith("temp_")]
        total_done = len(final_clips)
        log.info(f"[process] Final clip count in Nextcloud folder: {total_done}")
    except Exception:
        total_done = 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()