diff --git a/backend/models/detail_section.py b/backend/models/detail_section.py index 89c0c50..b0469dd 100644 --- a/backend/models/detail_section.py +++ b/backend/models/detail_section.py @@ -11,7 +11,8 @@ class DetailSection: index: int # 그 원본 안에서의 순번 y0: int y1: int - image: Image.Image + # compose는 좌표만 쓴다 — 조각 그림 없이 만들어도 된다 + image: Image.Image | None = None tag: str = UNTAGGED @property diff --git a/backend/models/longcut.py b/backend/models/longcut.py index 7a8bfba..e132bba 100644 --- a/backend/models/longcut.py +++ b/backend/models/longcut.py @@ -11,8 +11,12 @@ class Scene(BaseModel): """나레이션 큐 하나에 대응하는 화면.""" cue: int kind: SceneKind - section: str | None = None # scroll일 때 쓸 섹션 이름 + section: str | None = None # scroll 시작점이 된 섹션 이름 tag: str | None = None # 그 섹션의 태그. 왜 골랐는지 남긴다 + # scroll은 조각이 아니라 롱이미지 위에서 일어난다. 아래 셋이 그 좌표다 + source: str | None = None # 훑을 롱이미지 (detail_01 …) + y0: float | None = None # 시작 위치. 롱이미지 높이 대비 0~1 + y1: float | None = None # 끝 위치 class LongcutPlan(BaseModel): diff --git a/backend/pipelines/gate.py b/backend/pipelines/gate.py index d1e9079..8153333 100644 --- a/backend/pipelines/gate.py +++ b/backend/pipelines/gate.py @@ -141,12 +141,15 @@ def final_checks(task: PlayreelTask) -> list[dict]: playreel에는 그 단계가 없어 검사 자체가 돌지 않는다. """ plan = LongcutPlan.model_validate(task.scene_plan or {"scenes": []}) - scroll_tags = {scene.tag for scene in plan.scenes if scene.kind == "scroll"} + scrolls = [scene for scene in plan.scenes if scene.kind == "scroll"] + scroll_tags = {scene.tag for scene in scrolls} return [ {"key": "hybrid", "label": "원본 글자 합성", "ok": bool(task.hybrid_clip_url)}, + # 씬 종류가 scroll이라고 적힌 것만으로는 부족하다 — 실제로 훑은 거리를 본다. + # 조각이 화면보다 짧아 한 픽셀도 안 움직인 채 통과한 적이 있다 {"key": "scroll", "label": "상세페이지 스크롤", - "ok": any(scene.kind == "scroll" for scene in plan.scenes)}, + "ok": any((scene.y1 or 0.0) > (scene.y0 or 0.0) for scene in scrolls)}, {"key": "schedule", "label": "캐스팅 스케줄 포함", "ok": REQUIRED_SECTION_TAG in scroll_tags}, {"key": "bgm", "label": "배경음악", diff --git a/backend/pipelines/playreel.py b/backend/pipelines/playreel.py index 4c6e54f..5e52cb0 100644 --- a/backend/pipelines/playreel.py +++ b/backend/pipelines/playreel.py @@ -55,6 +55,20 @@ async def load_sections(entries: list[dict]) -> list[DetailSection]: 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 @@ -234,7 +248,8 @@ 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) + sections = section_specs(selected) + details = await load_details(task) # 하이브리드 합성을 못 거친 클립은 제목이 흔들림 clip = await blob.download_bytes(task.hybrid_clip_url or task.clip_url) @@ -245,7 +260,8 @@ async def run_compose(session: AsyncSession, task: PlayreelTask) -> None: 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, preview=True) + 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, diff --git a/backend/services/compose_long.py b/backend/services/compose_long.py index 62a590a..de48b5b 100644 --- a/backend/services/compose_long.py +++ b/backend/services/compose_long.py @@ -1,8 +1,9 @@ """⑪ compose — 무빙포스터 훅 + 상세페이지 스크롤로 30초 롱컷을 조립한다. 씬 경계는 나레이션 큐다. 0초는 포스터 원본이라 첫 프레임이 곧 썸네일이 된다. -상세페이지는 잘라 붙이지 않고 롱이미지를 실제로 스크롤한다. -포스터·롱이미지는 세로 롱이라 9:16을 못 채워서 포스터 블러판 위에 얹는다. +상세페이지는 잘라 붙이지 않고 롱이미지를 실제로 스크롤한다 — 조각은 "어디부터 읽을지"를 +정하는 시작점일 뿐이고, 화면에 흐르는 것은 언제나 원본 롱이미지다. +포스터는 세로 롱이라 9:16을 못 채워서 포스터 블러판 위에 얹는다. 텍스트 오버레이는 넣지 않는다. 남는 글자는 엔드밴드뿐이다. """ @@ -12,7 +13,7 @@ from collections.abc import Iterator, Sequence from pathlib import Path import numpy as np -from PIL import Image, ImageDraw, ImageFilter +from PIL import Image, ImageDraw, ImageFilter, ImageOps from models.detail_section import DetailSection from models.longcut import EndBandInfo, LongcutPlan, LongcutResult, Scene @@ -30,9 +31,15 @@ DUCK_THRESHOLD = 0.03 DUCK_ATTACK_MS, DUCK_RELEASE_MS = 60, 500 BGM_FADE_OUT = 1.2 -# 큐 1초당 화면 높이의 몇 배를 훑을지. 원본은 사람이 구간을 골랐다(known_issue 참조). +# 큐 1초당 화면 높이의 몇 배를 훑을지. 5초 씬에 이미지 높이의 15~30%가 읽기 좋은 속도다 SCROLL_SCREENS_PER_SECOND = 0.35 +# 프레임보다 짧은 롱이미지를 늘릴 때 쓰는 값. 글자 없는 배경 띠를 거울 타일링한다 +INK_DEVIATION = 32 # 배경색에서 이만큼 벗어난 픽셀을 잉크로 본다 +INK_ROW_RATIO = 0.003 # 행의 잉크 비율이 이 미만이면 글자 없는 행 +EXTEND_FEATHER = 40 # 이음매를 이만큼 겹쳐 섞는다 +MIN_BLANK_BAND = 24 # 늘리기에 쓸 배경 띠의 최소 높이 + PROOF_COUNT = 8 # 완성본 전체에서 고르게 뽑는 시점 수 PROOF_SIZE = (180, 320) PROOF_GAP = 10 @@ -46,7 +53,8 @@ NOL_WORDMARK = ASSET_DIR / "nol_wordmark_white.png" # 큐 순서대로 채울 씬. 앞 둘은 클립, 마지막은 포스터, 나머지는 이 태그 우선순위로 고른다. CLIP_CUES = 2 -SCROLL_TAG_PRIORITY = ("synopsis", "still", "cast", "schedule", "discount", "event", "notice") +# 공지·교차배너는 스크롤 시작점으로 쓰지 않는다 — 약관을 읽히려고 만드는 영상이 아니다 +SCROLL_TAG_PRIORITY = ("synopsis", "still", "cast", "schedule", "discount", "event", "info") REQUIRED_TAGS = ("schedule",) # 캐스팅 스케줄은 예매 전에 확인하는 정보라 빠지면 안 된다 @@ -61,20 +69,78 @@ def blurred_bg(poster: Image.Image) -> Image.Image: return Image.eval(blurred, lambda v: int(v * 0.45)) -def trim_blank(image: Image.Image, threshold: int = 235) -> Image.Image: - """상하단 흰·균일 여백 제거. split이 남긴 12px 여백이 화면에 흰 줄로 보인다.""" - pixels = np.asarray(image.convert("RGB")).astype(int) - white = (pixels.min(axis=2) > threshold).mean(axis=1) > 0.98 - uniform = pixels.std(axis=1).max(axis=1) < 4 - content = np.where(~(white | uniform))[0] - if len(content) == 0: +def ink_rows(image: Image.Image) -> np.ndarray: + """행마다 글자·그림이 있는지. 배경 중앙값에서 벗어난 픽셀의 비율로 본다.""" + pixels = np.asarray(image.convert("L")).astype(np.int16) + background = np.median(pixels) + return (np.abs(pixels - background) > INK_DEVIATION).mean(axis=1) > INK_ROW_RATIO + + +def blank_band(image: Image.Image) -> tuple[int, int]: + """늘리기에 쓸, 글자 없는 가장 긴 행 구간. 없으면 맨 아래 띠를 쓴다.""" + blank = ~ink_rows(image) + best = (0, 0) + start = None + for y, empty in enumerate(np.append(blank, False)): + if empty and start is None: + start = y + elif not empty and start is not None: + if y - start > best[1] - best[0]: + best = (start, y) + start = None + if best[1] - best[0] >= MIN_BLANK_BAND: + return best + return max(0, image.height - MIN_BLANK_BAND), image.height + + +def extend_to_height(image: Image.Image, height: int = H) -> Image.Image: + """프레임보다 짧은 롱이미지를 배경 띠의 거울 타일링으로 늘린다. + + 짧다고 블러 배경 위에 얹으면 화면 절반이 포스터로 남는다. 배경만 이어 붙여 + 풀프레임을 채우고, 이음매는 페더로 지운다. + """ + if image.height >= height: return image - return image.crop((0, int(content[0]), image.width, int(content[-1]) + 1)) + + top, bottom = blank_band(image) + band = image.crop((0, top, image.width, bottom)) + canvas = Image.new("RGB", (image.width, height)) + canvas.paste(image, (0, 0)) + + y, flipped = image.height, True + while y < height: + tile = ImageOps.flip(band) if flipped else band + tile = tile.crop((0, 0, tile.width, min(tile.height, height - y))) + canvas.paste(tile, (0, y)) + _feather_seam(canvas, y, image.width) + y += tile.height + flipped = not flipped + return canvas -def fit_width(image: Image.Image) -> Image.Image: - image = trim_blank(image) - return image.resize((W, max(1, round(image.height * W / image.width))), Image.LANCZOS) +def _feather_seam(canvas: Image.Image, seam: int, width: int, + feather: int = EXTEND_FEATHER) -> None: + """이음매 위아래를 선형 혼합해 가로줄이 보이지 않게 한다.""" + top, bottom = max(0, seam - feather), min(canvas.height, seam + feather) + if bottom - top < 2: + return + patch = np.asarray(canvas.crop((0, top, width, bottom))).astype(np.float32) + blurred = np.asarray( + canvas.crop((0, top, width, bottom)).filter(ImageFilter.GaussianBlur(6)) + ).astype(np.float32) + weight = 1.0 - np.abs(np.linspace(-1.0, 1.0, bottom - top)) + mixed = patch + (blurred - patch) * weight[:, None, None] + canvas.paste(Image.fromarray(mixed.clip(0, 255).astype(np.uint8)), (0, top)) + + +def to_surface(image: Image.Image) -> Image.Image: + """롱이미지를 폭 1080에 맞추고 프레임 높이까지 늘린다. + + 자르지 않는다 — 섹션 y좌표가 이 이미지 기준이라 위를 깎으면 시작점이 어긋난다. + """ + scaled = image.convert("RGB").resize( + (W, max(1, round(image.height * W / image.width))), Image.LANCZOS) + return extend_to_height(scaled, H) def still_scene(still: Image.Image, progress: float, index: int) -> Image.Image: @@ -95,15 +161,17 @@ def still_scene(still: Image.Image, progress: float, index: int) -> Image.Image: return frame -def scroll_scene(image: Image.Image, background: Image.Image, progress: float, - duration: float) -> Image.Image: - """읽는 속도로 훑는다. 이미지가 길면 위쪽만 보여준다.""" - span = max(0, image.height - H) - reach = min(span, round(duration * SCROLL_SCREENS_PER_SECOND * H)) - top = int(reach * ease_sine(progress)) - frame = background.copy() - frame.paste(image.crop((0, top, W, min(image.height, top + H))), (0, 0)) - return frame +def scroll_top(surface: Image.Image, fraction: float) -> int: + """이미지 높이 대비 비율을 실제 crop 시작 y로. 끝을 넘지 않게 묶는다.""" + span = max(0, surface.height - H) + return min(max(0, round(surface.height * fraction)), span) + + +def scroll_scene(surface: Image.Image, progress: float, y0: float, y1: float) -> Image.Image: + """롱이미지를 y0~y1 구간까지 훑는다. 프레임을 가득 채운다 — 배경이 비지 않는다.""" + start, end = scroll_top(surface, y0), scroll_top(surface, y1) + top = start + round((end - start) * ease_sine(progress)) + return surface.crop((0, top, W, top + H)) def poster_scene(poster: Image.Image, background: Image.Image, progress: float) -> Image.Image: @@ -149,10 +217,19 @@ def end_band(frame: Image.Image, info: EndBandInfo, seconds: float, draw.text((60, top + offset + slide), text, font=font(weight, size), fill=fill) -def build_scene_plan(sections: Sequence[DetailSection], cue_count: int) -> LongcutPlan: - """큐 앞 둘은 클립, 마지막은 포스터, 사이는 태그 우선순위로 채운다.""" +def cue_spans(timeline: NarrationTimeline) -> list[tuple[float, float]]: + """큐마다 (시작, 끝). 0번은 0초에서 시작해야 첫 프레임이 포스터가 된다.""" + starts = [0.0] + [cue.start for cue in timeline.cues[1:]] + ends = [cue.start for cue in timeline.cues[1:]] + [timeline.total] + return list(zip(starts, ends, strict=True)) + + +def scroll_starts(sections: Sequence[DetailSection], + surfaces: dict[str, Image.Image]) -> list[DetailSection]: + """훑기 시작점이 될 섹션을 태그 우선순위대로. 롱이미지가 없는 섹션은 뺀다.""" + usable = [section for section in sections if section.source in surfaces] by_tag: dict[str, list[DetailSection]] = {} - for section in sections: + for section in usable: by_tag.setdefault(section.tag, []).append(section) ordered: list[DetailSection] = [] @@ -162,18 +239,48 @@ def build_scene_plan(sections: Sequence[DetailSection], cue_count: int) -> Longc for section in by_tag.get(tag, []): if section not in ordered: ordered.append(section) + return ordered - scenes = [] - scroll_index = 0 - for cue in range(cue_count): + +def scroll_window(section: DetailSection, source: Image.Image, surface: Image.Image, + seconds: float) -> tuple[float, float]: + """섹션이 시작점, 큐 길이가 훑는 거리. 둘 다 롱이미지 높이 대비 비율로 돌려준다.""" + scale = W / source.width + start = round(section.y0 * scale) + reach = round(seconds * SCROLL_SCREENS_PER_SECOND * H) + return start / surface.height, (start + reach) / surface.height + + +def build_scene_plan(sections: Sequence[DetailSection], spans: Sequence[tuple[float, float]], + sources: dict[str, Image.Image], + surfaces: dict[str, Image.Image]) -> LongcutPlan: + """큐 앞 둘은 클립, 마지막은 포스터, 사이는 롱이미지를 훑는다.""" + ordered = scroll_starts(sections, surfaces) + fallback = max(surfaces, key=lambda name: surfaces[name].height, default=None) + + scenes: list[Scene] = [] + index = 0 + cursor = 0.0 # 폴백이 이어 읽을 지점 + for cue, (start, end) in enumerate(spans): if cue < CLIP_CUES: scenes.append(Scene(cue=cue, kind="clip")) - elif cue == cue_count - 1: + elif cue == len(spans) - 1: scenes.append(Scene(cue=cue, kind="poster")) - elif scroll_index < len(ordered): - section = ordered[scroll_index] - scenes.append(Scene(cue=cue, kind="scroll", section=section.name, tag=section.tag)) - scroll_index += 1 + elif index < len(ordered): + section = ordered[index] + y0, y1 = scroll_window(section, sources[section.source], + surfaces[section.source], end - start) + scenes.append(Scene(cue=cue, kind="scroll", section=section.name, + tag=section.tag, source=section.source, y0=y0, y1=y1)) + index, cursor = index + 1, y1 + elif fallback is not None: + # 쓸 섹션이 떨어져도 포스터로 때우지 않는다. 앞 씬이 멈춘 자리에서 이어 읽어야 + # 같은 화면이 두 번 나오지 않는다 + reach = round((end - start) * SCROLL_SCREENS_PER_SECOND * H) + y1 = cursor + reach / surfaces[fallback].height + scenes.append(Scene(cue=cue, kind="scroll", source=fallback, + y0=cursor, y1=y1)) + cursor = y1 else: scenes.append(Scene(cue=cue, kind="poster")) return LongcutPlan(scenes=scenes) @@ -214,13 +321,11 @@ def mix_audio(narration_mp3: bytes, bgm_mp3: bytes | None, duration: float) -> n def compose_frames(timeline: NarrationTimeline, plan: LongcutPlan, poster: Image.Image, - clip: bytes, sections: dict[str, Image.Image], + clip: bytes, surfaces: dict[str, Image.Image], info: EndBandInfo, qr: Image.Image | None) -> Iterator[Image.Image]: background = blurred_bg(poster) - starts = [0.0] + [cue.start for cue in timeline.cues[1:]] - bounds = [cue.start for cue in timeline.cues[1:]] + [timeline.total] + spans = cue_spans(timeline) clip_frames = list(decode_frames(clip)) - scroll_images = {name: fit_width(image) for name, image in sections.items()} scenes = {scene.cue: scene for scene in plan.scenes} def frame_at(cue: int, at: float) -> Image.Image: @@ -231,23 +336,24 @@ def compose_frames(timeline: NarrationTimeline, plan: LongcutPlan, poster: Image frame = background.copy() frame.paste(fitted, (0, (H - fitted.height) // 2)) return frame - start, end = starts[cue], bounds[cue] + start, end = spans[cue] progress = (at - start) / max(end - start, 0.01) - if scene.kind == "scroll" and scene.section in scroll_images: - return scroll_scene(scroll_images[scene.section], background, progress, end - start) + if scene.kind == "scroll" and scene.source in surfaces: + return scroll_scene(surfaces[scene.source], progress, + scene.y0 or 0.0, scene.y1 or 0.0) return poster_scene(poster, background, progress) total_frames = round(timeline.total * FPS) for index in range(total_frames): at = index / FPS - cue = max(order for order in range(len(timeline.cues)) if at >= starts[order]) + cue = max(order for order, (start, _) in enumerate(spans) if at >= start) image = frame_at(cue, at) - remaining = (bounds[cue] - at) * FPS - if cue < len(timeline.cues) - 1 and remaining < CROSSFADE_FRAMES \ + remaining = (spans[cue][1] - at) * FPS + if cue < len(spans) - 1 and remaining < CROSSFADE_FRAMES \ and scenes[cue + 1].kind != "clip": image = Image.blend(image, frame_at(cue + 1, at), 1 - remaining / CROSSFADE_FRAMES) if scenes[cue].kind == "poster": - end_band(image, info, at - starts[cue], qr) + end_band(image, info, at - spans[cue][0], qr) yield image @@ -262,18 +368,21 @@ def proof_sheet(tiles: list[Image.Image]) -> Image.Image: def compose_longcut(poster: bytes | Path | Image.Image, clip: bytes, timeline: NarrationTimeline, narration_mp3: bytes, sections: Sequence[DetailSection], info: EndBandInfo, *, + details: dict[str, Image.Image] | None = None, bgm_mp3: bytes | None = None, qr: Image.Image | None = None, plan: LongcutPlan | None = None, preview: bool = False) -> LongcutResult: + """details는 source 이름(detail_01…) → 상세페이지 롱이미지. 스크롤은 이 위에서 일어난다.""" if isinstance(poster, Image.Image): source = poster.convert("RGB") else: source = Image.open(poster if isinstance(poster, Path) else io.BytesIO(poster)).convert("RGB") - plan = plan or build_scene_plan(sections, len(timeline.cues)) - used = {scene.section for scene in plan.scenes if scene.section} - images = {section.name: section.image for section in sections if section.name in used} + sources = dict(details or {}) + surfaces = {name: to_surface(image) for name, image in sources.items()} + spans = cue_spans(timeline) + plan = plan or build_scene_plan(sections, spans, sources, surfaces) total_frames = round(timeline.total * FPS) # 훅 구간만 보면 뒤쪽 스크롤과 엔드밴드를 검수할 수 없어 전체에서 고르게 뽑는다 @@ -283,7 +392,7 @@ def compose_longcut(poster: bytes | Path | Image.Image, clip: bytes, def frames() -> Iterator[Image.Image]: for index, frame in enumerate( - compose_frames(timeline, plan, source, clip, images, info, qr)): + compose_frames(timeline, plan, source, clip, surfaces, info, qr)): if index == 0: frame.save(thumbnail, "JPEG", quality=THUMBNAIL_QUALITY) if preview and index in proof_at: diff --git a/backend/tests/playreel/test_compose_long.py b/backend/tests/playreel/test_compose_long.py index d7ad463..10546bb 100644 --- a/backend/tests/playreel/test_compose_long.py +++ b/backend/tests/playreel/test_compose_long.py @@ -31,6 +31,16 @@ def load_sections(sections_dir: Path) -> list[DetailSection]: for item in json.loads(index_path.read_text(encoding="utf-8"))] +def load_details(product_dir: Path) -> dict[str, Image.Image]: + """상세페이지 롱이미지. 스크롤은 조각이 아니라 이 위에서 일어난다.""" + details = {} + for path in sorted((product_dir / "details").glob("detail_*.*")): + details[path.stem] = Image.open(path) + if not details: + raise SystemExit(f"{product_dir / 'details'} 비었음 — test_fetch_nol 을 먼저 돌릴 것") + return details + + 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) @@ -66,6 +76,7 @@ def main() -> None: product_dir / "upscaled.png", clip_path.read_bytes(), timeline, (product_dir / "narration.mp3").read_bytes(), load_sections(product_dir / "sections"), band_info(meta), + details=load_details(product_dir), bgm_mp3=bgm_path.read_bytes() if bgm_path.exists() else None, qr=Image.open(qr_path) if qr_path.exists() else None) @@ -75,7 +86,9 @@ def main() -> None: for scene in result.plan.scenes: label = f"{scene.section} [{scene.tag}]" if scene.section else "" - print(f" 큐{scene.cue} {scene.kind:7} {label}") + span = (f" {scene.source} {scene.y0:.3f}→{scene.y1:.3f}" + if scene.kind == "scroll" else "") + print(f" 큐{scene.cue} {scene.kind:7} {label}{span}") if not qr_path.exists(): print(f"※ {qr_path.name} 없음 — 엔드밴드에 QR을 안 넣는다") print(f"\n{result.frames}프레임 · {result.duration:.2f}초 · {len(result.video) / 1e6:.1f}MB")