76 lines
3.5 KiB
Python
76 lines
3.5 KiB
Python
"""⑥ i2v — 포스터 + 모션 프롬프트 → 생성형 애니메이션 클립.
|
|
|
|
생성 전에 계정 가드와 저해상 거절, 생성 후에 제목 훼손 게이트가 붙는다.
|
|
"""
|
|
from pathlib import Path
|
|
|
|
import httpx
|
|
from PIL import Image
|
|
|
|
from models.detect import Regions
|
|
from models.i2v import I2vResult, TitleGateVerdict
|
|
from services.title_gate import check_title
|
|
from settings import settings
|
|
from utils.higgsfield import (KLING_3_0, assert_account, parse_credits, parse_result_url,
|
|
run_cli)
|
|
|
|
VEO_QUALITY = "high" # veo 전용
|
|
CLIP_SECONDS = 8
|
|
ASPECT_RATIO = "9:16"
|
|
MAX_CREDITS = 60.0
|
|
MIN_LONG_EDGE_PX = 1000 # 긴 변이 이보다 짧으면 확대 시 뭉개진다
|
|
MIN_CLIP_BYTES = 10_000
|
|
|
|
|
|
def reject_low_resolution(poster: Path) -> None:
|
|
"""검출·나레이션·TTS·BGM은 해상도와 무관하게 유효하므로 i2v에서만 거절한다."""
|
|
with Image.open(poster) as image:
|
|
if max(image.size) < MIN_LONG_EDGE_PX:
|
|
raise RuntimeError(
|
|
f"저해상 포스터 {image.size} — 확대 시 뭉개져 애니메이션을 건너뜁니다. "
|
|
"고해상 원본으로 다시 올려주세요.")
|
|
|
|
|
|
def cli_args(prompt: str, poster: Path, model: str) -> list[str]:
|
|
args = ["--prompt", prompt, "--start-image", str(poster),
|
|
"--duration", str(CLIP_SECONDS), "--aspect_ratio", ASPECT_RATIO]
|
|
if model == KLING_3_0:
|
|
# Kling은 9:16을 지정해도 포스터 비율로 출력한다 — 합성 단계에서 맞춘다
|
|
return args + ["--mode", "pro", "--sound", "off"]
|
|
return args + ["--quality", VEO_QUALITY]
|
|
|
|
|
|
async def estimate_credits(prompt: str, poster: Path, model: str = KLING_3_0) -> float:
|
|
return parse_credits(await run_cli(
|
|
"generate", "cost", model, *cli_args(prompt, poster, model), "--json", timeout=300))
|
|
|
|
|
|
async def animate(prompt: str, poster: Path, regions: Regions | None = None, *,
|
|
model: str = KLING_3_0,
|
|
max_credits: float = MAX_CREDITS) -> I2vResult:
|
|
if not prompt.strip():
|
|
raise ValueError("i2v 프롬프트가 비어 있다 — motion 단계를 먼저 돌릴 것")
|
|
reject_low_resolution(poster)
|
|
await assert_account(settings.higgsfield_account) # 크레딧 쓰기 전에
|
|
|
|
credits = await estimate_credits(prompt, poster, model)
|
|
if credits > max_credits:
|
|
raise RuntimeError(f"비용 {credits:g}크레딧이 상한 {max_credits:g}를 넘는다 — 설정 확인")
|
|
|
|
output = await run_cli("generate", "create", model, *cli_args(prompt, poster, model),
|
|
"--wait", "--wait-timeout", "25m", "--json")
|
|
async with httpx.AsyncClient(timeout=600, follow_redirects=True) as client:
|
|
response = await client.get(parse_result_url(output))
|
|
response.raise_for_status()
|
|
clip = response.content
|
|
if len(clip) < MIN_CLIP_BYTES:
|
|
raise RuntimeError(f"클립이 너무 작다 ({len(clip)}바이트) — 다운로드 실패로 본다")
|
|
|
|
# 영역 좌표가 없으면 제목 훼손을 잴 근거가 없다. 통과시키되 그 사실을 남긴다
|
|
verdict = (await check_title(clip, regions) if regions
|
|
else TitleGateVerdict(passed=True, skip="regions 없음 — 제목 게이트 미실행"))
|
|
if not verdict.passed:
|
|
raise RuntimeError(f"제목 훼손 감지 — {verdict.reason} "
|
|
"(재렌더하거나 모션 요소를 줄여보세요)")
|
|
return I2vResult(clip=clip, credits=credits, model=model, title_gate=verdict)
|