87 lines
3.2 KiB
Python
87 lines
3.2 KiB
Python
"""fal.ai 백엔드 — Seedance 2.0 reference-to-video (멀티이미지 9장, 9:16 세로).
|
|
|
|
Higgsfield platform API 가 모든 모델(dop/seedance v1/kling)을 단일 이미지로만 노출하는
|
|
한계를 우회한다. 업로드한 사진 여러 장을 그대로 넣어 ① AI 환각(안 보이는 공간 창작)을
|
|
줄이고 ② 여러 공간을 둘러보는 영상을 만든다.
|
|
|
|
흐름: 로컬 사진 → fal 업로드(URL) → subscribe(자동 폴링) → 결과 video.url → 다운로드.
|
|
인증: fal SDK 가 FAL_KEY("keyid:secret") 를 환경에서 읽음(config 에서 주입).
|
|
dry_run=True 면 호출 없이 번들 데모 클립 반환.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
import uuid
|
|
from pathlib import Path
|
|
|
|
import httpx
|
|
|
|
from app import config as cfg
|
|
|
|
ENGINE = Path(__file__).resolve().parents[3] # engine/higgsfield_shorts
|
|
TMP = ENGINE / "server" / ".tmp"
|
|
DEMO_VIDEO = cfg.WEBAPP_DIR / "demo" / "mumum.mp4"
|
|
|
|
|
|
def _download(url: str) -> Path:
|
|
# httpx + 브라우저 UA (CDN이 기본 UA를 403 차단하는 경우 대비)
|
|
TMP.mkdir(parents=True, exist_ok=True)
|
|
dest = TMP / f"base_{uuid.uuid4().hex[:8]}.mp4"
|
|
with httpx.stream("GET", url, follow_redirects=True, timeout=180.0,
|
|
headers={"User-Agent": "Mozilla/5.0"}) as r:
|
|
r.raise_for_status()
|
|
with dest.open("wb") as f:
|
|
for chunk in r.iter_bytes():
|
|
f.write(chunk)
|
|
return dest
|
|
|
|
|
|
def generate(
|
|
prompt: str,
|
|
image_paths: list[Path],
|
|
duration: int | None = None,
|
|
dry_run: bool = False,
|
|
**_,
|
|
) -> tuple[Path, float]:
|
|
"""Return (local mp4 path, credits). credits 는 fal 응답에 없어 0.0."""
|
|
if dry_run:
|
|
return DEMO_VIDEO, 0.0
|
|
|
|
try:
|
|
import fal_client
|
|
except ImportError as e:
|
|
raise RuntimeError(
|
|
"fal-client 미설치. `pip install fal-client` 또는 이미지 재빌드 필요."
|
|
) from e
|
|
|
|
if not os.environ.get("FAL_KEY"):
|
|
raise RuntimeError(
|
|
"FAL_KEY 없음. .env 에 FAL_KEY='keyid:secret'(또는 FAL_API_KEY) 설정 필요."
|
|
)
|
|
|
|
# 1) 입력 사진 업로드 → URL. Seedance 2.0 는 최대 9장.
|
|
image_urls = [fal_client.upload_file(str(p)) for p in image_paths[:cfg.FAL_MAX_IMAGES]]
|
|
if not image_urls:
|
|
raise RuntimeError("업로드된 입력 이미지가 없습니다.")
|
|
|
|
# 2) 요청 인자. duration 은 "auto" 또는 4~12 정수.
|
|
arguments: dict = {
|
|
"prompt": prompt,
|
|
"image_urls": image_urls,
|
|
"aspect_ratio": cfg.FAL_ASPECT_RATIO,
|
|
"resolution": cfg.FAL_RESOLUTION,
|
|
}
|
|
dur = duration if duration is not None else cfg.FAL_DURATION
|
|
if str(dur).lower() != "auto":
|
|
arguments["duration"] = int(dur)
|
|
|
|
# 3) 동기 호출(자동 폴링). subscribe 가 완료까지 블록.
|
|
result = fal_client.subscribe(cfg.FAL_MODEL_ID, arguments=arguments)
|
|
|
|
# 4) 결과 영상 URL 추출 → 다운로드. (응답: {"video": {"url": ...}, "seed": ...})
|
|
video = result.get("video") if isinstance(result, dict) else None
|
|
url = video.get("url") if isinstance(video, dict) else None
|
|
if not url:
|
|
raise RuntimeError(f"fal 결과에서 영상 URL을 찾지 못함: {result!r}")
|
|
return _download(url), 0.0
|