backend api 추가

This commit is contained in:
jaehwang 2026-09-08 13:00:23 +09:00
parent 2d4201ce9e
commit f008ed4010
15 changed files with 1501 additions and 5 deletions

View File

@ -1,3 +1,28 @@
from contextlib import asynccontextmanager
from fastapi import FastAPI
app = FastAPI(title="poster-alive")
from pipelines import worker
from routers import f1, playreel
from utils import blob
from utils.database import close_engine
@asynccontextmanager
async def lifespan(app: FastAPI):
worker.start()
yield
await worker.stop()
await blob.close_client()
await close_engine()
app = FastAPI(title="poster-alive", lifespan=lifespan)
app.include_router(f1.router)
app.include_router(playreel.router)
@app.get("/api/health")
async def health():
return {"ok": True}

View File

@ -5,6 +5,21 @@
from enum import StrEnum
def stage_keys(state_type) -> list[str]:
"""실행 순서대로 나열된 단계 이름
COMPLETED는 종료 표시라 빠짐
"""
return [member.value for member in state_type if member.name != "COMPLETED"]
def initial_timings(state_type) -> dict:
"""스테퍼가 그리는 단계별 시각의 초기값
순서가 화면에 찍히는 순서라 생성 시점에 채워 둔다
"""
return {key: {"status": "idle", "started": None, "ended": None}
for key in stage_keys(state_type)}
class PosterAliveState(StrEnum):
DETECT = "detect"
NARRATION_TEXT = "narration_text"

View File

@ -0,0 +1,47 @@
"""단계 산출물의 blob 저장·회수
경로는 <파이프라인>/<task_id>/<이름>
"""
import io
from PIL import Image
from utils import blob
JPEG_QUALITY = 90
def artifact_path(pipeline: str, task_id: str, name: str) -> str:
return f"{pipeline}/{task_id}/{name}"
async def store_bytes(pipeline: str, task_id: str, name: str, data: bytes) -> str:
return await blob.upload_bytes(data, artifact_path(pipeline, task_id, name))
async def store_image(pipeline: str, task_id: str, name: str, image: Image.Image) -> str:
buffer = io.BytesIO()
if name.endswith(".png"):
image.save(buffer, "PNG")
else:
image.convert("RGB").save(buffer, "JPEG", quality=JPEG_QUALITY)
return await store_bytes(pipeline, task_id, name, buffer.getvalue())
async def load_image(url: str) -> Image.Image:
return Image.open(io.BytesIO(await blob.download_bytes(url)))
async def load_optional(url: str | None) -> bytes | None:
return await blob.download_bytes(url) if url else None
def contact_sheet(images: list[Image.Image], tile_width: int = 240,
gap: int = 8) -> Image.Image:
"""검수 화면에 한 줄로 늘어놓는 시트"""
tiles = [image.resize((tile_width, round(tile_width * image.height / image.width)))
for image in images]
height = max(tile.height for tile in tiles)
sheet = Image.new("RGB", ((tile_width + gap) * len(tiles) - gap, height), "black")
for order, tile in enumerate(tiles):
sheet.paste(tile, (order * (tile_width + gap), 0))
return sheet

196
backend/pipelines/gate.py Normal file
View File

@ -0,0 +1,196 @@
"""사람이 검토하는 지점의 정의와 검수 화면 재료
게이트는 특정 단계가 끝난 직후에
페이로드 모양은 프론트가 그리는 것과 1:1
"""
from enum import StrEnum
from models.i2v import TitleGateVerdict
from models.longcut import LongcutPlan
from models.nol import NolMeta
from models.pipeline_state import PlayreelState, PosterAliveState, stage_keys
from tables.task import PlayreelTask
class PlayreelGate(StrEnum):
FETCH_CONFIRM = "fetch_confirm"
ANALYSIS_CONFIRM = "analysis_confirm"
NARRATION_CONFIRM = "narration_confirm"
CLIP_CONFIRM = "clip_confirm"
FINAL_CONFIRM = "final_confirm"
# 이 단계가 끝나면 그 뒤에 게이트가 선다
PLAYREEL_GATE_AFTER = {
PlayreelState.SPLIT: PlayreelGate.FETCH_CONFIRM,
PlayreelState.ANALYZE: PlayreelGate.ANALYSIS_CONFIRM,
PlayreelState.NARRATION: PlayreelGate.NARRATION_CONFIRM,
PlayreelState.I2V: PlayreelGate.CLIP_CONFIRM,
PlayreelState.COMPOSE: PlayreelGate.FINAL_CONFIRM,
}
# 크레딧을 이미 쓴 지점 뒤로는 되돌리지 않는다
REVERSIBLE_GATES = (PlayreelGate.ANALYSIS_CONFIRM, PlayreelGate.NARRATION_CONFIRM)
GATE_ORDER = list(PLAYREEL_GATE_AFTER.values())
GATE_STAGE = {gate: stage for stage, gate in PLAYREEL_GATE_AFTER.items()}
def previous_gate(gate: PlayreelGate) -> PlayreelGate:
return GATE_ORDER[GATE_ORDER.index(gate) - 1]
def resume_stage(gate: PlayreelGate) -> str:
"""그 게이트를 승인하면 이어서 돌 단계"""
keys = stage_keys(PlayreelState)
return keys[keys.index(GATE_STAGE[gate].value) + 1]
# 숏폼은 나레이션과 모션을 한 번에 보여주고 끝
POSTER_ALIVE_GATE_AFTER = PosterAliveState.MOTION
SECTION_LABELS = {
"notice": "유의사항", "event": "이벤트", "keyvisual": "키비주얼", "still": "공연 장면",
"synopsis": "작품 소개", "cast": "캐스트", "creative": "제작진",
"schedule": "캐스팅 스케줄", "discount": "할인 안내", "crossbanner": "타 공연 배너",
"info": "관람 정보", "fragment": "조각", "untagged": "분류 안 됨",
}
REQUIRED_SECTION_TAG = "schedule"
LOW_RESOLUTION_WARN_PX = 800
# tts를 돌리기 전이라 실측 길이가 없다. 게이트 ③에 띄우는 값은 글자 수로 어림한 것이다
ESTIMATED_CHARS_PER_SECOND = 6.5
VOICE_NAMES = {"tc_61e748d0fd9fb2d2cacbb04d": "Yena (여성 · 또렷한 안내톤)"}
def period_text(meta: NolMeta) -> str:
period = f"{meta.play_start_date or ''} ~ {meta.play_end_date or ''}".strip(" ~")
return period.replace("-", ".")
def fetch_review(task: PlayreelTask) -> dict:
meta = NolMeta.model_validate(task.nol_meta)
# 출연진은 게이트 ③ 앞 단계에서 읽으므로 여기서는 대개 비어 있다
cast = (task.cast_extraction or {}).get("names", [])
return {
"poster_url": task.poster_url or "",
"poster_width": task.poster_width or 0,
"meta": {
"title": meta.goods_name or "",
"date_text": period_text(meta),
"place": meta.place_name or "",
"cast": cast,
"genre": meta.genre_name or "",
},
"sections": [{
"id": entry["name"],
"tag": entry["tag"],
"label": SECTION_LABELS.get(entry["tag"], entry["tag"]),
"thumb_url": entry["url"],
"height": entry["y1"] - entry["y0"],
"required": entry["tag"] == REQUIRED_SECTION_TAG,
"selected": entry["selected"],
} for entry in task.detail_sections or []],
}
def analysis_review(task: PlayreelTask) -> dict:
return {
"grid_url": task.analysis_grid_url or "",
"movable": [{"key": choice["key"], "label": choice["label"], "on": choice["on"]}
for choice in task.movable_elements or []],
"fixed": [{"key": layer["key"], "label": layer["label"], "on": layer["on"]}
for layer in task.fixed_layers or []],
"model": task.i2v_model or "",
"ip_risk": bool(task.has_ip_risk),
"has_qr": bool(task.has_qr),
}
def narration_review(task: PlayreelTask) -> dict:
lines = task.narration_lines or []
total_chars = sum(len(line["text"]) for line in lines)
voice_id = task.narration_voice_id or next(iter(VOICE_NAMES))
return {
"lines": [{"slot": line["slot"], "text": line["text"]} for line in lines],
"voice": {
"id": voice_id,
"name": VOICE_NAMES.get(voice_id, voice_id),
"sample_url": f"/api/playreel/voices/{voice_id}/sample",
},
"est_seconds": round(total_chars / ESTIMATED_CHARS_PER_SECOND, 1),
}
def clip_review(task: PlayreelTask) -> dict:
verdict = TitleGateVerdict.model_validate(task.title_gate or {"passed": True})
return {
"clip_url": task.clip_url or "",
"frames_url": task.clip_frames_url or "",
"title_mae": verdict.pixel_score or 0.0,
"gate_passed": verdict.passed,
"gate_reason": verdict.reason or verdict.skip or None,
"retry_credits": task.clip_credits or 0.0,
}
def final_checks(task: PlayreelTask) -> list[dict]:
"""컬럼으로 확인되는 것만 낸다 — 완성본을 다시 열어 본 결과가 아니다."""
verdict = TitleGateVerdict.model_validate(task.title_gate or {"passed": True})
plan = LongcutPlan.model_validate(task.scene_plan or {"scenes": []})
scroll_tags = {scene.tag for scene in plan.scenes if scene.kind == "scroll"}
return [
{"key": "hybrid", "label": "원본 글자 합성",
"ok": bool(task.hybrid_clip_url)},
{"key": "title", "label": "제목 훼손 검사",
"ok": verdict.passed and not verdict.skip},
{"key": "scroll", "label": "상세페이지 스크롤",
"ok": any(scene.kind == "scroll" for scene in plan.scenes)},
{"key": "schedule", "label": "캐스팅 스케줄 포함",
"ok": REQUIRED_SECTION_TAG in scroll_tags},
{"key": "bgm", "label": "배경음악",
"ok": bool(task.bgm_audio_url)},
]
def final_review(task: PlayreelTask) -> dict:
return {
"video_url": task.video_url or "",
"proof_url": task.compose_proof_url or task.hybrid_proof_url or "",
"duration": task.video_duration or 0.0,
"checks": final_checks(task),
"version": task.version + 1,
}
PLAYREEL_REVIEWS = {
PlayreelGate.FETCH_CONFIRM: fetch_review,
PlayreelGate.ANALYSIS_CONFIRM: analysis_review,
PlayreelGate.NARRATION_CONFIRM: narration_review,
PlayreelGate.CLIP_CONFIRM: clip_review,
PlayreelGate.FINAL_CONFIRM: final_review,
}
def build_review(task: PlayreelTask, gate: str) -> dict:
return {"gate": gate, "data": PLAYREEL_REVIEWS[PlayreelGate(gate)](task)}
def apply_playreel_edits(task: PlayreelTask, gate: str, edits: dict) -> None:
"""검수 화면에서 고친 값을 컬럼에 써넣는다. 다음 단계가 이 컬럼을 읽는다."""
if gate == PlayreelGate.FETCH_CONFIRM and "sections" in edits:
keep = set(edits["sections"])
task.detail_sections = [
{**entry, "selected": entry["name"] in keep or entry["tag"] == REQUIRED_SECTION_TAG}
for entry in task.detail_sections or []]
if gate == PlayreelGate.ANALYSIS_CONFIRM and "movable" in edits:
chosen = set(edits["movable"])
task.movable_elements = [{**choice, "on": choice["key"] in chosen}
for choice in task.movable_elements or []]
if gate == PlayreelGate.ANALYSIS_CONFIRM and "fixed" in edits:
chosen = set(edits["fixed"])
task.fixed_layers = [{**layer, "on": layer["key"] in chosen}
for layer in task.fixed_layers or []]
if gate == PlayreelGate.NARRATION_CONFIRM and edits.get("lines"):
task.narration_lines = [{**line, "text": text} for line, text
in zip(task.narration_lines or [], edits["lines"])]

