bot.py

#!/usr/bin/env python3
"""
MoviePluz Pipeline Bot — HuggingFace Space
/start → fetch movie from API → run full pipeline directly on this Space

Fixes:
  - Keep-alive thread: pings self every 4 min so HF Space never sleeps mid-run
  - Realtime logs via /logs endpoint (last N lines) + /logs_stream SSE
  - /status command shows detailed progress from the live status message
  - Pipeline crash recovery with full traceback sent to log channel
"""

import os
import sys
import threading
import logging
import traceback
import time
import collections
import requests
import config as _cfg
from flask import Flask, request, jsonify, Response, stream_with_context

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

app = Flask(__name__)

BOT_TOKEN    = _cfg.BOT_TOKEN
ALLOWED_IDS  = os.environ.get("ALLOWED_USER_IDS", "")

# ── Pipeline state ─────────────────────────────────────────────────────────────
_pipeline_running = False
_pipeline_lock    = threading.Lock()
_pipeline_start   = None   # epoch time when last run started

# ── In-memory log ring buffer (last 200 lines) ────────────────────────────────
_log_buffer    = collections.deque(maxlen=200)
_log_lock      = threading.Lock()
_log_callbacks = []   # SSE subscribers

def _log(line: str):
    ts = time.strftime("%H:%M:%S")
    entry = f"[{ts}] {line}"
    with _log_lock:
        _log_buffer.append(entry)
        for cb in list(_log_callbacks):
            try: cb(entry)
            except Exception: pass
    logger.info(line)

# Patch print so pipeline prints also go to buffer
_orig_print = print
def _patched_print(*args, **kwargs):
    msg = " ".join(str(a) for a in args)
    _log(msg)
    _orig_print(*args, **kwargs)

import builtins
builtins.print = _patched_print


# ── Keep-alive: ping this Space's own health endpoint every 4 minutes ─────────
def _keep_alive():
    url = "https://mrpoddaa-burn.hf.space/"
    while True:
        time.sleep(240)  # every 4 minutes
        with _pipeline_lock:
            running = _pipeline_running
        if running:
            try:
                requests.get(url, timeout=10)
                _log("💓 Keep-alive ping sent")
            except Exception as e:
                _log(f"⚠️  Keep-alive ping failed: {e}")

threading.Thread(target=_keep_alive, daemon=True).start()


# ── Telegram helpers ───────────────────────────────────────────────────────────
def allowed_user(user_id: int) -> bool:
    if not ALLOWED_IDS.strip():
        return True
    return str(user_id) in [x.strip() for x in ALLOWED_IDS.split(",")]


def tg_send(chat_id, text, parse_mode="HTML"):
    try:
        requests.post(
            f"https://api.telegram.org/bot{BOT_TOKEN}/sendMessage",
            json={"chat_id": chat_id, "text": text, "parse_mode": parse_mode},
            timeout=10,
        )
    except Exception as e:
        logger.error(f"tg_send error: {e}")


# ── Pipeline thread ────────────────────────────────────────────────────────────
def run_pipeline_thread():
    global _pipeline_running, _pipeline_start
    _log("🚀 Pipeline thread started")
    try:
        import pipeline
        pipeline.main()
        _log("✅ Pipeline finished successfully")
    except SystemExit as e:
        _log(f"Pipeline exited (code {e.code})")
    except Exception as e:
        tb = traceback.format_exc()
        _log(f"❌ Pipeline CRASHED: {e}\n{tb}")
        try:
            import pipeline as _p
            _p.tg_log_error(f"Pipeline crashed:\n{tb[:3000]}")
        except Exception:
            pass
    finally:
        with _pipeline_lock:
            _pipeline_running = False
        _log("🏁 Pipeline thread finished — idle")


