199 lines
7.9 KiB
Python
199 lines
7.9 KiB
Python
"""사람이 검토하는 지점의 정의와 검수 화면 재료
|
|
게이트는 특정 단계가 끝난 직후에 섬
|
|
페이로드 모양은 프론트가 그리는 것과 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": []})
|
|
scroll_tags = {scene.tag for scene in plan.scenes if scene.kind == "scroll"}
|
|
return [
|
|
{"key": "hybrid", "label": "원본 글자 합성",
|
|
"ok": bool(task.hybrid_clip_url)},
|
|
{"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 "",
|
|
# 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"])]
|