"""공연 상품페이지 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 (THUMBNAIL, artifact_url, 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 section_specs(entries: list[dict]) -> list[DetailSection]: """compose용 — 조각 그림은 안 받는다. 스크롤은 롱이미지 위에서 일어나고 섹션은 어디부터 훑을지를 정하는 좌표로만 쓰인다""" return [DetailSection(source=entry["source"], index=entry["index"], y0=entry["y0"], y1=entry["y1"], tag=entry["tag"]) for entry in entries] async def load_details(task: PlayreelTask) -> dict[str, Image.Image]: """상세페이지 롱이미지. 이름은 split이 붙인 source와 같아야 한다""" return {f"detail_{order:02d}": Image.open(io.BytesIO(await blob.download_bytes(url))) for order, url in enumerate(task.detail_image_urls or [], 1)} def working_poster_url(task: PlayreelTask) -> str: """업스케일이 있으면 그것이 이후 단계의 원본이 됨""" return task.upscaled_poster_url or task.poster_url def thumbnail_url(task: PlayreelTask) -> str | None: """compose가 끝났으면 정지컷이 정해진 경로에 있다""" return artifact_url(PIPELINE, task.id, THUMBNAIL) if task.video_url else None 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 = section_specs(selected) details = await load_details(task) # 하이브리드 합성을 못 거친 클립은 제목이 흔들림 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), details=details, bgm_mp3=bgm, qr=Image.open(qr_path) if qr_path.exists() else None, preview=True) filename = f"{task.slug}_30_v{task.version + 1}.mp4" task.video_url = await store_bytes(PIPELINE, task.id, filename, result.video, download_as=filename) # 경로가 고정이라 URL은 남기지 않는다. 읽는 쪽이 thumbnail_url()로 만든다 await store_bytes(PIPELINE, task.id, THUMBNAIL, result.thumbnail) 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()