"""s26 — main wave2 팩(생산측 발명 경계) 오프라인 canary (2026-07-08). 대상 = prompts/_base/visual_world_rules/10.202607080130 (전역 룩: 색감·질감· 장르 톤만, 날씨·광원 전역 고정 금지 — 장소 사실은 씬별 경로가 처리) + prompts/_base/scene_camera_flow/2.202607080130 (카메라 기하만, 세계 사실 발명 금지). 원본 v9/v1 불변. - vwr: production extract_visual_rules **직접 호출** — 입력=E2E 체크포인트 (text_cleanup.cleaned_text + episode_summary + 기획서 supplement 를 VisualWorldRulesStep 과 동일 형식으로 재현). old(체크포인트 v9 계보) t2i_context 와 대조. ★기획서 visual_concepts 가 "안개 자욱한 섬"을 명시하므로, 관찰 포인트=안개·특정 광원이 '전역 룩'에서 빠지는지(색감· 질감·톤은 유지). - camflow: SceneCameraFlowStep._process_scene 조립 재현(S11, S17) — old flow(concrete stairwell/새벽 안개) 대조. 판정=사용자 육안+텍스트 대조. 코드=조립·저장만. 사용: backend/.venv/bin/python s26_wave2_canary.py --only all|vwr|camflow|html 산출: plans/wave2_canary.json, wave2_canary.html (8897) DB write 0, 커밋 금지(scratchpad). """ import argparse import copy import html as _html import time from pathlib import Path import sys sys.path.insert(0, str(Path(__file__).parent)) import forest_lib as F # noqa: E402 import json # noqa: E402 from app.modules.llm.llm_client import call_structured # noqa: E402 from app.modules.pipeline.visual_world_rules import extract_visual_rules # noqa: E402 from app.modules.prompt_loader import load_prompt, load_schema # noqa: E402 PAGE = F.EXP / "wave2_canary.html" CAMFLOW_SCENES = [11, 17] def _manifest(step): return json.loads((F.CP / step / "manifest.json").read_text()) def _planning_supplement(): """VisualWorldRulesStep 의 기획서 주입 형식 재현 (manifest 최상위 필드).""" m = _manifest("planning_doc_analysis") parts = [] for section, header in [("world_setting", "## 기획서: 세계관"), ("tone_mood", "## 기획서: 톤/분위기"), ("visual_concepts", "## 기획서: 비주얼 컨셉")]: val = m.get(section) or "" if val: parts.append(f"\n\n{header}\n\n{val}\n" f"") return "".join(parts) def run_vwr(plan): fulltext = F.load_step("text_cleanup")["cleaned_text"] assert fulltext, "cleaned_text 없음" summary = F.load_step("episode_summary")["summary"] supplement = _planning_supplement() t0 = time.monotonic() new = extract_visual_rules(fulltext=fulltext, episode_summary=summary + supplement) F.runlog({"kind": "llm", "step": "s26_vwr_canary", "dur_s": round(time.monotonic() - t0, 1)}) old = F.load_step("visual_world_rules") plan["vwr"] = { "pack": "10.202607080130", "planning_supplement": supplement, "old_t2i_context": old.get("t2i_context", ""), "new_t2i_context": new.get("t2i_context", ""), "old_rules_n": len(old.get("rules") or []), "new_rules_n": len(new.get("rules") or []), "new_full": new, } F.save_plan("wave2_canary", plan) print("vwr canary 완료") print(" old ctx:", plan["vwr"]["old_t2i_context"]) print(" new ctx:", plan["vwr"]["new_t2i_context"]) return plan def run_camflow(plan): segs = {s["scene_index"]: s for s in F.load_step("scene_save")["segments"]} sel_map = {s["scene_index"]: set(s.get("selected_shot_indices") or []) for s in F.load_step("shot_selection")["scenes"]} shots_by = {s["scene_index"]: s.get("shots", []) for s in F.load_step("shot_extract")["scenes"]} em = F.load_step("entity_merge") id_to_name = {e.get("short_id"): e.get("name", "") for et in ("characters", "locations", "props") for e in em.get(et, []) if e.get("short_id")} vis_by = {s.get("scene_index"): [id_to_name.get(v, "") for v in (s.get("visible_entity_ids") or []) if id_to_name.get(v)] for s in F.load_step("scene_director")["scenes"]} old_by = {s["scene_index"]: s for s in F.load_step("scene_camera_flow")["scenes"]} system = load_prompt("scene_camera_flow", "system") user_template = load_prompt("scene_camera_flow", "user") schema = load_schema("scene_camera_flow", "schema") rows = [] for si in CAMFLOW_SCENES: sel = sel_map.get(si, set()) selected = [sh for sh in shots_by[si] if sh.get("shot_index") in sel] unselected = [sh for sh in shots_by[si] if sh.get("shot_index") not in sel] seg = segs[si] user_prompt = user_template.format( scene_index=si, scene_heading=seg.get("heading", ""), scene_text=seg.get("text", ""), selected_shots_block="\n".join( f" Shot {sh.get('shot_index')}: {sh.get('description', '')}" for sh in selected) or " (없음)", unselected_shots_block="\n".join( f" Shot {sh.get('shot_index')}: {sh.get('description', '')}" for sh in unselected) or " (없음)", entities_block=", ".join(vis_by.get(si) or []) or "(엔티티 정보 없음)", ) call_schema = copy.deepcopy(schema) idxs = [sh["shot_index"] for sh in selected] if idxs: call_schema["properties"]["shot_assignments"]["items"][ "properties"]["shot_index"] = {"type": "integer", "enum": idxs} t0 = time.monotonic() new = call_structured(step="scene_camera_flow", system_prompt=system, user_prompt=user_prompt, response_schema=call_schema, schema_name=f"s26_camflow_{si}") F.runlog({"kind": "llm", "step": f"s26_camflow_{si}", "dur_s": round(time.monotonic() - t0, 1), "system_prompt": system, "user_prompt": user_prompt}) rows.append({"scene_index": si, "old": old_by.get(si), "new": new}) plan["camflow"] = {"pack": "2.202607080130", "rows": rows} F.save_plan("wave2_canary", plan) print(f"camflow canary S{si} 완료") return plan def _esc(t): return _html.escape(str(t or "")) def _flow_block(f): if not f: return "(없음)" lines = [f"flow_summary: {_esc(f.get('flow_summary'))}"] for st in f.get("flow_stages") or []: lines.append(f" [{st.get('stage_index')}] {_esc(st.get('stage_label'))}" f" | pos: {_esc(st.get('camera_position'))}" f" | focus: {_esc(st.get('visual_focus'))}" f" | next: {_esc(st.get('transition_to_next'))}") return "
".join(lines) def build_page(plan): secs = [] v = plan.get("vwr") if v: secs.append(f"""