View File

@ -0,0 +1,254 @@
"""공연 상품페이지 URL → 30초 롱컷
함수가 단계를 맡아 단계 컬럼에서 입력을 되읽고 순수 서비스를 부른
산출물 컬럼과 state를 갱신하고 커밋함
이미지·오디오·영상은 blob에 올린 URL만 컬럼에 남음
"""
import asyncio
import io
from pathlib import Path
from PIL import Image
from sqlalchemy.ext.asyncio import AsyncSession
from models.detail_section import DetailSection
from models.hybrid import LayerConfig
from models.longcut import EndBandInfo
from models.motion import MotionPlan
from models.nol import NolMeta
from models.pipeline_state import PlayreelState, initial_timings
from models.tts import NarrationTimeline, TypecastVoice
from pipelines.artifact import (contact_sheet, load_image, load_optional, store_bytes,
store_image)
from services.analyze_poster import analyze_poster
from services.bgm import DEFAULT_BGM_STYLE, generate_bgm
from services.build_layer_config import build_layer_config
from services.compose_long import compose_longcut
from services.extract_cast import CAST_TAG, extract_cast
from services.fetch_nol import fetch_product
from services.hybrid_poster import render_hybrid
from services.i2v import animate
from services.motion import plan_motion
from services.narration_slots import build_narration
from services.split_detail import split_details
from services.tts import synthesize
from services.upscale_poster import as_png_path, upscale_poster
from tables.task import PlayreelTask
from utils import blob
from utils.higgsfield import KLING_3_0
from utils.video import frames_at
PIPELINE = "playreel"
QR_DIR = Path(__file__).parent.parent / "assets" / "nol"
CLIP_REVIEW_TIMES = (0.0, 2.0, 4.0, 6.0, 7.5) # 게이트 ④에 띄우는 5시점
def band_info(meta: NolMeta) -> EndBandInfo:
period = f"{meta.play_start_date or ''} ~ {meta.play_end_date or ''}".strip(" ~")
detail = " · ".join(part for part in (period.replace("-", "."), meta.place_name) if part)
return EndBandInfo(title=meta.goods_name or "", detail=detail)
async def load_sections(entries: list[dict]) -> list[DetailSection]:
return [DetailSection(source=entry["source"], index=entry["index"],
y0=entry["y0"], y1=entry["y1"], tag=entry["tag"],
image=Image.open(io.BytesIO(await blob.download_bytes(entry["url"]))))
for entry in entries]
def working_poster_url(task: PlayreelTask) -> str:
"""업스케일이 있으면 그것이 이후 단계의 원본이 됨"""
return task.upscaled_poster_url or task.poster_url
async def create_task(session: AsyncSession, url: str, goods_id: str,
slug: str) -> PlayreelTask:
task = PlayreelTask(source_url=url, goods_id=goods_id, slug=slug,
name=f"공연 {goods_id}",
stage_timings=initial_timings(PlayreelState))
session.add(task)
await session.commit()
return task
async def run_fetch(session: AsyncSession, task: PlayreelTask) -> None:
product = await fetch_product(task.goods_id)
task.nol_meta = product.meta.model_dump()
task.name = product.meta.goods_name or task.name
if product.poster:
task.poster_url = await store_bytes(
PIPELINE, task.id, f"poster.{product.poster.extension}", product.poster.data)
with Image.open(io.BytesIO(product.poster.data)) as image:
task.poster_width, task.poster_height = image.size
task.detail_image_urls = [
await store_bytes(PIPELINE, task.id, f"detail_{order:02d}.{image.extension}",
image.data)
for order, image in enumerate(product.details, 1)]
task.fetch_failed_urls = product.failed_urls
task.state = PlayreelState.SPLIT
await session.commit()
async def run_split(session: AsyncSession, task: PlayreelTask) -> None:
details = [await blob.download_bytes(url) for url in task.detail_image_urls or []]
result = await split_details(details)
entries = []
for section in result.sections:
url = await store_image(PIPELINE, task.id, f"sections/{section.name}.png",
section.image)
# selected는 게이트 ①에서 사람이 끄면 갱신됨
entries.append({"name": section.name, "source": section.source,
"index": section.index, "y0": section.y0, "y1": section.y1,
"tag": section.tag, "url": url, "selected": True})
task.detail_sections = entries
task.section_sheet_url = await store_image(PIPELINE, task.id, "sections/_sheet.jpg",
result.sheet)
task.state = PlayreelState.UPSCALE
await session.commit()
async def run_upscale(session: AsyncSession, task: PlayreelTask) -> None:
result = await upscale_poster(await blob.download_bytes(task.poster_url))
task.upscaled_poster_url = await store_bytes(PIPELINE, task.id, "upscaled.png",
result.image)
task.upscale_credits = result.credits
task.upscale_variant = result.variant
task.credits_used += result.credits
task.state = PlayreelState.ANALYZE
await session.commit()
async def run_analyze(session: AsyncSession, task: PlayreelTask) -> None:
meta = NolMeta.model_validate(task.nol_meta)
analysis = await analyze_poster(await load_image(working_poster_url(task)),
meta.goods_name or "")
task.analysis_grid_url = await store_image(PIPELINE, task.id, "review_grid.jpg",
analysis.grid)
task.movable_elements = [choice.model_dump() for choice in analysis.movable]
task.fixed_layers = [layer.model_dump() for layer in analysis.fixed]
task.i2v_model = analysis.model
task.has_ip_risk = analysis.ip_risk
task.has_qr = analysis.has_qr
task.state = PlayreelState.MOTION
await session.commit()
async def run_motion(session: AsyncSession, task: PlayreelTask) -> None:
# 게이트 ②에서 사람이 켜 둔 요소가 그대로 프롬프트가 됨
chosen = [choice["key"] for choice in task.movable_elements or [] if choice["on"]]
plan = await plan_motion(await load_image(working_poster_url(task)),
force_motions=chosen)
task.motion_plan = plan.model_dump()
task.state = PlayreelState.NARRATION
await session.commit()
async def run_narration(session: AsyncSession, task: PlayreelTask) -> None:
# 출연진은 cast 태그 섹션에서만 읽으므로 그 조각만 내려받음
tagged = [entry for entry in task.detail_sections or [] if entry["tag"] == CAST_TAG]
cast = await extract_cast(await load_sections(tagged))
narration = build_narration(NolMeta.model_validate(task.nol_meta), cast)
task.cast_extraction = cast.model_dump()
task.narration_lines = [line.model_dump() for line in narration.lines]
task.state = PlayreelState.TTS
await session.commit()
async def run_tts(session: AsyncSession, task: PlayreelTask) -> None:
voice = TypecastVoice()
audio = await synthesize([line["text"] for line in task.narration_lines],
typecast=voice)
task.narration_timeline = audio.timeline.model_dump()
task.narration_audio_url = await store_bytes(PIPELINE, task.id, "narration.mp3",
audio.mp3)
task.narration_voice_id = voice.voice_id
task.state = PlayreelState.BGM
await session.commit()
async def run_bgm(session: AsyncSession, task: PlayreelTask) -> None:
# 롱컷은 나레이션을 VLM이 쓰지 않아 스타일도 안 나옴
timeline = NarrationTimeline.model_validate(task.narration_timeline)
audio = await generate_bgm(DEFAULT_BGM_STYLE, timeline.total, title=task.name)
task.bgm_audio_url = await store_bytes(PIPELINE, task.id, "bgm.mp3", audio.mp3)
task.bgm_task_id = audio.task_id
task.bgm_target_seconds = audio.seconds
task.bgm_duration = audio.duration
task.state = PlayreelState.I2V
await session.commit()
async def run_i2v(session: AsyncSession, task: PlayreelTask) -> None:
plan = MotionPlan.model_validate(task.motion_plan)
poster = await blob.download_bytes(working_poster_url(task))
# detect를 안 거쳐 regions가 없으므로 제목 훼손 게이트는 건너뜀
with as_png_path(poster) as poster_path:
result = await animate(plan.prompt, poster_path, model=task.i2v_model or KLING_3_0)
task.clip_url = await store_bytes(PIPELINE, task.id, "clip.mp4", result.clip)
task.clip_frames_url = await store_image(
PIPELINE, task.id, "clip_frames.jpg",
contact_sheet(frames_at(result.clip, CLIP_REVIEW_TIMES)))
task.clip_model = result.model
task.clip_credits = result.credits
task.title_gate = result.title_gate.model_dump()
task.credits_used += result.credits
task.state = PlayreelState.HYBRID
await session.commit()
async def run_hybrid(session: AsyncSession, task: PlayreelTask) -> None:
poster = await blob.download_bytes(working_poster_url(task))
config = (LayerConfig.model_validate(task.layer_config) if task.layer_config
else await build_layer_config(Image.open(io.BytesIO(poster))))
clip = await blob.download_bytes(task.clip_url)
# 프레임 합성과 인코딩이 수 분간 CPU를 잡아 이벤트 루프를 막음
result = await asyncio.to_thread(render_hybrid, poster, clip, config, preview=True)
task.layer_config = config.model_dump()
task.hybrid_clip_url = await store_bytes(PIPELINE, task.id, "hybrid.mp4", result.video)
task.hybrid_frames = result.frames
task.hybrid_duration = result.duration
if result.proof:
task.hybrid_proof_url = await store_image(PIPELINE, task.id, "hybrid_proof.jpg",
result.proof)
task.state = PlayreelState.COMPOSE
await session.commit()
async def run_compose(session: AsyncSession, task: PlayreelTask) -> None:
meta = NolMeta.model_validate(task.nol_meta)
timeline = NarrationTimeline.model_validate(task.narration_timeline)
selected = [entry for entry in task.detail_sections or [] if entry["selected"]]
sections = await load_sections(selected)
# 하이브리드 합성을 못 거친 클립은 제목이 흔들림
clip = await blob.download_bytes(task.hybrid_clip_url or task.clip_url)
poster = await load_image(working_poster_url(task))
narration = await blob.download_bytes(task.narration_audio_url)
bgm = await load_optional(task.bgm_audio_url)
qr_path = QR_DIR / f"qr_{task.goods_id}.png"
result = await asyncio.to_thread(
compose_longcut, poster, clip, timeline, narration, sections, band_info(meta),
bgm_mp3=bgm, qr=Image.open(qr_path) if qr_path.exists() else None)
task.video_url = await store_bytes(PIPELINE, task.id,
f"longcut_v{task.version + 1}.mp4", result.video)
task.scene_plan = result.plan.model_dump()
task.video_duration = result.duration
task.video_frames = result.frames
if result.proof:
task.compose_proof_url = await store_image(PIPELINE, task.id, "compose_proof.jpg",
result.proof)
task.state = PlayreelState.REVIEW
await session.commit()

