render_worker.py

#!/usr/bin/env python3
"""Black Volt render worker — the Stage 2 hybrid render bridge.

Black Volt Mobility offloads heavy video rendering to BitTrader. This is the thin,
dependency-free worker that wires the two together:

    Black Volt  ──POST /render (HMAC-signed)──►  this worker
                                                   produce_single()  (or sample)
    Black Volt  ◄──POST callback (HMAC-signed)──  finished mp4 (base64)

Both directions are authenticated with one shared secret (`BV_RENDER_SIGNING_KEY`,
HMAC-SHA256 over the raw body, header `x-bv-render-signature`). The worker renders
asynchronously and posts the finished mp4 back inline as base64 — Black Volt writes
it under its public /media mount (no file hosting needed here, no SSRF surface).

Uses only the Python standard library + ffmpeg (already a BitTrader dependency).
Real rendering goes through `agents.producer.produce_single`; if that's
unavailable or `BV_RENDER_SAMPLE=1`, it emits a short branded ffmpeg sample clip so
the whole signed round-trip is testable without the paid MiniMax video APIs.

Run:
    BV_RENDER_SIGNING_KEY=<shared-secret> [BV_RENDER_SAMPLE=1] python3 render_worker.py
    # listens on 0.0.0.0:8090 (override with BV_RENDER_PORT)
"""
from __future__ import annotations

import base64
import hashlib
import hmac
import json
import logging
import os
import subprocess
import tempfile
import threading
import time
import urllib.request
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from pathlib import Path

logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s")
log = logging.getLogger("bv.render_worker")

SIGNING_KEY = os.environ.get("BV_RENDER_SIGNING_KEY", "")
PORT = int(os.environ.get("BV_RENDER_PORT", "8090"))
SAMPLE_ONLY = os.environ.get("BV_RENDER_SAMPLE", "") in ("1", "true", "yes")
MAX_BODY = 4 * 1024 * 1024  # inbound render request bodies are tiny (script JSON)


def _sign(body: bytes) -> str:
    return base64.b64encode(
        hmac.new(SIGNING_KEY.encode("utf-8"), body, hashlib.sha256).digest()
    ).decode("ascii")


def _verify(body: bytes, signature: str) -> bool:
    if not SIGNING_KEY:
        return False
    return hmac.compare_digest(_sign(body), signature or "")


_FONT_CANDIDATES = (
    "/usr/share/fonts/truetype/dejavu/DejaVuSans-Bold.ttf",
    "/usr/share/fonts/truetype/dejavu/DejaVuSans.ttf",
    "/System/Library/Fonts/Supplemental/Arial Bold.ttf",
    "/Library/Fonts/Arial.ttf",
)


def _font_path() -> str | None:
    for p in _FONT_CANDIDATES:
        if os.path.exists(p):
            return p
    return None


def _ffmpeg_sample(title: str = "") -> bytes:
    """A short, valid, on-brand vertical mp4 — a stand-in render so the signed
    pipeline works without the paid video APIs. Draws the brand over void-black
    when a font is available; falls back to a plain color clip otherwise."""
    base = [
        "ffmpeg", "-y",
        "-f", "lavfi", "-i", "color=c=0x0A0A0F:s=1080x1920:d=3",
        "-f", "lavfi", "-i", "anullsrc=channel_layout=stereo:sample_rate=44100",
        "-shortest", "-pix_fmt", "yuv420p", "-c:v", "libx264", "-c:a", "aac",
        "-movflags", "+faststart",
    ]
    font = _font_path()
    vf = []
    if font:
        vf.append(
            f"drawtext=fontfile={font}:text='BLACK VOLT':fontcolor=0x00E5FF:"
            "fontsize=120:x=(w-text_w)/2:y=(h/2)-170"
        )
        vf.append(
            f"drawtext=fontfile={font}:text='Silent Power. Premium Arrival.':"
            "fontcolor=0xC7CBD4:fontsize=40:x=(w-text_w)/2:y=(h/2)+30"
        )
    with tempfile.TemporaryDirectory() as d:
        out = os.path.join(d, "sample.mp4")
        cmd = base + (["-vf", ",".join(vf)] if vf else []) + [out]
        try:
            subprocess.run(cmd, check=True, capture_output=True)
        except subprocess.CalledProcessError:
            # drawtext can fail on some ffmpeg builds — fall back to plain color.
            subprocess.run(base + [out], check=True, capture_output=True)
        return Path(out).read_bytes()


def _post_progress(progress_url: str, job: dict, stage: str, pct: int) -> None:
    """Best-effort: tell Black Volt the current render stage/percent via the same
    signed scheme as the final callback. Any failure is swallowed — progress is
    cosmetic and must never break or slow the render."""
    if not progress_url:
        return
    try:
        payload = {
            "job_id": job.get("job_id"),
            "tenant_id": job.get("tenant_id"),
            "post_id": job.get("post_id"),
            "stage": str(stage)[:48],
            "progress": max(0, min(100, int(pct))),
        }
        body = json.dumps(payload).encode("utf-8")
        headers = {
            "Content-Type": "application/json",
            "x-bv-render-signature": _sign(body),
            "User-Agent": (
                "Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 "
                "(KHTML, like Gecko) Chrome/124.0.0.0 Safari/537.36"
            ),
        }
        req = urllib.request.Request(progress_url, data=body, headers=headers, method="POST")
        with urllib.request.urlopen(req, timeout=10):
            pass
    except Exception as e:
        log.debug("progress post failed (%s %d): %s", stage, pct, e)


