"""s25 — main v16 팩(발명 경계) 오프라인 canary (2026-07-08). 대상 = prompts/_base/{shot_extract,shot_staging}/16.202607080045 (신규 팩, v15 원본 불변). production 코드 경로를 그대로 차용해 같은 입력(E2E 체크포인트) 으로 재저작하고, 기존 체크포인트 산출(v15 계보)과 대조한다. - staging: app.modules.pipeline.shot_staging.run_shot_staging **직접 호출** (같은 시스템/스키마/검증/재시도 경로, load_prompt 가 v16 latest 해석) — 대상 씬만 필터해 배치 최소화. 체크포인트 write 0 (반환 dict 만 저장). - extract: ShotExtractStep._call_bundle 의 조립을 재현(시각규칙+t2i_context 래퍼+beat+원문+인물 enum+scene_index enum 주입, ref_section 은 빈 값 — caveat: production 은 이전 번들 누적 컨텍스트 전달, canary 는 독립). - html: v16_canary.html — 샷별 old(v15 계보 체크포인트) vs new(v16) 대조. 판정(발명 소멸 여부)=사용자 육안 + 텍스트 대조. 코드는 조립·저장만. 사용: backend/.venv/bin/python s25_main_v16_canary.py --only all|staging|extract|html 산출: plans/main_v16_canary.json, v16_canary.html (8897) DB write 0, 커밋 금지(scratchpad). """ import argparse import copy import html as _html import json import sys import time from pathlib import Path sys.path.insert(0, str(Path(__file__).parent)) import forest_lib as F # noqa: E402 from app.modules.llm.llm_client import call_structured # noqa: E402 from app.modules.pipeline.shot_staging import run_shot_staging # noqa: E402 from app.modules.prompt_loader import load_prompt, load_schema # noqa: E402 PAGE = F.EXP / "v16_canary.html" STAGING_SCENES = [4, 11, 13, 17] # 발명 5건 소재 씬 (45m/철제모순/안개/앰버/wet) EXTRACT_SCENES = [13, 17] # shot_extract 단계 안개 발명 소재 씬 def _filter_scenes(data, key, scenes): out = copy.deepcopy(data) out[key] = [s for s in out.get(key) or [] if s.get("scene_index") in scenes] return out def run_staging_canary(plan): se = F.load_step("shot_extract") sel = F.load_step("shot_selection") old = {(s["scene_index"], s["shot_index"]): s for s in F.load_step("shot_staging")["shots"]} new_result = run_shot_staging( shot_extract_data=_filter_scenes(se, "scenes", STAGING_SCENES), shot_selection_data=_filter_scenes(sel, "scenes", STAGING_SCENES), scene_save_data=F.load_step("scene_save"), entity_merge_data=F.load_step("entity_merge"), vwr_data=F.load_step("visual_world_rules"), camera_flow_data=F.load_step("scene_camera_flow"), ) rows = [] for s in new_result["shots"]: k = (s["scene_index"], s["shot_index"]) o = old.get(k) assert o is not None, f"기존 staging 없음: {k}" rows.append({"scene_index": k[0], "shot_index": k[1], "old": o, "new": s}) plan["staging"] = {"scenes": STAGING_SCENES, "rows": rows, "pack": "16.202607080045"} F.save_plan("main_v16_canary", plan) print(f"staging canary: {len(rows)} shots 재저작 (씬 {STAGING_SCENES})") F.runlog({"kind": "llm", "step": "s25_staging_canary", "shots": len(rows)}) return plan def run_extract_canary(plan): beats_by = {s["scene_index"]: s.get("beats", []) for s in F.load_step("beat_extract")["scenes"]} segs = {s["scene_index"]: s for s in F.load_step("scene_save")["segments"]} vwr = F.load_step("visual_world_rules") old_scenes = {s["scene_index"]: s for s in F.load_step("shot_extract")["scenes"]} chars_cp = F.load_step("entity_character_list") character_names = (chars_cp.get("characters") or chars_cp.get("names") or []) if character_names and isinstance(character_names[0], dict): character_names = [c.get("name", "") for c in character_names] # visual_rules 조립 — ShotExtractStep.run 과 동일 (director_notes 우선) visual_rules = "" notes = vwr.get("director_notes", []) if notes: visual_rules = ("\n[물리적 존재 판단 기준 — description에는 카메라에" " 보이는 인물/바디만 묘사]\n" + "\n".join(f"- {n}" for n in notes)) ctx = vwr.get("t2i_context", "") if ctx: visual_rules = (f"\n[시각적 배경 — 시나리오를 이미지화하는 작업입니다." f" description 묘사 시 아래 맥락을 반영하세요]\n{ctx}\n" + visual_rules) system = load_prompt("shot_extract", "system") user_template = load_prompt("shot_extract", "user") schema = load_schema("shot_extract", "shot_schema") def _format_beats(beats): if not beats: return " (beat 없음 — 상황 묘사 기반으로 Shot 추출)" return "\n".join( f" Beat {b['beat_index']}: [{b['change_type']}] " f"{b['before_state']} → {b['after_state']}" for b in beats) character_list_section = "" if character_names: character_list_section = ( "[인물 목록 — characters에는 반드시 이 목록의 이름만 사용]\n" + "\n".join(f"- {n}" for n in character_names) + "\n\n") scenes_parts = [] for si in EXTRACT_SCENES: seg = segs[si] heading = seg.get("heading") or seg.get("scene_heading") or "" scenes_parts.append( f"--- Scene {si}: {heading} ---\n" f"[Beats]\n{_format_beats(beats_by.get(si))}\n\n" f"[원문]\n{seg['text']}") scenes_section = "\n\n".join(scenes_parts) if visual_rules: scenes_section = visual_rules + "\n\n" + scenes_section if "{character_list_section}" in user_template: user_prompt = user_template.format( ref_section="", scenes_section=scenes_section, character_list_section=character_list_section) else: user_prompt = user_template.format( ref_section="", scenes_section=scenes_section) call_schema = copy.deepcopy(schema) call_schema["properties"]["scenes"]["items"]["properties"]["scene_index"] = { "type": "integer", "enum": EXTRACT_SCENES} t0 = time.monotonic() result = call_structured( step="shot_extract", system_prompt=system, user_prompt=user_prompt, response_schema=call_schema, schema_name="s25_extract_canary") F.runlog({"kind": "llm", "step": "s25_extract_canary", "dur_s": round(time.monotonic() - t0, 1), "system_prompt": system, "user_prompt": user_prompt}) rows = [] for s in result.get("scenes", []): si = s.get("scene_index") rows.append({"scene_index": si, "old_shots": (old_scenes.get(si) or {}).get("shots", []), "new_shots": s.get("shots", [])}) plan["extract"] = {"scenes": EXTRACT_SCENES, "rows": rows, "pack": "16.202607080045", "caveat_ko": "production 은 이전 번들 누적 컨텍스트를" " 전달하나 canary 는 독립 호출"} F.save_plan("main_v16_canary", plan) print(f"extract canary: 씬 {[r['scene_index'] for r in rows]} 재추출") return plan def run_chain_canary(plan): """v16 체인 재현: extract canary 의 새 desc 를 staging 입력으로 — 구 desc 발명(예: 안개)이 체인 차원에서 끊기는지 검증. 대상=EXTRACT_SCENES.""" assert plan.get("extract"), "extract canary 먼저" new_by_scene = {r["scene_index"]: r["new_shots"] for r in plan["extract"]["rows"]} se = _filter_scenes(F.load_step("shot_extract"), "scenes", EXTRACT_SCENES) for s in se["scenes"]: s["shots"] = copy.deepcopy(new_by_scene[s["scene_index"]]) sel = _filter_scenes(F.load_step("shot_selection"), "scenes", EXTRACT_SCENES) # 새 추출의 shot_index 구성이 구 선택과 다를 수 있음 — 존재하는 index 로 교집합 avail = {s["scene_index"]: {sh.get("shot_index") for sh in s["shots"]} for s in se["scenes"]} for s in sel["scenes"]: keep = [i for i in (s.get("selected_shot_indices") or []) if i in avail.get(s["scene_index"], set())] s["selected_shot_indices"] = keep or sorted( avail.get(s["scene_index"], set()))[:3] result = run_shot_staging( shot_extract_data=se, shot_selection_data=sel, scene_save_data=F.load_step("scene_save"), entity_merge_data=F.load_step("entity_merge"), vwr_data=F.load_step("visual_world_rules"), camera_flow_data=F.load_step("scene_camera_flow"), ) desc_by = {(s["scene_index"], sh.get("shot_index")): sh.get("description") for s in se["scenes"] for sh in s["shots"]} rows = [{"scene_index": s["scene_index"], "shot_index": s["shot_index"], "input_desc": desc_by.get((s["scene_index"], s["shot_index"])), "new": s} for s in result["shots"]] plan["chain"] = {"scenes": EXTRACT_SCENES, "rows": rows, "note_ko": "입력 desc=v16 extract canary 산출(구 체크포인트" " 아님) — v16 체인 전체의 발명 차단 검증"} F.save_plan("main_v16_canary", plan) print(f"chain canary: {len(rows)} shots (씬 {EXTRACT_SCENES})") return plan def _esc(t): return _html.escape(str(t or "")) def _kbe(els): return "
".join( f"· {_esc(e.get('element'))} — state: {_esc(e.get('state'))}" for e in els or []) def build_page(plan): secs = [] st = plan.get("staging") if st: for r in st["rows"]: o, n = r["old"], r["new"] secs.append(f"""