View File

@ -0,0 +1,140 @@
"""포스터 한 장 → 8초 숏폼
함수가 단계를 맡아 단계 컬럼에서 입력을 되읽고 순수 서비스를 부른
산출물 컬럼과 state를 갱신하고 커밋함
이미지·오디오·영상은 blob에 올린 URL만 컬럼에 남음
"""
import asyncio
import io
from PIL import Image
from sqlalchemy.ext.asyncio import AsyncSession
from answers.narration_answer import BgmStyle
from models.detect import Regions
from models.motion import MotionPlan
from models.pipeline_state import PosterAliveState, initial_timings
from models.tts import NarrationTimeline
from pipelines.artifact import load_image, load_optional, store_bytes, store_image
from services.bgm import generate_bgm
from services.detect import detect
from services.i2v import MIN_LONG_EDGE_PX, animate
from services.motion import plan_motion
from services.narration import generate_narration
from services.render import render
from services.tts import synthesize
from services.upscale_poster import as_png_path
from tables.task import PosterAliveTask
from utils import blob
from utils.image import sniff_extension
PIPELINE = "poster_alive"
async def create_task(session: AsyncSession, name: str, poster: bytes, *,
skip_review: bool = False) -> PosterAliveTask:
task = PosterAliveTask(name=name, skip_review=skip_review,
stage_timings=initial_timings(PosterAliveState))
with Image.open(io.BytesIO(poster)) as image:
task.poster_width, task.poster_height = image.size
# i2v만 거절하는 하드 게이트라 여기서는 기록만 함
task.is_low_resolution = max(image.size) < MIN_LONG_EDGE_PX
session.add(task)
await session.flush() # blob 경로에 쓸 id를 확정
task.poster_url = await store_bytes(PIPELINE, task.id,
f"poster.{sniff_extension(poster)}", poster)
await session.commit()
return task
async def run_detect(session: AsyncSession, task: PosterAliveTask) -> None:
result = await detect(await load_image(task.poster_url))
task.detect_regions = result.data.model_dump()
task.detect_grid_url = await store_image(PIPELINE, task.id, "detect_grid.jpg", result.grid)
task.detect_check_url = await store_image(PIPELINE, task.id, "detect_check.jpg",
result.check)
task.state = PosterAliveState.NARRATION_TEXT
await session.commit()
async def run_narration_text(session: AsyncSession, task: PosterAliveTask) -> None:
regions = Regions.model_validate(task.detect_regions)
answer = await generate_narration(await load_image(task.poster_url), regions)
task.narration_lines = answer.narration
task.narration_voice = answer.voice
task.bgm_style = answer.bgm_style.model_dump()
task.poster_metadata = answer.metadata.model_dump()
task.state = PosterAliveState.MOTION
await session.commit()
async def run_motion(session: AsyncSession, task: PosterAliveTask) -> None:
plan = await plan_motion(await load_image(task.poster_url))
task.motion_plan = plan.model_dump()
task.state = PosterAliveState.TTS
await session.commit()
async def run_tts(session: AsyncSession, task: PosterAliveTask) -> None:
audio = await synthesize(task.narration_lines, task.narration_voice)
task.narration_timeline = audio.timeline.model_dump()
task.narration_audio_url = await store_bytes(PIPELINE, task.id, "narration.mp3", audio.mp3)
task.state = PosterAliveState.BGM
await session.commit()
async def run_bgm(session: AsyncSession, task: PosterAliveTask) -> None:
timeline = NarrationTimeline.model_validate(task.narration_timeline)
audio = await generate_bgm(BgmStyle.model_validate(task.bgm_style), timeline.total,
title=task.name)
task.bgm_audio_url = await store_bytes(PIPELINE, task.id, "bgm.mp3", audio.mp3)
task.bgm_task_id = audio.task_id
task.bgm_target_seconds = audio.seconds
task.bgm_duration = audio.duration
task.state = PosterAliveState.I2V
await session.commit()
async def run_i2v(session: AsyncSession, task: PosterAliveTask) -> None:
plan = MotionPlan.model_validate(task.motion_plan)
regions = Regions.model_validate(task.detect_regions)
poster = await blob.download_bytes(task.poster_url)
# CLI가 파일 경로만 받고, 팔레트 이미지를 그대로 올리면 잡이 실패로 돌아옴
with as_png_path(poster) as poster_path:
result = await animate(plan.prompt, poster_path, regions)
task.clip_url = await store_bytes(PIPELINE, task.id, "clip.mp4", result.clip)
task.clip_model = result.model
task.clip_credits = result.credits
task.title_gate = result.title_gate.model_dump()
task.credits_used += result.credits
task.state = PosterAliveState.RENDER
await session.commit()
async def run_render(session: AsyncSession, task: PosterAliveTask) -> None:
timeline = (NarrationTimeline.model_validate(task.narration_timeline)
if task.narration_timeline else None)
clip = await blob.download_bytes(task.clip_url)
poster = await load_image(task.poster_url)
narration = await load_optional(task.narration_audio_url)
bgm = await load_optional(task.bgm_audio_url)
# 프레임 합성과 인코딩이 수 분간 CPU를 잡아 이벤트 루프를 막음
result = await asyncio.to_thread(render, clip, poster, narration=narration, bgm=bgm,
timeline=timeline)
task.video_url = await store_bytes(PIPELINE, task.id, "final.mp4", result.video)
task.thumbnail_url = await store_bytes(PIPELINE, task.id, "thumbnail.jpg",
result.thumbnail)
task.video_duration = result.duration
task.video_frames = result.frames
task.video_width = result.width
task.video_height = result.height
task.state = PosterAliveState.COMPLETED
await session.commit()