visual_world_rules — t2i_context (전역 룩)

old (v9 계보 체크포인트) new (v10)
{_esc(v['old_t2i_context'])} {_esc(v['new_t2i_context'])}
기획서 supplement (입력 재현 — visual_concepts 에 '안개 자욱' 명시)
{_esc(v['planning_supplement'])}

rules 수: old {v['old_rules_n']} / new {v['new_rules_n']}

""") c = plan.get("camflow") if c: for r in c["rows"]: secs.append(f"""

scene_camera_flow — S{r['scene_index']}

old (v1 계보)new (v2)
{_flow_block(r['old'])}{_flow_block(r['new'])}
""") PAGE.write_text(f""" s26 — main wave2 canary (생산측 발명 경계)

s26 — main wave2 canary: visual_world_rules v10(전역 룩 경계) + scene_camera_flow v2(카메라 기하만)

관찰 포인트: ① t2i_context 에서 안개·특정 광원(sickly amber practical lights)의 '전역 사실' 선언이 빠지고 색감·질감·장르 톤은 유지되는지 (기획서의 섬 안개 컨셉은 장소 사실 — 씬별 경로가 처리) ② camera_flow 에서 concrete(S11)·새벽 안개(S17) 같은 세계 사실 어휘가 카메라 기하 서술로 대체되는지. 체크포인트 write 0.

{''.join(secs)} """, encoding="utf-8") print(f"page -> {PAGE}") def main(): ap = argparse.ArgumentParser() ap.add_argument("--only", default="all", choices=["all", "vwr", "camflow", "html"]) args = ap.parse_args() try: plan = F.load_plan("wave2_canary") except Exception: plan = {} if args.only in ("all", "vwr"): plan = run_vwr(plan) build_page(plan) if args.only in ("all", "camflow"): plan = run_camflow(plan) build_page(plan) build_page(plan) F.runlog({"kind": "stage", "stage": "s26_wave2_canary", "done": args.only}) if __name__ == "__main__": main()