"""[임시] 메인 v4 프로젝트의 selected shots을 final image로 렌더 (chain v6 배경 ref). == 절대 규칙 == - 메인 v4 프로젝트(/projects//...)는 READ-ONLY로만 접근 - 메인 코드/프롬프트/체크포인트/이미지 절대 수정 X - set_design은 항상 DISABLED — 코드/데이터/이미지 의존 X (사용자 명시) - 결과는 임시 디렉토리(--out-dir)에만 저장 - 기존 chain_*.py 등 코드도 손대지 않음 (재사용은 OK) == 데이터 흐름 == 1) 메인 v4 shot_selection/manifest.json → 모든 selected shots 2) (옵션) scene_index 필터 (예: 옥탑방 scope.scenes) 3) 각 selected shot에 대해: a. 메인 v4 scene_detail → t2i_variations[0].t2i_prompt + visible_entities + camera_effect b. **chain v6 plan**(chain_planning_gpt_v6/chain_structure.json)에서 shot_id로 노드 lookup c. **chain v6 image**(chain_render_gpt_v6/.png)를 배경 ref로 사용 - 없으면 누락 노드 명시 (Step C 필요) d. (선택) entity_t2i.completed → 인물(C##)/소품(P##) 영어 visual_traits 보강 e. gpt-image-2 image edit: chain v6 ref + scene_detail prompt → final shot 4) 결과는 out_dir에만 (file: shot_S##_Shot#.png) """ from __future__ import annotations import argparse import base64 import json import logging import os import re import sys from pathlib import Path BACKEND = Path(__file__).resolve().parent.parent sys.path.insert(0, str(BACKEND)) from dotenv import load_dotenv # noqa: E402 load_dotenv(BACKEND / ".env") from openai import OpenAI # noqa: E402 # 기존 chain_render_gpt 코드 재사용 (수정 X — import만) sys.path.insert(0, str(Path(__file__).resolve().parent)) from experiment_chain_structure_render_gpt import ( # noqa: E402 PROMPT_GEN_SYSTEM as CHAIN_PROMPT_GEN_SYSTEM, PROMPT_GEN_SCHEMA as CHAIN_PROMPT_GEN_SCHEMA, build_per_node_user_prompt as chain_build_per_node_user_prompt, call_gpt_text_json as chain_call_gpt_text_json, build_node_meta as chain_build_node_meta, collect_context as chain_collect_context, ) logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s") logger = logging.getLogger("v4_main_shot_render") # ====== Read-only main-project access helpers ====== def latest_manifest(dir_path: Path) -> Path | None: """디렉토리 안 manifest_*.json 중 _pre_review가 아닌 가장 최근 파일.""" if not dir_path.exists(): return None candidates: list[Path] = [] for f in dir_path.iterdir(): if not f.is_file(): continue if not f.name.startswith("manifest"): continue if not f.name.endswith(".json"): continue if "_pre_review" in f.name: continue candidates.append(f) if not candidates: return None return sorted(candidates, key=lambda p: p.name)[-1] def read_json(path: Path) -> dict: """READ-ONLY로만.""" return json.loads(path.read_text(encoding="utf-8")) def load_shot_selection(ckpt_root: Path) -> dict: p = ckpt_root / "shot_selection" / "manifest.json" if not p.exists(): p = latest_manifest(ckpt_root / "shot_selection") if not p: raise RuntimeError(f"shot_selection manifest 없음: {ckpt_root / 'shot_selection'}") d = read_json(p) return d.get("data", {}) def load_scene_detail(ckpt_root: Path) -> list[dict]: p = latest_manifest(ckpt_root / "scene_detail") if not p: raise RuntimeError(f"scene_detail manifest 없음: {ckpt_root / 'scene_detail'}") d = read_json(p) return d.get("data", {}).get("scenes", []) def load_entity_t2i(ckpt_root: Path) -> dict: p = latest_manifest(ckpt_root / "entity_t2i") if not p: return {} d = read_json(p) return d.get("data", {}).get("completed", {}) or {} def load_chain_v6_plan(planning_dir: Path) -> dict: p = planning_dir / "chain_structure.json" if not p.exists(): raise RuntimeError(f"chain v6 plan 없음: {p}") return read_json(p) def find_chain_v6_image(background_dir: Path, node_id: str) -> Path | None: """chain v6 image dir에서 노드 PNG. 없으면 None.""" p = background_dir / f"{node_id}.png" return p if p.exists() else None # ====== Korean → English entity-trait extraction ====== def english_traits_for_entity(entity_meta: dict) -> str: """entity_t2i.completed에서 영어 visual_traits만 합침.""" short_id = entity_meta.get("short_id") or "" desc_en = entity_meta.get("description") or "" traits = entity_meta.get("visual_traits") or [] parts = [f"[{short_id}]"] if desc_en and not re.search(r"[가-힣]", desc_en): parts.append(desc_en) if traits: en_traits = [t for t in traits if not re.search(r"[가-힣]", str(t))] if en_traits: parts.append("traits: " + ", ".join(en_traits)) return " — ".join(parts) # ====== gpt-image-2 image edit ====== def gpt_image_edit_with_ref(client: OpenAI, model: str, ref_path: Path, prompt: str, size: str, quality: str, out_path: Path) -> Path: logger.info("[image edit] %s + ref[%s] → %s (%s, %s)", model, ref_path.name, out_path.name, size, quality) with open(ref_path, "rb") as f: resp = client.images.edit( model=model, image=[f], prompt=prompt, size=size, quality=quality, n=1, ) b64 = resp.data[0].b64_json if not b64: raise RuntimeError(f"empty b64 for {out_path.name}") out_path.write_bytes(base64.b64decode(b64)) logger.info(" saved %d KB", out_path.stat().st_size // 1024) return out_path def main() -> int: p = argparse.ArgumentParser() p.add_argument("--main-project-root", required=True, help="메인 v4 프로젝트 root, 예: /Users/manta/.../projects/") p.add_argument("--episode-id", required=True, help="에피소드 UUID") p.add_argument("--planning-dir", required=True, help="chain v6 planning 디렉토리 (chain_structure.json 위치). " "chain v6 plan으로 shot→노드 매핑.") p.add_argument("--background-dir", required=True, help="chain v6 배경 image 디렉토리 (예: chain_render_gpt_v6). " "여기 PNG들이 ref로 사용됨. set_design 사용 X.") p.add_argument("--out-dir", required=True, help="결과 저장 (반드시 메인 프로젝트 외부)") p.add_argument("--scene-filter", help="콤마 구분 scene_index. 비우면 모든 selected.") p.add_argument("--shot-filter", help="콤마 구분 shot_id (예: 'S04_Shot1,S12_Shot4'). 비우면 모든 selected.") p.add_argument("--image-model", default="gpt-image-2") p.add_argument("--size", default="1024x1024") p.add_argument("--quality", default="high") p.add_argument("--prompts-only", action="store_true", help="plan만 만들고 image gen 스킵") p.add_argument("--max-shots", type=int, default=0, help="0이면 전체. 디버깅용 제한.") p.add_argument("--gen-missing-backgrounds", action="store_true", help="누락된 chain v6 background image를 자동 보강 후 shot image gen.") p.add_argument("--text-model", default="gpt-5.5", help="누락 prompt 보강 시 사용할 GPT 모델.") p.add_argument("--chain-run-dir", help="chain v6 run dir (도면 PNG + step1/2/7 위치). " "비우면 background_dir의 부모 사용.") p.add_argument("--chain-base-plan-ids", default="", help="chain v6 prompt 보강에 필요한 base_plan_ids (콤마 구분).") args = p.parse_args() proj_root = Path(args.main_project_root).resolve() out_dir = Path(args.out_dir).resolve() # 안전 가드: out_dir이 메인 프로젝트 안이면 거부 try: out_dir.relative_to(proj_root) logger.error("--out-dir이 메인 프로젝트 안. 거부 (메인 데이터 보호).") return 1 except ValueError: pass out_dir.mkdir(parents=True, exist_ok=True) ckpt_root = proj_root / "checkpoints" / "episodes" / args.episode_id planning_dir = Path(args.planning_dir).resolve() background_dir = Path(args.background_dir).resolve() if not ckpt_root.exists(): logger.error("checkpoint 디렉토리 없음: %s", ckpt_root); return 1 if not planning_dir.exists(): logger.error("planning 디렉토리 없음: %s", planning_dir); return 1 if not background_dir.exists(): logger.error("background 디렉토리 없음: %s", background_dir); return 1 if not os.getenv("OPENAI_API_KEY"): logger.error("OPENAI_API_KEY not set"); return 1 # === 메인 v4 read-only 로드 === sel_data = load_shot_selection(ckpt_root) scene_detail = load_scene_detail(ckpt_root) entity_t2i = load_entity_t2i(ckpt_root) logger.info("loaded main v4 read-only: shot_selection scenes=%d, scene_detail=%d, entity_t2i=%d", len(sel_data.get("scenes", [])), len(scene_detail), len(entity_t2i)) # === chain v6 read-only 로드 === chain_plan = load_chain_v6_plan(planning_dir) node_for_shot: dict = {} for n in chain_plan.get("nodes", []): for sid in (n.get("shot_ids") or []): if sid in node_for_shot: logger.warning("shot %s in multiple nodes — keep first %s, ignore %s", sid, node_for_shot[sid], n["id"]) continue node_for_shot[sid] = n["id"] logger.info("chain v6 plan loaded: %d nodes, %d shot mappings", len(chain_plan.get("nodes", [])), len(node_for_shot)) # === selected shots === scene_filter: set | None = None if args.scene_filter: scene_filter = {int(s.strip()) for s in args.scene_filter.split(",") if s.strip()} shot_filter: set | None = None if args.shot_filter: shot_filter = {s.strip() for s in args.shot_filter.split(",") if s.strip()} selected: list[tuple[int, int, str]] = [] for sc in sel_data.get("scenes", []): si = sc.get("scene_index") if scene_filter and si not in scene_filter: continue for s in sc.get("selected_shots", []): sx = s.get("shot_index") if sx is None: continue sid = f"S{si:02d}_Shot{sx}" if shot_filter and sid not in shot_filter: continue selected.append((si, sx, sid)) logger.info("selected after filter: %d", len(selected)) if args.max_shots > 0: selected = selected[:args.max_shots] # scene_detail 인덱싱 (scene, _shot_index) detail_by_id: dict = {} for entry in scene_detail: si = entry.get("scene_index"); sx = entry.get("_shot_index") if si is not None and sx is not None: detail_by_id[(si, sx)] = entry entity_by_short_id: dict = {} for nm, meta in entity_t2i.items(): sid_e = meta.get("short_id") if sid_e: entity_by_short_id[sid_e] = meta # === 각 selected shot에 대해 plan 결정 === plan: dict = {} missing_chain_image: set = set() for si, sx, sid in selected: entry = detail_by_id.get((si, sx)) if not entry: logger.warning("scene_detail 없음: %s — skip", sid); continue visible = entry.get("visible_entities") or [] node_id = node_for_shot.get(sid) if not node_id: logger.error("chain v6 매핑 없음: %s — skip", sid); continue ref_path = find_chain_v6_image(background_dir, node_id) if ref_path is None: missing_chain_image.add(node_id) # t2i_variations[0] variations = entry.get("t2i_variations") or [] if not variations: logger.warning("t2i_variations 비어있음: %s — skip", sid); continue v0 = variations[0] base_prompt = v0.get("t2i_prompt", "") camera_effect = v0.get("camera_effect", "") outfit_assignments = v0.get("outfit_assignments") or [] # entity 보강 ent_blocks: list[str] = [] for eid in visible: if eid in entity_by_short_id: meta = entity_by_short_id[eid] tr = english_traits_for_entity(meta) if tr.strip(): ent_blocks.append(tr) full_prompt_parts = [ "Same room/space as the reference image — match its wall finish, floor, " "ceiling, lighting tone, color palette. Photorealistic 35mm cinematic still.", ] if camera_effect: full_prompt_parts.append(f"Camera/effect: {camera_effect}") if base_prompt: full_prompt_parts.append(base_prompt) if ent_blocks: full_prompt_parts.append( "Entity references (use as English visual descriptors only, " "ignore Korean proper names): " + " | ".join(ent_blocks)) if outfit_assignments: full_prompt_parts.append( "Outfit assignments: " + json.dumps(outfit_assignments, ensure_ascii=False)) full_prompt = "\n\n".join(full_prompt_parts) plan[sid] = { "shot_id": sid, "scene_index": si, "shot_index": sx, "visible_entities": visible, "chain_node_id": node_id, "ref_path": str(ref_path) if ref_path else None, "ref_missing": ref_path is None, "prompt": full_prompt, } plan_p = out_dir / "v4_shot_plan.json" plan_p.write_text(json.dumps(plan, ensure_ascii=False, indent=2), encoding="utf-8") logger.info("saved plan: %s (%d shots)", plan_p, len(plan)) # === stdout 출력 === print("\n" + "=" * 70) print("=== V4 SHOT PLAN (chain v6 ref, set_design DISABLED) ===") print("=" * 70) for sid, info in plan.items(): ref_name = Path(info["ref_path"]).name if info["ref_path"] else "✗ MISSING" print(f"\n--- {sid} node={info['chain_node_id']} visible={info['visible_entities']} ---") print(f" ref: {ref_name}") print(f" prompt[:300]: {info['prompt'][:300]}...") if missing_chain_image: print("\n" + "!" * 70) print(f"!! 누락 chain v6 image ({len(missing_chain_image)} 노드 — Step C 필요):") for n in sorted(missing_chain_image): print(f" - {n}") print("!" * 70) logger.warning("이 노드들의 image를 chain_render_gpt로 보강 후 다시 실행 필요") if args.prompts_only: logger.info("prompts-only → image gen 스킵"); return 0 client = OpenAI() # === Step C: 누락 chain v6 background image 보강 === if missing_chain_image: if not args.gen_missing_backgrounds: logger.error("누락 chain v6 image %d개 — --gen-missing-backgrounds 필요", len(missing_chain_image)) return 1 # ancestor 포함한 full render set (BFS) nodes_by_id = {n["id"]: n for n in chain_plan["nodes"]} full_render_set = set(missing_chain_image) for nid in list(missing_chain_image): cur = nid while True: n = nodes_by_id.get(cur) if not n: break pid = n.get("parent_id") or "" if not pid or pid in ("null", "None"): break if pid not in nodes_by_id: break if not (background_dir / f"{pid}.png").exists(): full_render_set.add(pid) cur = pid # execution_order 따라 parent first execution_order = chain_plan.get("execution_order", []) to_render = [nid for nid in execution_order if nid in full_render_set] logger.info("--gen-missing-backgrounds → %d 노드 추가 생성 (BFS): %s", len(to_render), to_render) # node_prompts.json 로드 (chain_render_gpt_v6 결과) node_prompts_p = background_dir / "node_prompts.json" node_prompts: dict = {} if node_prompts_p.exists(): node_prompts = read_json(node_prompts_p) # 비어있는 prompt 보강 (chain_render의 함수 재사용) chain_ctx = chain_collect_context( Path(args.chain_run_dir).resolve() if args.chain_run_dir else background_dir.parent, args.chain_base_plan_ids.split(",") if args.chain_base_plan_ids else [], ) chain_meta = chain_build_node_meta(chain_plan, chain_ctx) groups_by_node: dict = {} for g in chain_plan.get("groups", []): for nid in g.get("node_ids", []): groups_by_node[nid] = g for nid in to_render: n = nodes_by_id[nid] existing = node_prompts.get(nid, "") if isinstance(existing, str) and existing.strip(): continue # LLM call pid = n.get("parent_id") or "" parent = nodes_by_id.get(pid) if pid and pid not in ("null", "None") else None user = chain_build_per_node_user_prompt( n, parent, groups_by_node.get(nid), chain_meta.get(nid)) logger.info("[regen prompt] %s — empty in node_prompts.json", nid) try: resp = chain_call_gpt_text_json( client, args.text_model, CHAIN_PROMPT_GEN_SYSTEM, user, CHAIN_PROMPT_GEN_SCHEMA, ) node_prompts[nid] = resp.get("t2i_prompt", "") except Exception as e: logger.error("prompt regen 실패 %s: %s", nid, e); return 1 # 저장 node_prompts_p.write_text( json.dumps(node_prompts, ensure_ascii=False, indent=2), encoding="utf-8") # image gen (parent first) for nid in to_render: n = nodes_by_id[nid] prompt_text = node_prompts.get(nid) if not prompt_text: logger.error("prompt 여전히 없음 — skip %s", nid); continue out_p = background_dir / f"{nid}.png" if out_p.exists(): logger.info("[skip existing] %s", nid); continue pid = n.get("parent_id") or "" ref: Path | None = None if pid and pid not in ("null", "None"): ref = background_dir / f"{pid}.png" if not ref.exists(): logger.error("parent image 없음 — skip %s (parent=%s)", nid, pid); continue else: # root anchor — base_plan PNG spid = n.get("source_plan_id") or "" if not spid: logger.error("root anchor source_plan_id 없음 — skip %s", nid); continue # chain run_dir에서 base_plan PNG 찾기 cand = (Path(args.chain_run_dir).resolve() if args.chain_run_dir else background_dir.parent) / f"base_plan_{spid}.png" if not cand.exists(): logger.error("base_plan PNG 없음 — skip %s: %s", nid, cand); continue ref = cand logger.info("[bg gen] %s using ref=%s", nid, ref.name) try: gpt_image_edit_with_ref( client, args.image_model, ref, prompt_text, args.size, args.quality, out_p, ) except Exception as e: logger.error("bg gen 실패 %s: %s", nid, e); return 1 # 다시 plan ref_path 채워넣기 (방금 만든 image들) for sid, info in plan.items(): if info.get("ref_missing"): p = background_dir / f"{info['chain_node_id']}.png" if p.exists(): info["ref_path"] = str(p) info["ref_missing"] = False plan_p.write_text(json.dumps(plan, ensure_ascii=False, indent=2), encoding="utf-8") for i, (sid, info) in enumerate(plan.items(), 1): if not info["ref_path"]: continue ref_path = Path(info["ref_path"]) if not ref_path.exists(): logger.error("ref 파일 없음 — skip %s: %s", sid, ref_path); continue out_p = out_dir / f"shot_{sid}.png" logger.info("[image %d/%d] %s using ref=%s", i, len(plan), sid, ref_path.name) try: gpt_image_edit_with_ref( client, args.image_model, ref_path, info["prompt"], args.size, args.quality, out_p, ) except Exception as e: logger.error("image gen 실패 %s: %s", sid, e) logger.info("=== DONE === %s", out_dir) return 0 if __name__ == "__main__": sys.exit(main())