133
backend/pipelines/runner.py Normal file
View File

@ -0,0 +1,133 @@
"""현재 state부터 다음 게이트까지 단계를 이어 돌림
어느 잡을 언제 돌릴지는 worker가 정하고 여기는 실행만
"""
import time
import traceback
from sqlalchemy.ext.asyncio import AsyncSession
from models.pipeline_state import PlayreelState, PosterAliveState, stage_keys
from pipelines import playreel, poster_alive
from pipelines.gate import PLAYREEL_GATE_AFTER, POSTER_ALIVE_GATE_AFTER
from tables.task import PlayreelTask, PosterAliveTask
QUEUED = "queued"
RUNNING = "running"
AWAITING_REVIEW = "awaiting_review"
FAILED = "failed"
DONE = "done"
POSTER_ALIVE_RUNNERS = {
PosterAliveState.DETECT: poster_alive.run_detect,
PosterAliveState.NARRATION_TEXT: poster_alive.run_narration_text,
PosterAliveState.MOTION: poster_alive.run_motion,
PosterAliveState.TTS: poster_alive.run_tts,
PosterAliveState.BGM: poster_alive.run_bgm,
PosterAliveState.I2V: poster_alive.run_i2v,
PosterAliveState.RENDER: poster_alive.run_render,
}
PLAYREEL_RUNNERS = {
PlayreelState.FETCH: playreel.run_fetch,
PlayreelState.SPLIT: playreel.run_split,
PlayreelState.UPSCALE: playreel.run_upscale,
PlayreelState.ANALYZE: playreel.run_analyze,
PlayreelState.MOTION: playreel.run_motion,
PlayreelState.NARRATION: playreel.run_narration,
PlayreelState.TTS: playreel.run_tts,
PlayreelState.BGM: playreel.run_bgm,
PlayreelState.I2V: playreel.run_i2v,
PlayreelState.HYBRID: playreel.run_hybrid,
PlayreelState.COMPOSE: playreel.run_compose,
# ⑫ review는 대응하는 서비스가 없어 통과만 시킨다
}
STATES = {PosterAliveTask: PosterAliveState, PlayreelTask: PlayreelState}
RUNNERS = {PosterAliveTask: POSTER_ALIVE_RUNNERS, PlayreelTask: PLAYREEL_RUNNERS}
def mark(task, stage: str, status: str) -> None:
timings = dict(task.stage_timings or {})
entry = dict(timings.get(stage) or {"started": None, "ended": None})
entry["status"] = status
if status == RUNNING:
entry["started"] = time.time()
elif status in ("done", FAILED):
entry["ended"] = time.time()
timings[stage] = entry
task.stage_timings = timings
def reset_from(task, stage: str) -> None:
"""되돌리기·재시도에서 그 단계부터 다시 돌게 만든다"""
state_type = STATES[type(task)]
keys = stage_keys(state_type)
timings = dict(task.stage_timings or {})
for key in keys[keys.index(stage):]:
timings[key] = {"status": "idle", "started": None, "ended": None}
task.stage_timings = timings
task.state = state_type(stage)
def stops_here(task, stage) -> str | None:
"""이 단계 뒤에 사람이 볼 게이트가 서는가. playreel은 게이트 이름을 돌려준다"""
if isinstance(task, PlayreelTask):
gate = PLAYREEL_GATE_AFTER.get(stage)
return gate.value if gate else None
if stage == POSTER_ALIVE_GATE_AFTER and not task.skip_review:
return AWAITING_REVIEW
return None
async def step(session: AsyncSession, task) -> bool:
"""한 단계를 돌린다. 게이트에 닿으면 False를 돌려 멈춤을 알린다"""
state_type = STATES[type(task)]
stage = state_type(task.state)
runner = RUNNERS[type(task)].get(stage)
if runner is None: # 러너 없는 마지막 단계
mark(task, stage.value, "done")
task.state = state_type.COMPLETED
await session.commit()
return True
mark(task, stage.value, RUNNING)
await session.commit()
await runner(session, task) # state를 다음 단계로 올리고 커밋한다
mark(task, stage.value, "done")
gate = stops_here(task, stage)
if gate:
task.status = AWAITING_REVIEW
if isinstance(task, PlayreelTask):
task.gate = gate
await session.commit()
return False
await session.commit()
return True
async def run_until_gate(session: AsyncSession, task) -> None:
"""다음 게이트나 끝까지 민다. 실패는 잡에 기록하고 조용히 돌아온다"""
state_type = STATES[type(task)]
try:
while task.state != state_type.COMPLETED:
if not await step(session, task):
return
except Exception:
failed_stage = str(task.state)
await session.rollback()
await session.refresh(task)
mark(task, failed_stage, FAILED)
task.status = FAILED
task.error_stage = failed_stage
task.error_detail = traceback.format_exc()[-2000:]
await session.commit()
return
task.status = DONE
if isinstance(task, PlayreelTask):
task.gate = None
task.version += 1
await session.commit()