def _render_bytes(script: dict, progress=None):
    """Produce (asset_bytes, ext) for a job. Image posts (media_kind="image") render one
    Kling still (jpg/png); everything else renders the mp4. Falls back to a sample mp4."""
    is_image = str(script.get("media_kind") or "").lower() == "image"
    if not SAMPLE_ONLY:
        try:
            # Black Volt's own renderer (brand + language-aware voice + Kling mix),
            # NOT BitTrader's produce_single (which bakes in BitTrader branding).
            if is_image:
                from bv_producer import produce_blackvolt_image

                with tempfile.TemporaryDirectory() as d:
                    result = produce_blackvolt_image(script, Path(d), progress=progress)
                    if isinstance(result, dict) and result.get("status") == "ok":
                        out = result.get("output_file")
                        if out and os.path.exists(out):
                            return Path(out).read_bytes(), (result.get("ext") or "jpg")
                    log.warning("produce_blackvolt_image did not yield a file (%s) — sample", result)
            else:
                from bv_producer import produce_blackvolt

                with tempfile.TemporaryDirectory() as d:
                    result = produce_blackvolt(script, Path(d), progress=progress)
                    if isinstance(result, dict) and result.get("status") == "ok":
                        out = result.get("output_file")
                        if out and os.path.exists(out):
                            return Path(out).read_bytes(), "mp4"
                    log.warning("produce_blackvolt did not yield a file (%s) — using sample", result)
        except Exception as e:  # missing deps / API keys / render error → sample
            log.warning("producer unavailable (%s) — using sample", e)
    return _ffmpeg_sample(), "mp4"


def _process(job: dict) -> None:
    job_id = job.get("job_id")
    callback_url = job.get("callback_url")
    progress_url = job.get("progress_url")
    try:
        def _emit(stage: str, pct: int) -> None:
            _post_progress(progress_url, job, stage, pct)

        media, ext = _render_bytes(job.get("script") or {}, progress=_emit)
        payload = {
            "job_id": job_id,
            "tenant_id": job.get("tenant_id"),
            "post_id": job.get("post_id"),
            "media_b64": base64.b64encode(media).decode("ascii"),
            "media_ext": ext,
        }
        body = json.dumps(payload).encode("utf-8")
        headers = {
            "Content-Type": "application/json",
            "x-bv-render-signature": _sign(body),
            # A real browser UA so Cloudflare's bot check doesn't 1010 the
            # callback when it traverses the public edge.
            "User-Agent": (
                "Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 "
                "(KHTML, like Gecko) Chrome/124.0.0.0 Safari/537.36"
            ),
        }
        # Retry the callback POST: the mp4 upload traverses the public Cloudflare
        # edge and occasionally hits a transient TLS/network blip (e.g.
        # SSLV3_ALERT_BAD_RECORD_MAC). The render is already done; only the POST
        # is retried so a transient hiccup never strands a finished video.
        last = None
        for attempt in range(4):
            try:
                req = urllib.request.Request(
                    callback_url, data=body, headers=headers, method="POST"
                )
                with urllib.request.urlopen(req, timeout=90) as resp:
                    log.info(
                        "job %s callback %s (%d bytes %s, attempt %d)",
                        job_id, resp.status, len(media), ext, attempt + 1,
                    )
                    return
            except Exception as e:
                last = e
                log.warning(
                    "job %s callback attempt %d failed: %s", job_id, attempt + 1, e
                )
                time.sleep(3 * (attempt + 1))
        log.error("job %s callback gave up after retries: %s", job_id, last)
    except Exception as e:
        log.error("job %s failed: %s", job_id, e)


class Handler(BaseHTTPRequestHandler):
    def _json(self, code: int, obj: dict) -> None:
        self.send_response(code)
        self.send_header("Content-Type", "application/json")
        self.end_headers()
        self.wfile.write(json.dumps(obj).encode("utf-8"))

    def log_message(self, *args):  # quiet default access log
        pass

    def do_GET(self):
        if self.path == "/health":
            return self._json(200, {"ok": True, "sample_only": SAMPLE_ONLY})
        self._json(404, {"error": "not_found"})

    def do_POST(self):
        if self.path != "/render":
            return self._json(404, {"error": "not_found"})
        length = int(self.headers.get("Content-Length") or 0)
        if length <= 0 or length > MAX_BODY:
            return self._json(400, {"error": "bad_length"})
        body = self.rfile.read(length)
        if not _verify(body, self.headers.get("x-bv-render-signature", "")):
            return self._json(403, {"error": "invalid_signature"})
        try:
            job = json.loads(body)
        except Exception:
            return self._json(400, {"error": "invalid_json"})
        if not (job.get("callback_url") and job.get("post_id") and job.get("tenant_id")):
            return self._json(400, {"error": "missing_fields"})
        threading.Thread(target=_process, args=(job,), daemon=True).start()
        self._json(202, {"accepted": job.get("job_id")})


def main() -> None:
    if not SIGNING_KEY:
        raise SystemExit("BV_RENDER_SIGNING_KEY is required")
    log.info("Black Volt render worker on :%d (sample_only=%s)", PORT, SAMPLE_ONLY)
    ThreadingHTTPServer(("0.0.0.0", PORT), Handler).serve_forever()


if __name__ == "__main__":
    main()