playreel/backend/pipelines/gate.py

"""사람이 검토하는 지점의 정의와 검수 화면 재료
게이트는 특정 단계가 끝난 직후에 섬
페이로드 모양은 프론트가 그리는 것과 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]:
    """컬럼으로 확인되는 것만 낸다 — 완성본을 다시 열어 본 결과가 아니다.

    제목 훼손 검사는 넣지 않는다. 비교 기준이 되는 좌표를 detect가 만드는데
    playreel에는 그 단계가 없어 검사 자체가 돌지 않는다.
    """
    plan = LongcutPlan.model_validate(task.scene_plan or {"scenes": []})
    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.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": "배경음악",
         "ok": bool(task.bgm_audio_url)},
    ]


def final_review(task: PlayreelTask) -> dict:
    return {
        "video_url": task.video_url or "",
        # hybrid 시트는 앞 8초짜리라 완성본 검수 자리에 대신 세우지 않는다
        "proof_url": task.compose_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"])]