View File

@ -0,0 +1,77 @@
"""queued 잡을 하나씩 집어 다음 게이트까지 돌리는 배경 실행기
큐가 DB에 있어 동시 실행 수를 올리거나 컨테이너를 늘려도 파이프라인 코드는 그대로
승인이 status를 queued로 되돌리면 다음 폴링에서 이어
"""
import asyncio
import traceback
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession
from pipelines.runner import QUEUED, RUNNING, run_until_gate
from tables.task import PlayreelTask, PosterAliveTask
from utils.database import session_factory
POLL_INTERVAL = 2.0
TABLES = (PlayreelTask, PosterAliveTask)
_worker: asyncio.Task | None = None
async def claim_next(session: AsyncSession):
"""queued 잡 하나를 running으로 바꿔 가져온다
워커가 하나인 동안은 경합이 없다. 늘릴 SELECT ... FOR UPDATE SKIP LOCKED로 바꾼다
"""
for table in TABLES:
found = await session.scalars(
select(table).where(table.status == QUEUED)
.order_by(table.created_at).limit(1))
task = found.first()
if task is not None:
task.status = RUNNING
await session.commit()
return task
return None
async def run_once() -> bool:
async with session_factory() as session:
task = await claim_next(session)
if task is None:
return False
await run_until_gate(session, task)
return True
async def requeue_running() -> None:
"""기동 시 running으로 남은 잡을 되살린다 — 어디까지 갔는지는 state가 기억한다"""
async with session_factory() as session:
for table in TABLES:
for task in await session.scalars(select(table).where(table.status == RUNNING)):
task.status = QUEUED
await session.commit()
async def loop() -> None:
await requeue_running()
while True:
try:
if not await run_once():
await asyncio.sleep(POLL_INTERVAL)
except asyncio.CancelledError:
raise
except Exception:
# 워커는 죽지 않는다. DB가 끊긴 경우라 잠시 쉬고 다시 본다
traceback.print_exc()
await asyncio.sleep(POLL_INTERVAL)
def start() -> None:
global _worker
_worker = asyncio.create_task(loop(), name="pipeline-worker")
async def stop() -> None:
if _worker is not None:
_worker.cancel()
await asyncio.gather(_worker, return_exceptions=True)

163
backend/routers/f1.py Normal file
View File

@ -0,0 +1,163 @@
"""/api/f1 — 포스터 한 장 → 8초 숏폼"""
import io
from fastapi import APIRouter, Depends, File, Form, HTTPException, UploadFile
from PIL import Image
from pydantic import BaseModel
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession
from models.pipeline_state import PosterAliveState
from pipelines import poster_alive
from pipelines.artifact import load_image
from pipelines.runner import AWAITING_REVIEW, FAILED, QUEUED, RUNNING, reset_from
from routers.view import motion_elements, poster_alive_view
from services.motion import plan_motion
from tables.task import PosterAliveTask
from utils.database import get_session
router = APIRouter(prefix="/api/f1", tags=["f1"])
ACCEPTED_TYPES = {"image/jpeg", "image/png", "image/webp", "image/gif"}
MAX_UPLOAD_BYTES = 30 * 1024 * 1024
NARRATION_COUNT = 3
NARRATION_MAX_CHARS = 40
class NarrationBody(BaseModel):
narration: list[str]
class MetadataBody(BaseModel):
event_name: str
date_text: str = ""
place: str
keywords: list[str] | None = None
class ApproveBody(BaseModel):
# 검수 화면에서 사람이 모션을 더하거나 뺀 결과. None이면 그대로 간다
motions: list[str] | None = None
async def find_task(session: AsyncSession, task_id: str) -> PosterAliveTask:
task = await session.get(PosterAliveTask, task_id)
if task is None:
raise HTTPException(404, "잡을 찾을 수 없습니다")
return task
async def rewrite_motion_plan(task: PosterAliveTask, motions: list[str]) -> None:
"""고른 모션으로 프롬프트를 다시 쓴다. 바뀐 게 없으면 건드리지 않는다"""
chosen = [key.strip() for key in motions if key.strip()]
if sorted(chosen) == sorted(motion_elements(task.motion_plan) or []):
return
plan = await plan_motion(await load_image(task.poster_url), force_motions=chosen)
task.motion_plan = plan.model_dump()
@router.post("/jobs")
async def create(poster: UploadFile = File(...), name: str = Form(""),
auto: bool = Form(False),
session: AsyncSession = Depends(get_session)):
if poster.content_type not in ACCEPTED_TYPES:
raise HTTPException(415, f"지원 형식: JPG/PNG/WEBP/GIF (받은 것: {poster.content_type})")
raw = await poster.read()
if len(raw) > MAX_UPLOAD_BYTES:
raise HTTPException(413, "30MB 이하만 올릴 수 있습니다")
try:
Image.open(io.BytesIO(raw)).verify()
except Exception:
raise HTTPException(422, "이미지 파일을 열 수 없습니다")
task = await poster_alive.create_task(session, name.strip() or "이름 없는 행사", raw,
skip_review=auto)
return {"id": task.id, "ahead": 0}
@router.get("/jobs")
async def list_jobs(session: AsyncSession = Depends(get_session)):
found = await session.scalars(
select(PosterAliveTask).order_by(PosterAliveTask.created_at.desc()))
return [poster_alive_view(task) for task in found]
@router.get("/jobs/{task_id}")
async def get_job(task_id: str, session: AsyncSession = Depends(get_session)):
return poster_alive_view(await find_task(session, task_id))
@router.put("/jobs/{task_id}/narration")
async def edit_narration(task_id: str, body: NarrationBody,
session: AsyncSession = Depends(get_session)):
task = await find_task(session, task_id)
if task.status != AWAITING_REVIEW:
raise HTTPException(409, "나레이션은 검수 대기 상태에서만 고칠 수 있습니다")
lines = [line.strip() for line in body.narration]
if len(lines) != NARRATION_COUNT or any(
not line or len(line) > NARRATION_MAX_CHARS for line in lines):
raise HTTPException(422, f"나레이션은 {NARRATION_MAX_CHARS}자 이내 "
f"{NARRATION_COUNT}문장이어야 합니다")
task.narration_lines = lines
await session.commit()
return poster_alive_view(task)
@router.put("/jobs/{task_id}/metadata")
async def edit_metadata(task_id: str, body: MetadataBody,
session: AsyncSession = Depends(get_session)):
"""VLM이 연도나 지명을 잘못 읽어도 아카이브에는 사실이 남게 한다"""
task = await find_task(session, task_id)
if task.status != AWAITING_REVIEW:
raise HTTPException(409, "메타태그는 검수 대기 상태에서만 고칠 수 있습니다")
if not body.event_name.strip() or not body.place.strip():
raise HTTPException(422, "행사명과 장소는 비울 수 없습니다")
metadata = dict(task.poster_metadata or {})
metadata["event_name"] = body.event_name.strip()
metadata["date_text"] = body.date_text.strip()
metadata["place"] = body.place.strip()
if body.keywords is not None:
metadata["keywords"] = [word.strip() for word in body.keywords if word.strip()]
task.poster_metadata = metadata
await session.commit()
return poster_alive_view(task)
@router.post("/jobs/{task_id}/approve")
async def approve(task_id: str, body: ApproveBody | None = None,
session: AsyncSession = Depends(get_session)):
task = await find_task(session, task_id)
if task.status != AWAITING_REVIEW:
raise HTTPException(409, f"검수 대기 상태가 아닙니다: {task.status}")
if body and body.motions is not None:
await rewrite_motion_plan(task, body.motions)
task.status = QUEUED
await session.commit()
return {"id": task.id, "ahead": 0}
@router.post("/jobs/{task_id}/retry")
async def retry(task_id: str, body: ApproveBody | None = None,
session: AsyncSession = Depends(get_session)):
task = await find_task(session, task_id)
if task.status != FAILED:
raise HTTPException(409, f"실패 상태가 아닙니다: {task.status}")
if body and body.motions is not None:
await rewrite_motion_plan(task, body.motions)
reset_from(task, task.error_stage or PosterAliveState.DETECT.value)
task.status = QUEUED
task.error_stage = None
task.error_detail = None
await session.commit()
return {"id": task.id, "ahead": 0}
@router.delete("/jobs/{task_id}")
async def delete(task_id: str, session: AsyncSession = Depends(get_session)):
task = await find_task(session, task_id)
if task.status == RUNNING:
raise HTTPException(409, "진행 중인 작업은 삭제할 수 없습니다. 끝난 뒤 지워주세요")
await session.delete(task)
await session.commit()
return {"id": task_id, "removed": True}

