401 lines
16 KiB
Python
401 lines
16 KiB
Python
"""
|
|
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, '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:
|
|
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
|
|
env = os.environ.copy()
|
|
env["NEXTCLOUD_TARGET"] = target
|
|
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:
|
|
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. 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)
|
|
.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()
|