"""[임시 v2] 옥탑방 shot 이미지 — entity ref multi-ref + moderation sanitize retry. 기존 experiment_v4_main_shot_render_temp.py를 import해서 plan 로직 재사용. 변경 사항: - DB의 image_asset에서 entity short_id → ref image PNG 매핑 로드 - shot 이미지 생성 시 [bg_ref, char_ref, outfit_ref, prop_ref] 멀티 ref - moderation 차단 시 PromptSanitizer.sanitize() 자동 retry (film_previs/movie_poster/aftermath) - HTML gallery 자동 생성 == 절대 규칙 == - 메인 v4 프로젝트는 READ-ONLY (DB SELECT만, 체크포인트/이미지 절대 수정 X) - set_design DISABLED - 결과는 --out-dir 임시 디렉토리에만 - 기존 experiment_v4_main_shot_render_temp.py 코드 수정 X (import만) """ from __future__ import annotations import argparse import base64 import json import logging import os 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 from sqlalchemy import create_engine, text # noqa: E402 # 기존 임시 스크립트 import (수정 X) sys.path.insert(0, str(Path(__file__).resolve().parent)) from experiment_v4_main_shot_render_temp import ( # noqa: E402 load_shot_selection, load_scene_detail, load_entity_t2i, load_chain_v6_plan, find_chain_v6_image, english_traits_for_entity, read_json, ) # PromptSanitizer는 backend 코드 그대로 (수정 X, 재사용 OK) from app.modules.prompt_sanitizer import PromptSanitizer # noqa: E402 logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s") logger = logging.getLogger("shot_render_with_refs") DB_URL = "postgresql://theroad:theroad_dev_2026@localhost:5432/theroad" # ====== entity ref 매핑 ====== def load_entity_ref_map(project_id: str) -> dict[str, Path]: """DB image_asset(reference, primary)에서 short_id → PNG 경로.""" eng = create_engine(DB_URL, isolation_level="AUTOCOMMIT") refs: dict[str, Path] = {} with eng.connect() as c: rows = c.execute(text( "SELECT ec.short_id, ia.file_path, ec.entity_type, ec.name " "FROM image_asset ia JOIN entity_canon ec ON ec.id = ia.entity_id " "WHERE ia.project_id = :p AND ia.asset_type='reference' " "AND ia.status='generated' AND ia.is_primary=1" ), {"p": project_id}).fetchall() eng.dispose() for sid, fp, etype, name in rows: if sid and fp and Path(fp).exists(): refs[sid] = Path(fp) logger.info("entity ref 매핑 로드: %d (chars/outlooks/props)", len(refs)) return refs def select_refs_for_shot(bg_path: Path, visible_entities: list[str], entity_refs: dict[str, Path], outfit_assignments: list[dict], max_refs: int = 8) -> list[Path]: """[bg, char_ref(s), outfit_ref(s), prop_ref(s)] 우선순위로 multi-ref 구성. gpt-image-2의 image edit은 이미지 여러개를 받을 수 있음 (list). 너무 많으면 응답 품질 저하 — max_refs 제한. """ refs: list[Path] = [bg_path] seen: set[Path] = {bg_path} # 1. characters (visible_entities 중 C##) for eid in visible_entities: if not eid.startswith("C"): continue p = entity_refs.get(eid) if p and p not in seen and len(refs) < max_refs: refs.append(p); seen.add(p) # 2. outfits from outfit_assignments for oa in outfit_assignments or []: oid = oa.get("outlook_id") if isinstance(oa, dict) else None if not oid: continue p = entity_refs.get(oid) if p and p not in seen and len(refs) < max_refs: refs.append(p); seen.add(p) # 3. props (visible_entities 중 P##) for eid in visible_entities: if not eid.startswith("P"): continue p = entity_refs.get(eid) if p and p not in seen and len(refs) < max_refs: refs.append(p); seen.add(p) return refs # ====== image edit + sanitize retry ====== def edit_with_sanitize(client: OpenAI, model: str, refs: list[Path], prompt: str, size: str, quality: str, out_path: Path, sanitizer: PromptSanitizer | None, max_attempts: int = 4) -> tuple[bool, dict]: """multi-ref image edit + moderation sanitize retry. 반환: (성공 여부, info dict {attempts, strategies, final_block_reason}) """ info = {"attempts": 0, "strategies": [], "final_block_reason": None, "ref_count": len(refs)} current_prompt = prompt for attempt in range(1, max_attempts + 1): info["attempts"] = attempt logger.info(" attempt %d/%d: %d refs, prompt=%dch", attempt, max_attempts, len(refs), len(current_prompt)) files = [] try: for p in refs: files.append(open(p, "rb")) try: resp = client.images.edit( model=model, image=files if len(files) > 1 else files[0], prompt=current_prompt, size=size, quality=quality, n=1, ) finally: for f in files: f.close() b64 = resp.data[0].b64_json if not b64: raise RuntimeError("empty b64 response") out_path.write_bytes(base64.b64decode(b64)) logger.info(" ✅ saved %d KB", out_path.stat().st_size // 1024) return True, info except Exception as e: err_str = str(e) err_low = err_str.lower() is_blocked = any(k in err_low for k in [ "moderation", "safety", "content_policy", "prohibited", "policy" ]) if is_blocked: logger.warning(" 🚫 moderation block (attempt %d): %s", attempt, err_str[:120]) if sanitizer is None or attempt >= max_attempts: info["final_block_reason"] = err_str[:200] return False, info # sanitize attempt = 1/2/3 → film_previs/movie_poster/aftermath try: s_attempt = min(attempt, 3) sanitize_result = sanitizer.sanitize( original_prompt=current_prompt, block_reason="SAFETY", block_categories=[], attempt=s_attempt, ) current_prompt = sanitize_result["sanitized_prompt"] strat = sanitize_result.get("strategy") info["strategies"].append(strat) logger.info(" → sanitized (strategy=%s)", strat) continue except Exception as se: logger.error(" sanitize 실패: %s", se) info["final_block_reason"] = f"sanitize_failed: {se}" return False, info else: logger.error(" ❌ non-moderation error: %s", err_str[:200]) info["final_block_reason"] = err_str[:200] return False, info return False, info # ====== HTML gallery ====== def build_gallery_html(out_dir: Path, plan: dict, shot_results: dict) -> Path: """간단한 grid HTML로 결과 확인.""" rows = [] for sid in sorted(plan.keys()): info = plan[sid] result = shot_results.get(sid, {}) ok = result.get("success", False) attempts = result.get("attempts", 0) strategies = result.get("strategies", []) block_reason = result.get("final_block_reason") node = info.get("chain_node_id", "") bg_name = Path(info["ref_path"]).name if info.get("ref_path") else "✗" visible = ", ".join(info.get("visible_entities", [])) ref_count = result.get("ref_count", 1) img_html = "" if ok: img_html = f'' else: img_html = f'
❌ {block_reason or "unknown"}
' sanitize_info = "" if strategies: sanitize_info = f' · sanitize: {"→".join(strategies)}' rows.append(f"""
{sid}
{img_html}
node: {node}
bg: {bg_name}
refs: {ref_count}
visible: {visible}
attempts: {attempts}{sanitize_info}
""") html = f""" 옥탑방 shot gallery