159
backend/routers/playreel.py Normal file
View File

@ -0,0 +1,159 @@
"""/api/playreel — 공연 상품페이지 → 30초 롱컷"""
import re
from fastapi import APIRouter, Depends, HTTPException
from pydantic import BaseModel
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession
from models.pipeline_state import PlayreelState
from pipelines import playreel
from pipelines.gate import (REVERSIBLE_GATES, PlayreelGate, apply_playreel_edits,
previous_gate, resume_stage)
from pipelines.runner import AWAITING_REVIEW, FAILED, QUEUED, RUNNING, reset_from
from routers.view import playreel_view
from tables.task import PlayreelTask
from utils.database import get_session
router = APIRouter(prefix="/api/playreel", tags=["playreel"])
# 프론트 parseGoodsId와 같은 규칙. 프론트 검사는 입력 도중의 안내이고 판정은 여기서 한다
GOODS_ID_PATTERNS = (
r"tickets\.interpark\.com/goods/(\d{5,})",
r"nol\.interpark\.com/[^?]*?(\d{8,})",
r"[?&]goodsCode=(\d{5,})",
r"^(\d{8,})$",
)
SLUG_PATTERN = re.compile(r"[a-z0-9_]+")
def parse_goods_id(url: str) -> str | None:
for pattern in GOODS_ID_PATTERNS:
found = re.search(pattern, (url or "").strip(), re.I)
if found:
return found.group(1)
return None
class CreateBody(BaseModel):
url: str
# 하이브리드 합성이 작품별 좌표를 요구한다. 검증된 작품을 다시 돌릴 때 지정한다
slug: str | None = None
class ApproveBody(BaseModel):
gate: str | None = None
edits: dict | None = None
async def find_task(session: AsyncSession, task_id: str) -> PlayreelTask:
task = await session.get(PlayreelTask, task_id)
if task is None:
raise HTTPException(404, "playreel 잡을 찾을 수 없습니다")
return task
@router.post("/jobs")
async def create(body: CreateBody, session: AsyncSession = Depends(get_session)):
goods_id = parse_goods_id(body.url)
if not goods_id:
raise HTTPException(400, "지원하지 않는 주소입니다. 공연 상품페이지 주소를 넣어 주세요.")
slug = (body.slug or "").strip() or f"g{goods_id}"
if not SLUG_PATTERN.fullmatch(slug):
raise HTTPException(400, "slug는 영문 소문자·숫자·밑줄만 씁니다")
task = await playreel.create_task(session, body.url.strip(), goods_id, slug)
return {"id": task.id}
@router.get("/jobs")
async def list_jobs(session: AsyncSession = Depends(get_session)):
found = await session.scalars(
select(PlayreelTask).order_by(PlayreelTask.created_at.desc()))
return [playreel_view(task) for task in found]
@router.get("/jobs/{task_id}")
async def get_job(task_id: str, session: AsyncSession = Depends(get_session)):
return playreel_view(await find_task(session, task_id))
@router.post("/jobs/{task_id}/approve")
async def approve(task_id: str, body: ApproveBody | None = None,
session: AsyncSession = Depends(get_session)):
"""게이트 승인. edits가 없으면 검수 기본값 그대로 간다"""
task = await find_task(session, task_id)
if task.status != AWAITING_REVIEW:
raise HTTPException(409, "지금은 승인할 수 있는 상태가 아닙니다")
if body and body.gate and body.gate != task.gate:
raise HTTPException(409, f"게이트가 어긋납니다: 화면 {body.gate} / 서버 {task.gate}")
if body and body.edits:
apply_playreel_edits(task, task.gate, body.edits)
task.status = QUEUED
task.gate = None
await session.commit()
return playreel_view(task)
@router.post("/jobs/{task_id}/back")
async def back(task_id: str, session: AsyncSession = Depends(get_session)):
"""직전 게이트로. 크레딧을 이미 쓴 지점 뒤로는 되돌리지 않는다"""
task = await find_task(session, task_id)
if task.status != AWAITING_REVIEW or not task.gate:
raise HTTPException(409, "되돌릴 수 있는 상태가 아닙니다")
# 프론트가 버튼을 숨기는 것에 기대면 14크레딧을 쓴 뒤 되돌아가 다시 태우는 길이 열린다
if PlayreelGate(task.gate) not in REVERSIBLE_GATES:
raise HTTPException(409, "이 단계에서는 되돌릴 수 없습니다. "
"영상을 다시 만들려면 '다시 만들기'를 쓰세요.")
previous = previous_gate(PlayreelGate(task.gate))
reset_from(task, resume_stage(previous))
task.status = AWAITING_REVIEW
task.gate = previous.value
await session.commit()
return playreel_view(task)
@router.post("/jobs/{task_id}/retry")
async def retry(task_id: str, session: AsyncSession = Depends(get_session)):
"""실패 재시도와 클립 재생성. i2v를 다시 돌리면 크레딧이 다시 든다"""
task = await find_task(session, task_id)
restart = (PlayreelState.I2V.value if task.gate == PlayreelGate.CLIP_CONFIRM
else task.error_stage or PlayreelState.FETCH.value)
reset_from(task, restart)
task.status = QUEUED
task.gate = None
task.error_stage = None
task.error_detail = None
await session.commit()
return playreel_view(task)
@router.post("/jobs/{task_id}/force")
async def force(task_id: str, session: AsyncSession = Depends(get_session)):
"""게이트 실패를 알고도 진행"""
task = await find_task(session, task_id)
if task.status not in (FAILED, AWAITING_REVIEW):
raise HTTPException(409, "지금은 진행할 수 있는 상태가 아닙니다")
task.status = QUEUED
task.gate = None
task.error_stage = None
task.error_detail = None
await session.commit()
return playreel_view(task)
@router.delete("/jobs/{task_id}")
async def delete(task_id: str, session: AsyncSession = Depends(get_session)):
task = await find_task(session, task_id)
if task.status == RUNNING:
raise HTTPException(409, "진행 중인 작업은 삭제할 수 없습니다. 끝난 뒤 지워주세요")
await session.delete(task)
await session.commit()
return {"id": task_id, "removed": True}
@router.get("/voices/{voice_id}/sample")
async def voice_sample(voice_id: str):
raise HTTPException(404, "보이스 샘플이 아직 준비되지 않았습니다")