staging — S{r['scene_index']} Shot{r['shot_index']}

old (v15 계보 체크포인트)new (v16 발명 경계 팩)
camera_direction{_esc(o.get('camera_direction'))} {_esc(n.get('camera_direction'))}
lighting_mood{_esc(o.get('lighting_mood'))} {_esc(n.get('lighting_mood'))}
key_bg_elements{_kbe(o.get('key_bg_elements'))} {_kbe(n.get('key_bg_elements'))}
""") ch = plan.get("chain") if ch: for r in ch["rows"]: n = r["new"] secs.append(f"""

chain (v16 extract desc → v16 staging) — S{r['scene_index']} Shot{r['shot_index']}

입력 desc (v16){_esc(r.get('input_desc'))}
camera_direction{_esc(n.get('camera_direction'))}
lighting_mood{_esc(n.get('lighting_mood'))}
key_bg_elements{_kbe(n.get('key_bg_elements'))}

{_esc(ch.get('note_ko'))}

""") ex = plan.get("extract") if ex: for r in ex["rows"]: old_rows = "".join( f"Shot{s.get('shot_index')}" f"{_esc(s.get('description'))}" for s in r["old_shots"]) new_rows = "".join( f"Shot{s.get('shot_index')}" f"{_esc(s.get('description'))}" for s in r["new_shots"]) secs.append(f"""

