pipeline/seedance.py

"""Seedance B-roll via Volcengine Ark (SPEC_NEW §15.3).

Image-to-video from the product image + a Claude-written prompt (§15.4). The owner
uses Volcengine Ark; the provider is still env-switchable. DRY_RUN copies a fixture so
the overlay pipeline is fully demoable without credentials.

Ark video API (verify the exact model id + param flags in YOUR Ark console — they
drift, like Veo's model ids, SPEC_NEW §3):
  POST {base}/contents/generations/tasks
       {model, content:[{type:text,text}, {type:image_url,image_url:{url}}]}  -> {id}
  GET  {base}/contents/generations/tasks/{id}
       -> {status: queued|running|succeeded|failed, content:{video_url}}
Seedance generation params ride in the text as flags: --resolution 720p --ratio 9:16
--duration {5|10} --watermark false.
"""

from __future__ import annotations

import base64
import shutil
import time
import uuid
from pathlib import Path

import httpx

from . import db
from .config import FIXTURES_DIR, MEDIA_DIR, get_settings

BROLL_DIR = MEDIA_DIR / "broll"
DRY_RUN_OP_PREFIX = "dry-run-broll/"
FIXTURE_CLIP = FIXTURES_DIR / "good_clip.mp4"  # placeholder until a real B-roll fixture exists
POLL_INTERVAL_S = 10
POLL_TIMEOUT_S = 12 * 60


class SeedanceNotConfigured(RuntimeError):
    pass


class SeedanceError(RuntimeError):
    pass


def _image_url_value(image: str | None) -> str | None:
    """Ark wants a URL or a base64 data URI for the product image."""
    if not image:
        return None
    if image.startswith(("http://", "https://")):
        return image
    p = Path(image)
    if not p.is_file():
        return None
    mime = {".png": "image/png", ".webp": "image/webp"}.get(p.suffix.lower(), "image/jpeg")
    return f"data:{mime};base64,{base64.b64encode(p.read_bytes()).decode()}"


def _ark_headers() -> dict[str, str]:
    s = get_settings()
    return {"Authorization": f"Bearer {s.require('SEEDANCE_API_KEY')}", "content-type": "application/json"}


def start_broll(broll_id: str, prompt: str, product_image: str | None, duration_s: int) -> str:
    s = get_settings()
    BROLL_DIR.mkdir(parents=True, exist_ok=True)
    db.execute("update broll set status = 'generating' where id = %s", (broll_id,))

    if s.dry_run:
        op = f"{DRY_RUN_OP_PREFIX}{uuid.uuid4()}"
        db.execute("update broll set seedance_operation = %s where id = %s", (op, broll_id))
        return op

    provider = (s.seedance_provider or "").lower()
    if provider != "volcengine":
        raise SeedanceNotConfigured(
            f"only the Volcengine provider is wired (SEEDANCE_PROVIDER={provider!r}). Set it to "
            f"'volcengine' + SEEDANCE_API_KEY, or ask to wire another provider (SPEC_NEW §15.8)."
        )
    dur = 5 if (duration_s or 6) <= 5 else 10  # Seedance lite supports 5 s / 10 s; overlay loops/trims
    text = f"{prompt.strip()} --resolution 720p --ratio 9:16 --duration {dur} --watermark false"
    content: list[dict] = [{"type": "text", "text": text}]
    img = _image_url_value(product_image)
    if img:
        content.append({"type": "image_url", "image_url": {"url": img}})

    body = {"model": s.require("SEEDANCE_MODEL"), "content": content}
    r = httpx.post(f"{s.seedance_base_url}/contents/generations/tasks",
                   headers=_ark_headers(), json=body, timeout=120)
    if r.status_code >= 400:
        raise SeedanceError(f"Ark create {r.status_code}: {r.text[:400]}")
    op = r.json().get("id")
    if not op:
        raise SeedanceError(f"Ark create returned no task id: {r.text[:300]}")
    db.execute("update broll set seedance_operation = %s where id = %s", (op, broll_id))
    return op


def poll(op: str, broll_id: str) -> dict:
    s = get_settings()
    if op.startswith(DRY_RUN_OP_PREFIX):
        dest = BROLL_DIR / f"{broll_id}.mp4"
        dest.parent.mkdir(parents=True, exist_ok=True)
        shutil.copyfile(FIXTURE_CLIP, dest)
        return {"status": "done", "file": str(dest)}

    r = httpx.get(f"{s.seedance_base_url}/contents/generations/tasks/{op}",
                  headers=_ark_headers(), timeout=60)
    if r.status_code >= 400:
        raise SeedanceError(f"Ark poll {r.status_code}: {r.text[:300]}")
    data = r.json()
    status = (data.get("status") or "").lower()
    if status in ("queued", "running", "pending", "processing", ""):
        return {"status": "running"}
    if status not in ("succeeded", "success", "done"):
        raise SeedanceError(f"Ark task {status}: {data.get('error') or str(data)[:300]}")
    url = (data.get("content") or {}).get("video_url") or data.get("video_url")
    if not url:
        raise SeedanceError(f"Ark succeeded but no video_url: {str(data)[:300]}")
    vid = httpx.get(url, timeout=300)
    if vid.status_code >= 400:
        raise SeedanceError(f"Ark video download {vid.status_code}")
    dest = BROLL_DIR / f"{broll_id}.mp4"
    dest.parent.mkdir(parents=True, exist_ok=True)
    dest.write_bytes(vid.content)
    return {"status": "done", "file": str(dest)}


def generate_and_download(broll_id: str) -> str:
    """Submit + wait + download one B-roll clip; set broll.file_path + status='ready'."""
    row = db.fetch_one(
        "select b.*, p.image_path from broll b left join products p on p.id = b.product_id where b.id = %s",
        (broll_id,),
    )
    if row is None:
        raise RuntimeError(f"no broll {broll_id}")
    op = start_broll(broll_id, row["prompt"], row.get("image_path"), int(row.get("duration_s") or 6))
    deadline = time.monotonic() + POLL_TIMEOUT_S
    while True:
        res = poll(op, broll_id)
        if res["status"] == "done":
            break
        if time.monotonic() > deadline:
            raise SeedanceError(f"Seedance task {op} still running after {POLL_TIMEOUT_S}s")
        time.sleep(POLL_INTERVAL_S)
    db.execute("update broll set file_path = %s, status = 'ready' where id = %s", (res["file"], broll_id))
    return res["file"]