73
backend/routers/view.py Normal file
View File

@ -0,0 +1,73 @@
"""잡 행을 프론트가 그리는 모양으로 옮김
필드는 web/lib/api.ts Job, web/lib/playreel.ts PlayreelJob 맞춤
"""
from models.motion import MotionPlan
from models.pipeline_state import PlayreelState, PosterAliveState, initial_timings
from pipelines.gate import build_review
from tables.task import PlayreelTask, PosterAliveTask
def current_stage(task) -> str | None:
return next((key for key, entry in (task.stage_timings or {}).items()
if entry["status"] == "running"), None)
def error_view(task) -> dict | None:
if not task.error_stage:
return None
return {"stage": task.error_stage, "detail": task.error_detail or ""}
def motion_elements(motion_plan: dict | None) -> list[str] | None:
if not motion_plan:
return None
return [item.motion for item in MotionPlan.model_validate(motion_plan).kept]
def base_view(task, state_type) -> dict:
return {
"id": task.id,
"name": task.name,
"status": task.status,
"stage": current_stage(task),
"stages": task.stage_timings or initial_timings(state_type),
"error": error_view(task),
"created_at": task.created_at.timestamp(),
"credits_used": task.credits_used,
"queue_size": 0,
}
def poster_alive_view(task: PosterAliveTask) -> dict:
return {
**base_view(task, PosterAliveState),
"kind": "f1",
"narration": task.narration_lines,
"motion_elements": motion_elements(task.motion_plan),
"metadata": task.poster_metadata,
"template_id": None,
"poster_size": [task.poster_width, task.poster_height],
"low_res": task.is_low_resolution,
"artifacts": {key: url for key, url in (
("video", task.video_url),
("thumbnail", task.thumbnail_url),
("check_jpg", task.detect_check_url),
) if url},
}
def playreel_view(task: PlayreelTask) -> dict:
return {
**base_view(task, PlayreelState),
"kind": "playreel",
"source": {"url": task.source_url, "goods_id": task.goods_id, "slug": task.slug},
"gate": task.gate,
"review": build_review(task, task.gate) if task.gate else None,
"version": task.version,
"narration": [line["text"] for line in task.narration_lines or []] or None,
"metadata": None,
"artifacts": {key: url for key, url in (
("video", task.video_url),
("thumbnail", task.analysis_grid_url),
) if url},
}

191
backend/tables/task.py Normal file
View File