shot_extract — S{r['scene_index']}

old (v15 계보, {len(r['old_shots'])} shots)

{old_rows}

new (v16, {len(r['new_shots'])} shots)

{new_rows}

caveat: {_esc(ex.get('caveat_ko'))}

""") PAGE.write_text(f""" s25 — main v16 팩 canary (발명 경계)

s25 — main v16 팩 오프라인 canary: shot_staging / shot_extract 발명 경계 (old=v15 계보 체크포인트 vs new=v16 재저작)

같은 E2E 입력, production 호출 경로 그대로(load_prompt latest=v16). 관찰 포인트: 45m 수치·안개/빗물·sickly amber·wet/damp·무근거 재질이 new 에서 사라지는지 + 연출 언어(카메라/구도)는 유지되는지. 체크포인트 write 0.

{''.join(secs)} """, encoding="utf-8") print(f"page -> {PAGE}") def main(): ap = argparse.ArgumentParser() ap.add_argument("--only", default="all", choices=["all", "staging", "extract", "chain", "html"]) args = ap.parse_args() try: plan = F.load_plan("main_v16_canary") except Exception: plan = {} if args.only in ("all", "staging"): plan = run_staging_canary(plan) build_page(plan) if args.only in ("all", "extract"): plan = run_extract_canary(plan) build_page(plan) if args.only == "chain": plan = run_chain_canary(plan) build_page(plan) build_page(plan) F.runlog({"kind": "stage", "stage": "s25_main_v16_canary", "done": args.only}) if __name__ == "__main__": main()