옥탑방 shot gallery — {len(plan)} shots

성공: {sum(1 for r in shot_results.values() if r.get("success"))} / {len(plan)} · 검열 우회 적용: {sum(1 for r in shot_results.values() if r.get("strategies"))} 건 · 최종 차단: {sum(1 for r in shot_results.values() if not r.get("success"))} 건

{"".join(rows)}
""" p = out_dir / "gallery.html" p.write_text(html, encoding="utf-8") return p # ====== main ====== def main() -> int: p = argparse.ArgumentParser() p.add_argument("--main-project-root", required=True) p.add_argument("--episode-id", required=True) p.add_argument("--planning-dir", required=True, help="chain v6 planning dir") p.add_argument("--background-dir", required=True, help="chain v6 background image dir") p.add_argument("--out-dir", required=True) p.add_argument("--scene-filter", help="콤마 구분 scene_index. 빈 값 = 모두") p.add_argument("--shot-filter", help="콤마 구분 shot_id") p.add_argument("--image-model", default="gpt-image-2") p.add_argument("--size", default="1024x1024") p.add_argument("--quality", default="high") p.add_argument("--max-shots", type=int, default=0) p.add_argument("--max-refs", type=int, default=6, help="multi-ref 최대 개수 (bg 포함)") p.add_argument("--prompts-only", action="store_true") p.add_argument("--max-attempts", type=int, default=4, help="moderation retry 최대 횟수") args = p.parse_args() proj_root = Path(args.main_project_root).resolve() out_dir = Path(args.out_dir).resolve() # safety: out_dir이 메인 프로젝트 안이면 거부 try: out_dir.relative_to(proj_root) logger.error("out-dir이 메인 프로젝트 내부 — 거부 (read-only 위반).") return 1 except ValueError: pass out_dir.mkdir(parents=True, exist_ok=True) project_id = proj_root.name 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 read-only: shot_selection scenes=%d, scene_detail=%d, entity_t2i=%d", len(sel_data.get("scenes", [])), len(scene_detail), len(entity_t2i)) # === entity ref 매핑 (DB에서) === entity_refs = load_entity_ref_map(project_id) # === chain v6 plan === 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 not in node_for_shot: node_for_shot[sid] = n["id"] logger.info("chain v6 plan: %d nodes, %d shot mappings", len(chain_plan.get("nodes", [])), len(node_for_shot)) # === selected shots filter === scene_filter = None if args.scene_filter: scene_filter = {int(s.strip()) for s in args.scene_filter.split(",") if s.strip()} shot_filter = 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 shots", len(selected)) if args.max_shots > 0: selected = selected[:args.max_shots] # === plan 구성 === 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 # location 기반 fallback 노드 (chain v6 plan에 없는 새 shot 처리) LOCATION_FALLBACK = { "L04": "photo_base_dense_low_rise_rooftop_site", # 옥탑방 외부 "L05": "photo_base_small_rooftop_room_interior", # 옥탑방 내부 } plan: dict = {} 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) used_fallback = False if not node_id: # location 기반 fallback for loc_id, fb_node in LOCATION_FALLBACK.items(): if loc_id in visible: node_id = fb_node used_fallback = True logger.info("chain v6 매핑 없음 → fallback %s → node=%s", sid, fb_node) break if not node_id: logger.error("chain v6 매핑/fallback 없음: %s (visible=%s) — skip", sid, visible) continue ref_path = find_chain_v6_image(background_dir, node_id) if ref_path is None: logger.warning("chain v6 image 없음: %s (node=%s) — skip", sid, node_id); continue 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 [] ent_blocks = [] for eid in visible: if eid in entity_by_short_id: tr = english_traits_for_entity(entity_by_short_id[eid]) if tr.strip(): ent_blocks.append(tr) # multi-ref 구성 refs = select_refs_for_shot(ref_path, visible, entity_refs, outfit_assignments, max_refs=args.max_refs) # 프롬프트 parts = [ "First reference image is the room/space — match its wall finish, floor, " "ceiling, lighting tone, color palette. Subsequent reference images are " "characters/outfits/props — preserve their appearance (face, clothing, design). " "Photorealistic 35mm cinematic still.", ] if camera_effect: parts.append(f"Camera/effect: {camera_effect}") if base_prompt: parts.append(base_prompt) if ent_blocks: parts.append("Entity references (English visual descriptors only, " "ignore Korean proper names): " + " | ".join(ent_blocks)) if outfit_assignments: parts.append("Outfit assignments: " + json.dumps(outfit_assignments, ensure_ascii=False)) full_prompt = "\n\n".join(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), "ref_paths_all": [str(r) for r in refs], "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("plan saved: %s (%d shots)", plan_p, len(plan)) # 요약 출력 print("\n" + "=" * 70) print(f"=== PLAN ({len(plan)} shots, multi-ref + sanitize) ===") print("=" * 70) for sid, info in plan.items(): ref_names = [Path(p).name for p in info["ref_paths_all"]] print(f"\n--- {sid} node={info['chain_node_id']} visible={info['visible_entities']} ---") print(f" refs ({len(ref_names)}): {ref_names}") print(f" prompt[:200]: {info['prompt'][:200]}...") if args.prompts_only: logger.info("--prompts-only → 종료") return 0 # === image gen + sanitize === client = OpenAI() sanitizer = PromptSanitizer() shot_results: dict = {} total = len(plan) for i, (sid, info) in enumerate(plan.items(), 1): refs = [Path(p) for p in info["ref_paths_all"]] out_p = out_dir / f"shot_{sid}.png" logger.info("[%d/%d] %s — %d refs", i, total, sid, len(refs)) success, run_info = edit_with_sanitize( client, args.image_model, refs, info["prompt"], args.size, args.quality, out_p, sanitizer, max_attempts=args.max_attempts, ) run_info["success"] = success shot_results[sid] = run_info # 결과 요약 success_count = sum(1 for r in shot_results.values() if r.get("success")) sanitized_count = sum(1 for r in shot_results.values() if r.get("strategies")) blocked_count = total - success_count logger.info("=== DONE === %d/%d 성공, %d sanitize 적용, %d 차단", success_count, total, sanitized_count, blocked_count) # 결과 JSON 저장 (out_dir / "shot_results.json").write_text( json.dumps(shot_results, ensure_ascii=False, indent=2), encoding="utf-8") # HTML 갤러리 gallery = build_gallery_html(out_dir, plan, shot_results) logger.info("gallery: %s", gallery) return 0 if __name__ == "__main__": sys.exit(main())