@ -0,0 +1,191 @@
"""두 파이프라인의 잡 테이블
단계별 산출물을 컬럼으로 들고 있음
이미지·오디오·영상은 blob에 올린 URL만 저장
"""
import uuid
from datetime import datetime
from sqlalchemy import JSON, Boolean, DateTime, Enum, Float, Integer, String, Text, func
from sqlalchemy.orm import Mapped, mapped_column
from models.pipeline_state import PlayreelState, PosterAliveState
from utils.database import Base
URL_LENGTH = 512
TASK_ID_LENGTH = 36
def new_task_id() -> str:
return str(uuid.uuid4())
def state_column(enum_type):
return Enum(enum_type, values_callable=lambda e: [member.value for member in e])
class TaskBase(Base):
"""두 테이블이 같이 쓰는 실행 메타
state는 진행도, status는 시점의 실행 상태로 축이 다름
"""
__abstract__ = True
id: Mapped[str] = mapped_column(String(TASK_ID_LENGTH), primary_key=True,
default=new_task_id)
name: Mapped[str] = mapped_column(String(255), default="")
status: Mapped[str] = mapped_column(String(32), default="queued", index=True)
# 스테퍼가 그리는 단계별 시각
# {stage: {"status": ..., "started": ..., "ended": ...}}
stage_timings: Mapped[dict] = mapped_column(JSON, default=dict)
error_stage: Mapped[str | None] = mapped_column(String(32))
error_detail: Mapped[str | None] = mapped_column(Text)
credits_used: Mapped[float] = mapped_column(Float, default=0.0)
created_at: Mapped[datetime] = mapped_column(DateTime, server_default=func.now())
updated_at: Mapped[datetime] = mapped_column(DateTime, server_default=func.now(),
onupdate=func.now())
class PosterAliveTask(TaskBase):
"""포스터 한 장 → 8초 숏폼"""
__tablename__ = "poster_alive_task"
state: Mapped[PosterAliveState] = mapped_column(
state_column(PosterAliveState), default=PosterAliveState.DETECT, index=True)
# 검수 게이트를 건너뛰고 끝까지 돌린다
skip_review: Mapped[bool] = mapped_column(Boolean, default=False)
# 입력
poster_url: Mapped[str] = mapped_column(String(URL_LENGTH))
poster_width: Mapped[int | None] = mapped_column(Integer)
poster_height: Mapped[int | None] = mapped_column(Integer)
# i2v가 명시적으로 거절하는 하드 게이트
is_low_resolution: Mapped[bool] = mapped_column(Boolean, default=False)
# ① detect
detect_regions: Mapped[dict | None] = mapped_column(JSON) # Regions
detect_grid_url: Mapped[str | None] = mapped_column(String(URL_LENGTH))
detect_check_url: Mapped[str | None] = mapped_column(String(URL_LENGTH))
# ② narration_text
narration_lines: Mapped[list | None] = mapped_column(JSON) # 3문장
narration_voice: Mapped[str | None] = mapped_column(String(32))
bgm_style: Mapped[dict | None] = mapped_column(JSON) # BgmStyle
poster_metadata: Mapped[dict | None] = mapped_column(JSON) # NarrationMetadata
# ③ motion
motion_plan: Mapped[dict | None] = mapped_column(JSON) # MotionPlan
# ④ tts
narration_timeline: Mapped[dict | None] = mapped_column(JSON) # NarrationTimeline
narration_audio_url: Mapped[str | None] = mapped_column(String(URL_LENGTH))
# ⑤ bgm
bgm_audio_url: Mapped[str | None] = mapped_column(String(URL_LENGTH))
bgm_task_id: Mapped[str | None] = mapped_column(String(64))
bgm_target_seconds: Mapped[int | None] = mapped_column(Integer)
bgm_duration: Mapped[float | None] = mapped_column(Float)
# ⑥ i2v
clip_url: Mapped[str | None] = mapped_column(String(URL_LENGTH))
clip_model: Mapped[str | None] = mapped_column(String(64))
clip_credits: Mapped[float | None] = mapped_column(Float)
title_gate: Mapped[dict | None] = mapped_column(JSON) # TitleGateVerdict
# ⑦ render
video_url: Mapped[str | None] = mapped_column(String(URL_LENGTH))
thumbnail_url: Mapped[str | None] = mapped_column(String(URL_LENGTH))
video_duration: Mapped[float | None] = mapped_column(Float)
video_frames: Mapped[int | None] = mapped_column(Integer)
video_width: Mapped[int | None] = mapped_column(Integer)
video_height: Mapped[int | None] = mapped_column(Integer)
class PlayreelTask(TaskBase):
"""공연 상품페이지 URL → 30초 롱컷"""
__tablename__ = "playreel_task"
state: Mapped[PlayreelState] = mapped_column(
state_column(PlayreelState), default=PlayreelState.FETCH, index=True)
# 검수 대기 중인 게이트. 승인하면 비워짐
gate: Mapped[str | None] = mapped_column(String(32))
# 최종 승인마다 올라감. 재작업은 새 버전이 됨
version: Mapped[int] = mapped_column(Integer, default=0)
# 입력
source_url: Mapped[str] = mapped_column(String(URL_LENGTH))
goods_id: Mapped[str] = mapped_column(String(32), index=True)
slug: Mapped[str] = mapped_column(String(64))
# ① fetch
nol_meta: Mapped[dict | None] = mapped_column(JSON) # NolMeta
poster_url: Mapped[str | None] = mapped_column(String(URL_LENGTH))
# 750px이면 검수 화면이 저해상 경고를 띄운다
poster_width: Mapped[int | None] = mapped_column(Integer)
poster_height: Mapped[int | None] = mapped_column(Integer)
detail_image_urls: Mapped[list | None] = mapped_column(JSON)
fetch_failed_urls: Mapped[list | None] = mapped_column(JSON)
# ② split
# [{name, source, index, y0, y1, tag, url, selected}]
detail_sections: Mapped[list | None] = mapped_column(JSON)
section_sheet_url: Mapped[str | None] = mapped_column(String(URL_LENGTH))
# ③ upscale
upscaled_poster_url: Mapped[str | None] = mapped_column(String(URL_LENGTH))
upscale_credits: Mapped[int | None] = mapped_column(Integer)
upscale_variant: Mapped[str | None] = mapped_column(String(64))
# ④ analyze
analysis_grid_url: Mapped[str | None] = mapped_column(String(URL_LENGTH))
movable_elements: Mapped[list | None] = mapped_column(JSON) # [MotionChoice]
fixed_layers: Mapped[list | None] = mapped_column(JSON) # [FixedLayer]
i2v_model: Mapped[str | None] = mapped_column(String(64))
has_ip_risk: Mapped[bool | None] = mapped_column(Boolean)
has_qr: Mapped[bool | None] = mapped_column(Boolean)
# ⑤ motion
motion_plan: Mapped[dict | None] = mapped_column(JSON) # MotionPlan
# ⑥ narration
cast_extraction: Mapped[dict | None] = mapped_column(JSON) # CastExtraction
narration_lines: Mapped[list | None] = mapped_column(JSON) # [{slot, text}]
# ⑦ tts
narration_timeline: Mapped[dict | None] = mapped_column(JSON) # NarrationTimeline
narration_audio_url: Mapped[str | None] = mapped_column(String(URL_LENGTH))
narration_voice_id: Mapped[str | None] = mapped_column(String(64))
# ⑧ bgm
bgm_audio_url: Mapped[str | None] = mapped_column(String(URL_LENGTH))
bgm_task_id: Mapped[str | None] = mapped_column(String(64))
bgm_target_seconds: Mapped[int | None] = mapped_column(Integer)
bgm_duration: Mapped[float | None] = mapped_column(Float)
# ⑨ i2v
clip_url: Mapped[str | None] = mapped_column(String(URL_LENGTH))
clip_frames_url: Mapped[str | None] = mapped_column(String(URL_LENGTH))
clip_model: Mapped[str | None] = mapped_column(String(64))
clip_credits: Mapped[float | None] = mapped_column(Float)
title_gate: Mapped[dict | None] = mapped_column(JSON) # TitleGateVerdict
# ⑩ hybrid
layer_config: Mapped[dict | None] = mapped_column(JSON) # LayerConfig
hybrid_clip_url: Mapped[str | None] = mapped_column(String(URL_LENGTH))
hybrid_proof_url: Mapped[str | None] = mapped_column(String(URL_LENGTH))
hybrid_frames: Mapped[int | None] = mapped_column(Integer)
hybrid_duration: Mapped[float | None] = mapped_column(Float)
# ⑪ compose
video_url: Mapped[str | None] = mapped_column(String(URL_LENGTH))
scene_plan: Mapped[dict | None] = mapped_column(JSON) # LongcutPlan
compose_proof_url: Mapped[str | None] = mapped_column(String(URL_LENGTH))
video_duration: Mapped[float | None] = mapped_column(Float)
video_frames: Mapped[int | None] = mapped_column(Integer)
# ⑫ review
review_checks: Mapped[list | None] = mapped_column(JSON) # [{key, label, ok}]

View File

@ -25,7 +25,7 @@ def main() -> None:
config = LayerConfig.model_validate_json(config_path.read_text(encoding="utf-8"))
print(f"config {config_path.name} · mask={config.mask.mode}"
f" · 고정사각 {len(config.fixed_rects)} · 글리프 {len(config.glyphs)}"
f" · 펀치 {len(config.punches)} · 정합 {config.hybrid.register}")
f" · 펀치 {len(config.punches)} · 정합 {config.hybrid.register_frames}")
result = render_hybrid(poster_path, clip_path.read_bytes(), config, preview=True)

View File

@ -40,10 +40,24 @@ def public_url(path: str) -> str:
return f"{settings.azure_blob_base_url.rstrip('/')}/{path.lstrip('/')}"
def upload_url(path: str) -> str:
def sas_token() -> str:
# SAS 토큰이 따옴표나 ? 로 감싸여 오는 경우가 있다
token = settings.azure_blob_sas_token.strip("?'\"")
return f"{public_url(path)}?{token}"
return settings.azure_blob_sas_token.strip("?'\"")
def upload_url(path: str) -> str:
return f"{public_url(path)}?{sas_token()}"
def with_sas(url: str) -> str:
"""컨테이너가 비공개면 읽기에도 토큰이 필요함
우리가 올린 URL일 때만 붙임
"""
token = sas_token()
base = settings.azure_blob_base_url.rstrip("/")
if token and base and url.startswith(base) and "?" not in url:
return f"{url}?{token}"
return url
async def upload_bytes(data: bytes, path: str, content_type: str | None = None) -> str:
@ -61,3 +75,11 @@ async def upload_bytes(data: bytes, path: str, content_type: str | None = None)
async def upload_file(file_path: Path, path: str, content_type: str | None = None) -> str:
return await upload_bytes(file_path.read_bytes(), path,
content_type or guess_content_type(file_path.name))
async def download_bytes(url: str) -> bytes:
"""upload_bytes가 돌려준 URL을 그대로 받아 내용을 가져온다."""
response = await get_client().get(with_sas(url))
if response.status_code != 200:
raise RuntimeError(f"blob 다운로드 실패 {response.status_code}: {url}")
return response.content

View File

@ -10,4 +10,5 @@
- `models/` — API 관련 스키마.
- `answers/` — LLM 응답용 스키마.
- `tables/` — SQLAlchemy ORM 테이블.
- `models`가 상위다. `answers``models`를 참조하는 것은 허용, 역방향은 금지.