# ── Webhook ────────────────────────────────────────────────────────────────────
@app.route("/webhook", methods=["POST"])
def webhook():
    global _pipeline_running, _pipeline_start

    data = request.get_json(silent=True)
    if not data:
        return jsonify({"ok": False}), 400

    message = data.get("message") or data.get("edited_message")
    if not message:
        return jsonify({"ok": True})

    chat_id = message["chat"]["id"]
    user_id = message.get("from", {}).get("id", 0)
    text    = message.get("text", "").strip()

    # ── /start ─────────────────────────────────────────────────────────────────
    if text == "/start":
        if not allowed_user(user_id):
            tg_send(chat_id, "⛔ You are not authorised to use this bot.")
            return jsonify({"ok": True})

        with _pipeline_lock:
            if _pipeline_running:
                elapsed = int(time.time() - (_pipeline_start or time.time()))
                m, s = divmod(elapsed, 60)
                tg_send(chat_id,
                    f"⚠️ <b>Pipeline already running!</b>\n"
                    f"⏱ Running for {m}m {s}s\n"
                    f"Send /status for details or /logs for live output.",
                    parse_mode="HTML"
                )
                return jsonify({"ok": True})
            _pipeline_running = True
            _pipeline_start   = time.time()

        with _log_lock:
            _log_buffer.clear()

        tg_send(chat_id,
            "🚀 <b>Pipeline started!</b>\n\n"
            "Fetching movie details and beginning encode...\n"
            "📋 Check the log channel for live progress.\n"
            "🔍 Use /logs for live output here.",
            parse_mode="HTML"
        )

        t = threading.Thread(target=run_pipeline_thread, daemon=True)
        t.start()
        _log(f"/start triggered by user {user_id}")

    # ── /status ────────────────────────────────────────────────────────────────
    elif text == "/status":
        with _pipeline_lock:
            running = _pipeline_running
            start   = _pipeline_start

        if running and start:
            elapsed = int(time.time() - start)
            m, s = divmod(elapsed, 60)
            h, m = divmod(m, 60)
            if h:
                dur = f"{h}h {m}m {s}s"
            else:
                dur = f"{m}m {s}s"

            # Last 5 log lines as preview
            with _log_lock:
                last_lines = list(_log_buffer)[-5:]
            preview = "\n".join(f"  {l}" for l in last_lines) or "  (no logs yet)"

            tg_send(chat_id,
                f"⚙️ <b>Pipeline RUNNING</b>\n"
                f"⏱ Running for: <code>{dur}</code>\n\n"
                f"📋 <b>Recent activity:</b>\n<code>{preview}</code>\n\n"
                f"Use /logs for full output.",
                parse_mode="HTML"
            )
        else:
            tg_send(chat_id,
                "💤 <b>Pipeline is idle.</b>\n"
                "Send /start to trigger.",
                parse_mode="HTML"
            )

    # ── /logs ──────────────────────────────────────────────────────────────────
    elif text == "/logs":
        with _log_lock:
            lines = list(_log_buffer)[-30:]

        if not lines:
            tg_send(chat_id, "📋 No logs yet. Start the pipeline with /start.")
        else:
            chunk = "\n".join(lines)
            tg_send(chat_id,
                f"📋 <b>Last {len(lines)} log lines:</b>\n<code>{chunk[:4000]}</code>",
                parse_mode="HTML"
            )

    # ── /help ──────────────────────────────────────────────────────────────────
    elif text == "/help":
        tg_send(chat_id,
            "🤖 <b>MoviePluz Pipeline Bot</b>\n\n"
            "/start   — Fetch next movie &amp; run full pipeline\n"
            "/status  — Check running status + recent activity\n"
            "/logs    — Show last 30 log lines\n"
            "/help    — Show this message",
            parse_mode="HTML"
        )

    return jsonify({"ok": True})


# ── Health check ───────────────────────────────────────────────────────────────
@app.route("/", methods=["GET"])
def health():
    with _pipeline_lock:
        running = _pipeline_running
        start   = _pipeline_start
    elapsed = int(time.time() - start) if (running and start) else 0
    return jsonify({
        "status":           "ok",
        "pipeline_running": running,
        "elapsed_seconds":  elapsed,
    })


# ── /logs endpoint (HTTP — last N lines) ──────────────────────────────────────
@app.route("/logs")
def logs_http():
    n = min(int(request.args.get("n", 50)), 200)
    with _log_lock:
        lines = list(_log_buffer)[-n:]
    return jsonify({"lines": lines, "count": len(lines)})


# ── /logs_stream — SSE realtime streaming ─────────────────────────────────────
@app.route("/logs_stream")
def logs_stream():
    """
    Server-Sent Events endpoint.
    Connect with:  curl -N https://YOUR-SPACE.hf.space/logs_stream
    Or open in browser for live tail.
    """
    q = []

    def _cb(line):
        q.append(line)

    with _log_lock:
        # Send existing buffer first
        existing = list(_log_buffer)
        _log_callbacks.append(_cb)

    def generate():
        # Replay existing lines
        for line in existing:
            yield f"data: {line}\n\n"
        # Stream new lines
        while True:
            if q:
                line = q.pop(0)
                yield f"data: {line}\n\n"
            else:
                yield ": heartbeat\n\n"
                time.sleep(2)

    def cleanup():
        try:
            with _log_lock:
                _log_callbacks.remove(_cb)
        except ValueError:
            pass

    response = Response(
        stream_with_context(generate()),
        mimetype="text/event-stream"
    )
    response.headers["Cache-Control"] = "no-cache"
    response.headers["X-Accel-Buffering"] = "no"
    return response


# ── Webhook registration ───────────────────────────────────────────────────────
@app.route("/set_webhook", methods=["GET"])
def set_webhook():
    webhook_url = request.args.get("url", "")
    if not webhook_url:
        return jsonify({"error": "Pass ?url=https://your-space.hf.space/webhook"}), 400
    r = requests.post(
        f"https://api.telegram.org/bot{BOT_TOKEN}/setWebhook",
        json={"url": webhook_url, "allowed_updates": ["message"]},
        timeout=10,
    )
    return jsonify(r.json())


if __name__ == "__main__":
    port = int(os.environ.get("PORT", 7860))
    app.run(host="0.0.0.0", port=port)