71 lines
2.3 KiB
Python
71 lines
2.3 KiB
Python
"""② Higgsfield step — photos + prompt → completed 8s base video.
|
|
|
|
Thin wrapper around the verified CLI flow:
|
|
higgsfield generate cost/create marketing_studio_video --mode tv_spot ...
|
|
dry_run=True returns a bundled demo clip and spends no credits.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import re
|
|
import subprocess
|
|
import urllib.request
|
|
import uuid
|
|
from pathlib import Path
|
|
|
|
ENGINE = Path(__file__).resolve().parents[3] # engine/higgsfield_shorts
|
|
DEMO_VIDEO = ENGINE / "webapp" / "demo" / "mumum.mp4"
|
|
TMP = ENGINE / "server" / ".tmp"
|
|
|
|
MODEL = "marketing_studio_video"
|
|
_URL_RE = re.compile(r'"result_url"\s*:\s*"([^"]+)"')
|
|
_CREDITS_RE = re.compile(r'"credits(?:_exact)?"\s*:\s*([0-9.]+)')
|
|
|
|
|
|
def _base_args(prompt: str, image_paths: list[Path], duration: int) -> list[str]:
|
|
args = ["--prompt", prompt]
|
|
for p in image_paths:
|
|
args += ["--image", str(p)]
|
|
args += ["--mode", "tv_spot", "--duration", str(duration),
|
|
"--generate_audio", "true", "--aspect_ratio", "9:16"]
|
|
return args
|
|
|
|
|
|
def cost(prompt: str, image_paths: list[Path], duration: int = 8) -> float:
|
|
out = subprocess.run(
|
|
["higgsfield", "generate", "cost", MODEL, *_base_args(prompt, image_paths, duration), "--json"],
|
|
capture_output=True, text=True, check=True,
|
|
)
|
|
m = _CREDITS_RE.search(out.stdout)
|
|
return float(m.group(1)) if m else 0.0
|
|
|
|
|
|
def generate(
|
|
prompt: str,
|
|
image_paths: list[Path],
|
|
duration: int = 8,
|
|
dry_run: bool = False,
|
|
wait_timeout: str = "15m",
|
|
) -> tuple[Path, float]:
|
|
"""Return (local mp4 path, credits spent)."""
|
|
if dry_run:
|
|
return DEMO_VIDEO, 0.0
|
|
|
|
out = subprocess.run(
|
|
["higgsfield", "generate", "create", MODEL,
|
|
*_base_args(prompt, image_paths, duration),
|
|
"--wait", "--wait-timeout", wait_timeout, "--json"],
|
|
capture_output=True, text=True, check=True,
|
|
)
|
|
m = _URL_RE.search(out.stdout)
|
|
if not m:
|
|
raise RuntimeError(f"Higgsfield returned no result_url:\n{out.stdout[-2000:]}")
|
|
url = m.group(1)
|
|
credits_m = _CREDITS_RE.search(out.stdout)
|
|
credits = float(credits_m.group(1)) if credits_m else 0.0
|
|
|
|
TMP.mkdir(parents=True, exist_ok=True)
|
|
dest = TMP / f"base_{uuid.uuid4().hex[:8]}.mp4"
|
|
urllib.request.urlretrieve(url, dest)
|
|
return dest, credits
|