"""Experiment v4 — Multi-step floor plan + photo background generation (시나리오 무관). 상세 계획서: backend/scripts/experiment_floor_plan_v4_plan.md 참조: backend/scripts/experiment_floor_plan_v4_background_generation_plan.md CLAUDE.md 절대 규칙: 1. 시나리오 의존 어휘 (인명·지명·작품명) 코드/시스템 프롬프트/출력 모든 필드 절대 금지. 2. LLM에 전달하는 데이터 절대 자르지 마라. 3. 도면 라벨에만 한국어 일반명사 OK. T2I 본문엔 보통명사만. 흐름: Phase A — 분석 (LLM × 3+N) Step 1 spatial + environment_canon Step 2 plan_specs + anchor_clusters Step 3 shot_specs (per shot) Phase B — 도면 이미지 Step 4 domain anchor base plan generate × D Step 5 non-anchor base plan edit(domain anchor) Step 6 shot plan edit(matching base plan) Phase C — 사진 Step 7 photo_specs (LLM × 1) Step 8 domain anchor photo generate × D (input: anchor plan PNG) Step 9 non-anchor photo edit (input: target plan PNG + domain anchor photo) Step 10 shot photo edit (input: matching base photo + shot plan overlay PNG) Phase D — set_design adapter preview JSON """ from __future__ import annotations import argparse import base64 import json import logging import os import re import sys import time from pathlib import Path from typing import Dict, Any, List, Optional, Set, Tuple 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 app.modules.llm.gemini_text_client import GeminiTextClient # noqa: E402 from app.modules.llm.gemini_image_client import GeminiImageClient # noqa: E402 logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s") logger = logging.getLogger("v4") # ────────────────────────────────────────────────────────── # Provider wrapper — OpenAI / Gemini 전환 (멀티턴 패턴 동일) # ────────────────────────────────────────────────────────── class TextProvider: """LLM 텍스트 호출 wrapper. JSON 응답 강제.""" def __init__(self, name: str, model: str, openai_client: Optional[OpenAI] = None): self.name = name; self.model = model self._oai = openai_client self._gem: Optional[GeminiTextClient] = None if name == "gemini": self._gem = GeminiTextClient(model=model) def call_json(self, system: str, user: str, schema: Optional[Dict[str, Any]] = None) -> Dict[str, Any]: if self.name == "openai": resp = self._oai.chat.completions.create( model=self.model, messages=[{"role": "system", "content": system}, {"role": "user", "content": user}], response_format={"type": "json_object"}, ) return json.loads(resp.choices[0].message.content) # gemini — schema 강제 (없으면 free → 빈 응답 위험) return self._gem.send_structured( user_message=user, response_schema=schema or {"type": "object"}, schema_name="result", system_instruction=system, ) def _img_size_to_aspect(size: str) -> str: """'1024x1024' → '1:1', '1536x1024' → '3:2'.""" try: w, h = (int(x) for x in size.lower().split("x")) from math import gcd g = gcd(w, h) return f"{w // g}:{h // g}" except Exception: return "1:1" class ImageProvider: """이미지 generate / edit wrapper. multi-ref 지원.""" def __init__(self, name: str, model: str, openai_client: Optional[OpenAI] = None): self.name = name; self.model = model self._oai = openai_client self._gem: Optional[GeminiImageClient] = None if name == "gemini": self._gem = GeminiImageClient(model=model) def generate(self, prompt: str, size: str, quality: str, out_path: Path, max_retries: int = 3) -> Path: last_exc: Optional[Exception] = None for attempt in range(1, max_retries + 1): try: if self.name == "openai": resp = self._oai.images.generate( model=self.model, prompt=prompt, size=size, quality=quality, n=1, ) out_path.write_bytes(_decode_or_raise(resp, f"gen({out_path.name})")) else: img_bytes, _ = self._gem.generate_image( prompt=prompt, aspect_ratio=_img_size_to_aspect(size), ) out_path.write_bytes(img_bytes) logger.info(" saved %d KB (attempt=%d)", out_path.stat().st_size // 1024, attempt) return out_path except Exception as e: last_exc = e logger.warning(" gen attempt=%d failed: %s", attempt, e) if attempt < max_retries: time.sleep(min(2 ** attempt, 8)) raise RuntimeError(f"generate failed after {max_retries}: {last_exc}") def edit(self, ref_paths: List[Path], prompt: str, size: str, quality: str, out_path: Path, max_retries: int = 3) -> Path: """multi-ref edit. Gemini는 generate에 reference_images 전달. M1 fix: 마지막 retry에서 단일 ref fallback 시도.""" last_exc: Optional[Exception] = None for attempt in range(1, max_retries + 1): # M1: 마지막 시도에서 multi-ref가 다 실패면 첫 ref만으로 fallback current_refs = ref_paths if attempt == max_retries and len(ref_paths) > 1: current_refs = ref_paths[:1] logger.warning(" edit final attempt fallback to single ref: %s", ref_paths[0].name) try: if self.name == "openai": files = [open(p, "rb") for p in current_refs] try: resp = self._oai.images.edit( model=self.model, image=files, prompt=prompt, size=size, quality=quality, n=1, ) finally: for f in files: f.close() out_path.write_bytes(_decode_or_raise(resp, f"edit({out_path.name})")) else: ref_bytes = [p.read_bytes() for p in current_refs] img_bytes, _ = self._gem.generate_image( prompt=prompt, reference_images=ref_bytes, aspect_ratio=_img_size_to_aspect(size), ) out_path.write_bytes(img_bytes) logger.info(" saved %d KB (attempt=%d, refs=%d)", out_path.stat().st_size // 1024, attempt, len(current_refs)) return out_path except Exception as e: last_exc = e logger.warning(" edit attempt=%d failed: %s", attempt, e) if attempt < max_retries: time.sleep(min(2 ** attempt, 8)) raise RuntimeError(f"edit failed after {max_retries}: {last_exc}") # ────────────────────────────────────────────────────────── # 안전 어휘 (sanitize 검증용) # ────────────────────────────────────────────────────────── UNSAFE_WORDS_PLAN = [ # 사망/시신 "dead", "deceased", "corpse", "cadaver", "victim", # 혈흔 (일반어 'red' 등은 제외, 명백한 단어만) "blood", "bloody", "bloodstain", "gore", "hemorrhage", # 외상 (일반어 'beat'는 영화 'beat'(박자) 등과 충돌하므로 제외) "wound", "trauma", "torn flesh", "mutilated", # 행위 (영화 용어와 충돌하는 'shot'/'beat'/'rip' 제외 — 'shoot'/'shot'는 영화에서 정당, # 'rip'은 'ripped' 등 일상어로 오판 위험. 명백한 동사만 유지) "stab", "stabbed", "strangle", "choke", "attack", "assault", "slash", # 도구 "knife", "blade", "firearm", # 범죄 "murder", "homicide", # 'crime scene'는 띄어쓰기 phrase — substring 매치로 잡힘 "crime scene", ] UNSAFE_WORDS_PHOTO = UNSAFE_WORDS_PLAN + [ # 사진 단계 추가 (명백한 violence 묘사만) "splatter", ] # ────────────────────────────────────────────────────────── # JSON schema (Gemini 등 schema 강제 모델에 전달) # OpenAI json_object 모드는 schema 무시 — system prompt에만 의존. # ────────────────────────────────────────────────────────── SCHEMA_STEP1: Dict[str, Any] = { "type": "object", "properties": { "space_groups": { "type": "array", "minItems": 1, # H2 fix: Gemini 빈 응답 방지 "items": { "type": "object", "properties": { "id": {"type": "string"}, "label": {"type": "string"}, "covers_location_ids": {"type": "array", "items": {"type": "string"}}, "scale_estimate": {"type": "string"}, "type": {"type": "string"}, "visual_domain": {"type": "string"}, "rooms": { "type": "array", "items": { "type": "object", "properties": { "name": {"type": "string"}, "walls": {"type": "array"}, "doors": {"type": "array"}, "windows": {"type": "array"}, "furniture": {"type": "array"}, "fixtures": {"type": "array"}, }, "required": ["name"], }, }, }, "required": ["id", "label", "visual_domain", "scale_estimate"], }, }, "shared_elements": {"type": "array"}, "environment_canon": { "type": "object", "properties": { "building": {"type": "object"}, "interior": {"type": "object"}, "exterior": {"type": "object"}, "site_context": {"type": "object"}, "shared_furniture_styles": {"type": "array"}, }, "required": ["building", "interior"], }, }, "required": ["space_groups", "environment_canon"], } SCHEMA_STEP2: Dict[str, Any] = { "type": "object", "properties": { "anchor_clusters": { "type": "array", "items": { "type": "object", "properties": { "id": {"type": "string"}, "visual_domain": {"type": "string"}, "anchor_plan_id": {"type": "string"}, "purpose": {"type": "string"}, }, "required": ["id", "visual_domain", "anchor_plan_id"], }, }, "base_plans": { "type": "array", "items": { "type": "object", "properties": { "id": {"type": "string"}, "label": {"type": "string"}, "visual_domain": {"type": "string"}, "anchor_cluster_id": {"type": "string"}, "is_anchor": {"type": "boolean"}, "canon_refs": {"type": "array"}, "t2i_prompt": {"type": "string"}, "legend": {"type": "array"}, "elements_meta": {"type": "array"}, "annotations": {"type": "array"}, }, "required": ["id", "visual_domain", "t2i_prompt"], }, }, "shot_assignment": {"type": "array"}, }, "required": ["anchor_clusters", "base_plans"], } SCHEMA_STEP3: Dict[str, Any] = { "type": "object", "properties": { "scene_index": {"type": "integer"}, "shot_index": {"type": "integer"}, "base_plan_id": {"type": "string"}, "visual_domain": {"type": "string"}, "camera": {"type": "object"}, "characters": {"type": "array"}, "additions": {"type": "array"}, "t2i_prompt": {"type": "string"}, "legend_updates": {"type": "array"}, }, "required": ["base_plan_id", "t2i_prompt"], } SCHEMA_STEP7: Dict[str, Any] = { "type": "object", "properties": { "anchor_clusters": { "type": "array", "minItems": 1, # H3 fix "items": { "type": "object", "properties": { "id": {"type": "string"}, "visual_domain": {"type": "string"}, "anchor_plan_id": {"type": "string"}, "source_floor_plan_image_path": {"type": "string"}, "reference_strategy": {"type": "string"}, }, "required": ["id", "visual_domain", "anchor_plan_id"], }, }, "base_photos": { "type": "array", "items": { "type": "object", "properties": { "id": {"type": "string"}, "source_plan_id": {"type": "string"}, "visual_domain": {"type": "string"}, "source_floor_plan_image_path": {"type": "string"}, "source_anchor_photo_id": {"type": "string"}, "reference_strategy": {"type": "string"}, "is_anchor": {"type": "boolean"}, "label": {"type": "string"}, "lighting": {"type": "string"}, "camera_note": {"type": "string"}, "t2i_prompt": {"type": "string"}, }, "required": ["id", "source_plan_id", "t2i_prompt"], }, }, "shot_photos": { "type": "array", "items": { "type": "object", "properties": { "id": {"type": "string"}, "source_shot": { "type": "object", "properties": { "scene_index": {"type": "integer"}, "shot_index": {"type": "integer"}, }, "required": ["scene_index", "shot_index"], # H3 fix }, "source_plan_id": {"type": "string"}, "matching_base_photo_id": {"type": "string"}, "shot_overlay_plan_image_path": {"type": "string"}, "reference_strategy": {"type": "string"}, "lighting": {"type": "string"}, "camera_note": {"type": "string"}, "t2i_prompt": {"type": "string"}, }, "required": ["id", "source_shot", "source_plan_id", "t2i_prompt"], }, }, }, "required": ["anchor_clusters", "base_photos"], } # ────────────────────────────────────────────────────────── # Isometric intermediate (Phase B') — top-down → 3D isometric # 산업 표준 패턴: 평면 → 등각 부감 → 사진. 시점 도약을 단계화. # ────────────────────────────────────────────────────────── def compact_canon(canon: Dict[str, Any]) -> str: """environment_canon을 한 단락 영문으로 응축 (isometric/photo prompt 삽입용).""" parts: List[str] = [] b = canon.get("building") or {} if b: bp = [] for k in ("stories", "primary_material", "exterior_stairs", "rooftop_features", "window_pattern", "weathering"): v = b.get(k) if v: if isinstance(v, list): v = ", ".join(str(x) for x in v if x) bp.append(f"{k}={v}") if b.get("color_palette"): bp.append("colors=" + ", ".join(str(x) for x in b["color_palette"])) if bp: parts.append("Building: " + "; ".join(bp)) i = canon.get("interior") or {} if i: ip = [] for k in ("wall_finish", "floor_finish", "ceiling", "lighting_fixtures", "general_clutter_level"): v = i.get(k) if v: if isinstance(v, list): v = ", ".join(str(x) for x in v if x) ip.append(f"{k}={v}") if ip: parts.append("Interior: " + "; ".join(ip)) e = canon.get("exterior") or {} if e: # H1 fix: CLAUDE.md 절대 규칙 — 자르기 금지 parts.append("Exterior: " + json.dumps(e, ensure_ascii=False)) return ". ".join(parts).strip() def build_axo_semi_photo_prompt(canon: Dict[str, Any], visual_domain: str) -> str: """isometric → axonometric semi-photoreal 변환 프롬프트 (Phase B''). isometric의 정확한 layout을 유지하되 사진 texture/조명을 적용. eye-level 사진 직전의 중간 단계 — 시점 도약을 작게 분할 (Option 2). """ canon_text = compact_canon(canon) or "neutral residential setting" domain_hint = { "interior": "interior cutaway: top half of walls cut away, full floor and lower-wall furniture visible, mid-detail textures", "exterior": "exterior axonometric: facade and rooftop with mid-detail materials, no full perspective convergence", "site_map": "site axonometric: low-rise building masses at slight angle, simplified ground textures", }.get(visual_domain, "axonometric semi-photo rendering") return ( "Convert this 3D isometric maquette rendering into an AXONOMETRIC semi-photorealistic visualization. " "Strict requirements:\n" "- Keep the same 30-45 degree elevated viewing angle as the input isometric (NO eye-level shift)\n" "- Slight axonometric foreshortening allowed but no full single-vanishing-point perspective\n" f"- {domain_hint}\n" "- Apply photorealistic materials and natural lighting on top of the isometric structure (walls, floor, " "furniture, doors, windows keep EXACT same positions, sizes, orientations as the input)\n" "- Soft realistic shadows, subtle ambient occlusion in corners and under furniture\n" "- Texture detail: visible wood grain on floors, cloth weave on bedding/curtains, brushed surfaces on appliances\n" "- No people, no overlaid text, no diagram lines, no architectural label letters\n" "- Materials and color palette consistent with the following canon:\n" f" {canon_text}\n" "Treat the input isometric as the strict ground truth for layout — only add photo-realistic materials " "and lighting, never move or invent walls/furniture." ) def build_isometric_prompt(canon: Dict[str, Any], visual_domain: str) -> str: """top-down 도면 → 3D isometric maquette 변환 프롬프트 (Phase B').""" canon_text = compact_canon(canon) or "neutral residential setting" domain_hint = { "interior": "interior cutaway: no roof, all walls visible from above, full furniture visible", "exterior": "exterior building isometric: facade, exterior stairs, rooftop features visible", "site_map": "site map isometric: low building masses, paths, surrounding context simplified", }.get(visual_domain, "architectural maquette isometric view") return ( "Convert this top-down 2D architectural floor plan into a clean 3D isometric maquette rendering. " "Strict requirements:\n" "- 45-degree elevated viewing angle, true axonometric isometric projection (no perspective foreshortening)\n" f"- {domain_hint}\n" "- All walls, doors, windows, and furniture preserved at exact same relative positions, sizes, and orientations as in the floor plan\n" "- Do not invent any room, wall, door, window, or furniture not present in the input plan\n" "- Clean architectural visualization style, soft global illumination, neutral white-grey background\n" "- Subtle drop shadows for spatial readability\n" "- No human figures, no label text overlay\n" "- Materials and color palette consistent with the following canon:\n" f" {canon_text}\n" "Treat the input plan as the strict ground truth for layout." ) # ────────────────────────────────────────────────────────── # 시스템 프롬프트 (시나리오 무관, 모든 단계) # ────────────────────────────────────────────────────────── _SAFETY_NOTE_PLAN = """## 안전 어휘 (도면 단계라도 단어 자체 회피) - 사망: "motionless seated figure marker", "still figure outline" (avoid: dead, corpse, deceased, victim, body) - 혈흔: "dark fluid stippled region", "reddish marker stain" (avoid: blood, bloodstain, bloody, gore) - 상처: "condition note", "state indicator" (avoid: wound, trauma, torn flesh, injury) - 행위: "altercation marker", "approach trail" (avoid: kill, stab, strangle, attack, assault, beat, slash, bite) - 도구: 형태(triangular/elongated)만 (avoid: knife, blade, weapon, gun) - 범죄: "investigation marker zone" (avoid: crime scene, murder, homicide) """ _SAFETY_NOTE_PHOTO = _SAFETY_NOTE_PLAN + """- 사진은 인물 0: "no people, empty room, uninhabited" - 위험 조합 회피 (deterministic moderation 트리거): - "low angle through doorway" + "bedroom" + "floor stain" - "doll" + "backpack" + "dim bedroom" + "stain/footprint" - "open door" + "dark bedroom interior" + "reddish stain" - 도면 라벨/선이 사진에 번지지 않게: "clean photo, no overlaid text, no diagram lines, no architectural label letters" """ _NO_SCENARIO_NOTE = """## ⚠ 시나리오 고유명사 절대 금지 (CLAUDE.md 절대 규칙) - 인명·도시명·지역명·국가명·작품명 어떤 고유명사도 출력 모든 필드에 절대 금지. - 입력 메타에 한국어 인명·고유 공간명이 있어도 일반 명사로 변환. - 한국어 일반 명사(거실/주방/옥탑방 등)는 도면 그래픽 라벨에만 OK. t2i 본문엔 보통명사만. - characters 메타 필드의 name(한국어 인명)은 후속 파이프라인 매칭용 메타 — t2i_prompt 본문에 절대 미주입. """ SYSTEM_STEP1 = f"""당신은 건축 도면 분석가입니다. 영화 시나리오 씬/샷 데이터를 받아 공간 구조를 분석하고 도면 분할 계획 + environment_canon을 작성합니다. {_NO_SCENARIO_NOTE} ## 분석 원칙 1. 데이터 의존만 (씬 원문/샷 description/location description/fixed_elements). 추측·창작 금지. 2. 공간 그룹핑: 한 평면 도면에 모두 담기 어려우면 분할. - 실내/실외 차이 큼 → 분할 - 층/높이 차이 (옥상 vs 1층) → 분할 - 스케일 차이 10배 이상 → 분할 - 같은 층 같은 스케일은 통합 3. shared_elements: 여러 씬에 걸쳐 동일하게 유지되는 시각 요소 (fixed_elements 영문 description 활용). 4. scale_estimate: "approx Nm x Mm" 모호 표기. 정확한 평수 단정 금지. 5. environment_canon: 건물 외관 + 실내 마감 + 가구 스타일 spec. 모든 base 도면/사진이 공유할 visual identity. {_SAFETY_NOTE_PLAN} ## 출력 JSON (엄격) {{ "space_groups": [ {{ "id": "snake_case_no_proper_nouns", "label": "한영 일반명사", "covers_location_ids": ["L##"], "scale_estimate": "approx Nm x Mm", "type": "interior|exterior|site_map", "visual_domain": "interior|exterior|site_map", "rooms": [ {{ "name":"한영 일반명사", "walls":["..."], "doors":[{{"connects_to":"","note":""}}], "windows":[{{"position":"","note":""}}], "furniture":[{{"name":"","position":"","note":""}}], "fixtures":[{{"name":"","position":""}}] }} ] }} ], "shared_elements": [ {{ "id":"...","appears_in_scenes":[12],"in_space_group_id":"...", "in_room":"...","description":"영문","note":"한국어 보조" }} ], "environment_canon": {{ "building": {{ "stories":"...","primary_material":"...", "color_palette":[],"exterior_stairs":"...", "rooftop_features":[],"window_pattern":"...", "weathering":"..." }}, "interior": {{ "wall_finish":"...","floor_finish":"...","ceiling":"...", "lighting_fixtures":[],"general_clutter_level":"..." }}, "exterior": {{ }}, "site_context": {{ }}, "shared_furniture_styles": [{{"category":"...","style_note":"..."}}] }} }} JSON만 출력. 추가 설명 금지. """ SYSTEM_STEP2 = f"""당신은 건축 도면 작성 전문가입니다. Step 1의 spatial_analysis를 받아 각 space_group의 base T2I 프롬프트와 anchor_clusters를 결정합니다. {_NO_SCENARIO_NOTE} ## 도면 스타일 (모든 도면 공통) - Black-and-white architectural plan, top-down orthographic view, technical line drawing on white background - Walls = solid black lines, doors = swing arcs, windows = double parallel lines - Furniture = labeled simple geometric outlines - **All graphic labels in ENGLISH ONLY** — no Korean, no Chinese, no Japanese, no CJK characters in the rendered image - Compass rose (N) at corner, scale bar (e.g. 0-5m) at corner - No 3D perspective, strictly 2D plan view - No human figures in base plan (empty of characters) - White background, no shading ## anchor_clusters 결정 - 각 visual_domain마다 anchor base plan 1개씩 (정보량 최대) - site_map → 부감 site_map 자체가 anchor - exterior → 정면 외관 elevation이 가장 좋음 - interior → 가장 큰 평면도 또는 모든 방을 포함하는 도면 ## canon spec 강제 삽입 - 모든 base_plans[].t2i_prompt 시작부에 environment_canon 응축 단락(영문 일반 명사) 포함 - canon_refs 필드에 어떤 canon 키를 참조했는지 기록 {_SAFETY_NOTE_PLAN} ## 출력 JSON (엄격) {{ "anchor_clusters": [ {{ "id":"cluster_","visual_domain":"interior|exterior|site_map", "anchor_plan_id":"","purpose":"..." }} ], "base_plans": [ {{ "id":"snake_case","label":"한영 일반명사","visual_domain":"interior|exterior|site_map", "anchor_cluster_id":"cluster_", "is_anchor": true, "canon_refs":["building.primary_material","interior.floor_finish"], "t2i_prompt":"영문 (canon spec 응축 + 도면 스타일 + room/door/window/furniture)", "legend":[{{"symbol":"","meaning":""}}], "elements_meta":[{{"id":"","kind":"wall|door|window|furniture|fixture", "label":"","position_hint":"","note":""}}], "annotations":[{{"text":"","placement":""}}] }} ], "shot_assignment": [ {{ "scene_index":12,"shot_index":4,"base_plan_id":"", "visual_domain":"interior" }} ] }} JSON만 출력. """ SYSTEM_STEP3 = f"""당신은 영화 카메라 도면 분석가입니다. 한 샷의 카메라 위치 / 인물 자세 / 추가 요소를 base 도면 위에 overlay할 수 있도록 정밀 분석합니다. {_NO_SCENARIO_NOTE} ## 카메라 표기 규칙 - triangular wedge symbol pointing in lens direction - dashed FOV cone (wider for WS, narrower for CU) - 라벨: `CAM-S##-Shot# (WS/MS/CU/MCU/ECU, angle, lens height, optional handheld/dolly note)` ## 인물·요소 표기 규칙 - 살아있는 인물: filled circle + 작은 방향 화살표 - 사망/움직이지 않는 인물: outline marker(X-cross 또는 chalk-style outline) + dashed contour - 가구 변경: 변경된 상태 그림 - 새 요소: grey/muddy 텍스처 영역 또는 explicit annotation {_SAFETY_NOTE_PLAN} ## 출력 JSON (엄격) {{ "scene_index": int, "shot_index": int, "base_plan_id": "string (Step 2 base_plans 중 하나)", "visual_domain": "interior|exterior|site_map", "camera": {{ "position":"...","heading":"...","height":"low|eye-level|high|overhead", "fov":"WS|MS|MCU|CU|ECU","lens_note":"..." }}, "characters": [ {{ "id":"C##","name":"<메타 보존; t2i_prompt에 직접 미주입>", "status":"alive|motionless|other","posture":"...","position":"...","facing":"..." }} ], "additions": [{{ "type":"...","position":"...","note":"..." }}], "t2i_prompt": "영문 (base 위 overlay 지시. 인명·지명 절대 금지, 보통명사만)", "legend_updates": [{{"symbol":"","meaning":""}}] }} JSON만 출력. """ SYSTEM_STEP7 = f"""당신은 영화 미술·시네마토그래피 전문가입니다. v4 도면 결과(spatial+canon+plans+shots)와 도면 PNG 경로를 받아 그 공간의 실사 사진 T2I 프롬프트를 작성합니다. {_NO_SCENARIO_NOTE} ## 사진 스타일 - Photorealistic cinematic still, 35mm film aesthetic - 자연 조명 (씬 데이터의 시간 단서 반영) - 환경 디테일 풍부 (가구·텍스처·생활감) - **인물 0** (no people, empty room) ## 데이터 의존 - spatial/elements_meta/fixed_elements에 명시된 요소만 사용 - 임의 추가 금지 ## ⚠ 공간 레이아웃 강제 (CRITICAL — 도면-사진 매칭 핵심) 도면(top-down 2D)에서 사진(eye-level photorealistic)으로 시점 도약이 크기 때문에 모델이 도면 구조를 무시하기 쉽다. 이를 막기 위해 **photo의 t2i_prompt 시작부에 spatial layout을 산문으로 명시**: 1. **scale 명시**: `"approx room"` (Step 1 spatial.scale_estimate 활용) 2. **벽별 요소 배치**: elements_meta의 walls/doors/windows/furniture를 방위별로: - `"NORTH wall: "` - `"SOUTH wall: <...>"` - `"EAST wall: <...>"` - `"WEST wall: <...>"` - `"CENTER/FLOOR: "` - `"CEILING: "` 3. **camera viewpoint** (사진별): 어느 위치에서 어느 방향을 보는지 — 산문으로 (예: `"camera at south wall corner, looking north toward bed and curtain"`). 4. **공간 연결**: 인접 공간이 보이면 명시 (예: `"open kitchen visible through doorway on east wall"`). 위 layout 기술을 prompt **첫 150 단어 안에 모두 포함**해야 한다. 모호하게 "small room with bed" 식으로 줄이지 말 것. ## reference 사용 (multi-image edit 권장) - domain anchor photo: input은 anchor plan PNG + canon - non-anchor base photo: input은 target plan PNG + domain anchor photo - shot photo: input은 matching base photo + (선택) shot plan overlay PNG + (선택) prev shot photo - reference_strategy 필드로 어떤 조합을 쓸지 명시: - "plan_png+domain_anchor_photo" (preferred) - "plan_png_only" (fallback) - "domain_anchor_photo_only" (fallback) - "base_photo+shot_overlay" (shot, preferred) - "base_photo_only" (shot, fallback) {_SAFETY_NOTE_PHOTO} ## 출력 JSON (엄격) {{ "anchor_clusters": [ {{ "id":"cluster__photo","visual_domain":"interior|exterior|site_map", "anchor_plan_id":"", "source_floor_plan_image_path":"base_plan_.png", "reference_strategy":"plan_png_only" }} ], "base_photos": [ {{ "id":"photo_base_","source_plan_id":"", "visual_domain":"interior|exterior|site_map", "source_floor_plan_image_path":"base_plan_.png", "source_anchor_photo_id":"photo_base_", "reference_strategy":"plan_png+domain_anchor_photo|plan_png_only|domain_anchor_photo_only", "is_anchor": false, "label":"한영 일반명사 (사진엔 그래픽 라벨 그리지 않음)", "lighting":"...","camera_note":"...", "t2i_prompt":"영문 (canon 동일 포함, 인명·지명 절대 금지)" }} ], "shot_photos": [ {{ "id":"photo_shot_S##_Shot#", "source_shot":{{"scene_index":int,"shot_index":int}}, "source_plan_id":"", "matching_base_photo_id":"photo_base_", "shot_overlay_plan_image_path":"shot_plan_.png", "reference_strategy":"base_photo+shot_overlay|base_photo_only", "lighting":"...","camera_note":"...","t2i_prompt":"영문" }} ] }} JSON만 출력. """ # ────────────────────────────────────────────────────────── # 데이터 수집 (자르기 금지) # ────────────────────────────────────────────────────────── def cp_path(project_id: str, episode_id: str, step_id: str) -> Path: return BACKEND.parent / "projects" / project_id / "checkpoints" / "episodes" / episode_id / step_id def load_cp(project_id: str, episode_id: str, step_id: str) -> dict: base = cp_path(project_id, episode_id, step_id) cur = base / "manifest.json" if cur.exists(): return json.loads(cur.read_text(encoding="utf-8")) arches = sorted(base.glob("manifest_*.json"), reverse=True) if arches: return json.loads(arches[0].read_text(encoding="utf-8")) raise FileNotFoundError(f"{step_id} (project={project_id} episode={episode_id})") def collect_context(project_id: str, episode_id: str, scene_indices: List[int], location_ids: List[str]) -> Dict[str, Any]: ctx: Dict[str, Any] = { "project_id": project_id, "episode_id": episode_id, "scope": {"scenes": scene_indices, "locations": location_ids}, } ss = load_cp(project_id, episode_id, "scene_save") ctx["scenes"] = [ {"scene_index": s["scene_index"], "heading": s["heading"], "text": s["text"]} for s in ss["data"]["segments"] if s["scene_index"] in scene_indices ] try: shv = load_cp(project_id, episode_id, "shot_validator") except FileNotFoundError: shv = load_cp(project_id, episode_id, "shot_extract") ctx["shots"] = [] for sc in shv["data"].get("scenes", []): if sc.get("scene_index") in scene_indices: for sh in sc.get("shots", []): ctx["shots"].append({ "scene_index": sc["scene_index"], "shot_index": sh.get("shot_index"), "description": sh.get("description", ""), "characters": sh.get("characters", []), }) st = load_cp(project_id, episode_id, "shot_staging") ctx["staging"] = [ {"scene_index": s["scene_index"], "shot_index": s["shot_index"], "camera_direction": s.get("camera_direction", ""), "lighting_mood": s.get("lighting_mood", ""), "perspective": s.get("perspective", ""), "key_bg_elements": s.get("key_bg_elements", []), "character_angles": s.get("character_angles", [])} for s in st["data"].get("shots", []) if s.get("scene_index") in scene_indices ] sc = load_cp(project_id, episode_id, "scene_consistency") ctx["fixed_elements"] = [ {"scene_index": s["scene_index"], "analysis_summary": s.get("analysis_summary", ""), "fixed_elements": s.get("fixed_elements", [])} for s in sc["data"].get("scenes", []) if s["scene_index"] in scene_indices ] loc = load_cp(project_id, episode_id, "entity_extract_location") locs = {l["short_id"]: l for l in loc["data"].get("locations", [])} missing = [lid for lid in location_ids if lid not in locs] if missing: logger.warning("location_ids 누락(checkpoint에 없음): %s", missing) ctx["locations"] = [ {"short_id": lid, "name": locs[lid].get("name"), "description": locs[lid].get("description", ""), "visual_traits": locs[lid].get("visual_traits", [])} for lid in location_ids if lid in locs ] em = load_cp(project_id, episode_id, "entity_merge") ctx["characters"] = [ {"short_id": c.get("short_id"), "name": c.get("name"), "description": c.get("description", "")} for c in em["data"].get("characters", []) ] try: vwr = load_cp(project_id, episode_id, "visual_world_rules") ctx["visual_world"] = { "era": vwr["data"].get("era"), "region": vwr["data"].get("region"), "director_notes": vwr["data"].get("director_notes", []), } except FileNotFoundError: ctx["visual_world"] = {} return ctx _COMMON_KOR_LOCATION_NOUNS = { "거실", "주방", "방", "현관", "옥상", "마당", "계단", "안방", "외부", "내부", "문", "창문", "벽", "바닥", "천장", "복도", "실내", "실외", "식탁", "침대", "옥탑방", "다세대", "빌라", "아파트", "건물", "주택", "자택", "부엌", "화장실", "욕실", "정원", "테라스", "발코니", "베란다", "지붕", "지하실", "다락방", "사무실", "교실", "병원", "카페", "식당", "공장", "창고", } def auto_extract_banned_words(ctx: Dict[str, Any]) -> List[str]: """entity_merge 인명 자동 추출 (D8, 보수적 — 인명만). 이전 버전은 locations description의 한국어 일반명사도 banned에 포함시켜 "가스레인지", "낡은", "작은" 같은 일반어가 LLM 출력에 등장하면 sanitize 실패가 발생했다. 실제 시나리오 의존 어휘는 인명 + 명백한 고유 지명 정도. 일반 명사는 LLM에 정상 등장 OK. 범위: - 인명 (한국어/영문, entity_merge.characters[].name) - 인명 토큰 분리 (공백/하이픈 분리된 부분 포함) - 영문 인명 변형(공백 제거)은 LLM 출력에서 잡기 어려우므로 수동 `--scenario-banned-words`로 보강 - 지명/일반명사는 자동 추출하지 않음 (오탐 위험 大). 필요 시 수동 banned 사용. """ banned: Set[str] = set() for c in ctx.get("characters", []): name = (c.get("name") or "").strip() if name and len(name) >= 2: banned.add(name) for tok in re.findall(r"[가-힣]{2,}|[A-Za-z][A-Za-z\-]+", name): if tok != name and len(tok) >= 2: banned.add(tok) return sorted(banned) # ────────────────────────────────────────────────────────── # 사용자 메시지 빌더 # ────────────────────────────────────────────────────────── def msg_step1(ctx: Dict[str, Any]) -> str: return ( "## 입력 데이터 (자르지 않음)\n\n" f"### 씬 원문 ({len(ctx['scenes'])}개)\n" + "\n\n".join(f"【{s['heading']}】\n{s['text']}" for s in ctx["scenes"]) + "\n\n" f"### 샷 description ({len(ctx['shots'])}개)\n" + "\n".join( f"- S{sh['scene_index']}_Shot{sh['shot_index']} (chars={sh['characters']}): {sh['description']}" for sh in ctx["shots"] ) + "\n\n" f"### shot_staging ({len(ctx['staging'])}개)\n" + "\n".join( f"- S{s['scene_index']}_Shot{s['shot_index']}: cam={s['camera_direction']} | " f"perspective={s['perspective']} | key_bg={json.dumps(s['key_bg_elements'], ensure_ascii=False)} | " f"angles={json.dumps(s['character_angles'], ensure_ascii=False)}" for s in ctx["staging"] ) + "\n\n" f"### fixed_elements ({len(ctx['fixed_elements'])}개)\n" + "\n".join( f"S{fe['scene_index']}: {fe.get('analysis_summary','')}\n" + "\n".join( f" - [{e['element_type']}] {e.get('element_id')} ({e.get('character_name','')}): " f"{e.get('description','')} | applies_to: {e.get('applies_to_shots')}" for e in fe.get("fixed_elements", []) ) for fe in ctx["fixed_elements"] ) + "\n\n" f"### locations ({len(ctx['locations'])}개)\n" + "\n".join( f"- {l['short_id']} ({l['name']}): {l['description']} | traits={l.get('visual_traits')}" for l in ctx["locations"] ) + "\n\n" f"### visual_world (참고)\n" f"era={ctx.get('visual_world', {}).get('era')}, " f"region={ctx.get('visual_world', {}).get('region')}, " f"director_notes={ctx.get('visual_world', {}).get('director_notes', [])}\n\n" "위 데이터로 Step 1 spatial_analysis JSON을 출력하세요. " "**id·label·room name·canon 모든 필드에 인명·지명 등 고유명사 절대 금지**." ) def msg_step2(ctx: Dict[str, Any], spatial: Dict[str, Any]) -> str: return ( "## Step 1 spatial_analysis 결과 (자르지 않음)\n```json\n" + json.dumps(spatial, ensure_ascii=False, indent=2) + "\n```\n\n" "## 원본 locations description (한국어 일반명사 보존용)\n" + "\n".join( f"- {l['short_id']} ({l['name']}): {l['description']} | traits={l.get('visual_traits')}" for l in ctx["locations"] ) + "\n\n" "## 관련 씬 헤딩\n" + "\n".join(f"- S{s['scene_index']}: {s['heading']}" for s in ctx["scenes"]) + "\n\n" "위 데이터로 Step 2 plan_specs JSON을 출력하세요. " "**각 visual_domain마다 anchor 1개씩**. " "**모든 base_plans[].t2i_prompt 시작부에 environment_canon 응축 단락(영문, 일반 명사)**을 포함하세요. " "라벨엔 한국어 일반명사 OK, t2i 본문엔 보통명사만." ) def msg_step3(ctx: Dict[str, Any], spatial: Dict[str, Any], plans: Dict[str, Any], scene_index: int, shot_index: int) -> str: shot_desc = next( (sh for sh in ctx["shots"] if sh["scene_index"] == scene_index and sh["shot_index"] == shot_index), None) staging = next( (s for s in ctx["staging"] if s["scene_index"] == scene_index and s["shot_index"] == shot_index), None) fixed = next((fe for fe in ctx["fixed_elements"] if fe["scene_index"] == scene_index), None) scene_text = next((s for s in ctx["scenes"] if s["scene_index"] == scene_index), None) def _matches_shot(applies_to: Any) -> bool: for x in (applies_to or []): try: if int(x) == int(shot_index): return True except (TypeError, ValueError): continue return False return ( f"## 대상 샷: S{scene_index}_Shot{shot_index}\n\n" "### 해당 씬 원문\n" + (f"【{scene_text['heading']}】\n{scene_text['text']}\n\n" if scene_text else "(없음)\n\n") + "### 샷 description\n" + (f"{shot_desc['description']}\n characters: {shot_desc['characters']}\n\n" if shot_desc else "(없음)\n\n") + "### shot_staging\n" + (json.dumps(staging, ensure_ascii=False, indent=2) + "\n\n" if staging else "(없음)\n\n") + "### fixed_elements (이 샷 applies_to 포함)\n" + ("\n".join( f" - [{e['element_type']}] {e.get('element_id')} ({e.get('character_name','')}): " f"{e.get('description','')} | applies_to: {e.get('applies_to_shots')}" for e in (fixed["fixed_elements"] if fixed else []) if _matches_shot(e.get("applies_to_shots")) ) or "(없음)") + "\n\n### Step 1 spatial_analysis (자르지 않음)\n```json\n" + json.dumps(spatial, ensure_ascii=False) + "\n```\n\n### Step 2 base_plans + anchor_clusters (자르지 않음)\n```json\n" + json.dumps(plans, ensure_ascii=False) + "\n```\n\n" f"위 데이터로 S{scene_index}_Shot{shot_index} overlay JSON을 출력하세요. " "**reasoning step**: (a) base_plan_id 매칭 → (b) visual_domain 추론 → " "(c) 카메라 좌표를 도면 좌표계 산문으로 → (d) 인물 status/posture/position 데이터 직접 추출 (추측 금지) → " "(e) additions에 fixed_elements + 샷 고유 변경 → (f) t2i_prompt 영문 작성 (인명·지명 절대 금지)." ) def msg_step7(ctx: Dict[str, Any], spatial: Dict[str, Any], plans: Dict[str, Any], shots: List[Dict[str, Any]], plan_image_paths: Dict[str, str]) -> str: return ( "## v4 도면 결과 (사진 T2I 작성용, 자르지 않음)\n\n" "### Step 1 spatial_analysis\n```json\n" + json.dumps(spatial, ensure_ascii=False, indent=2) + "\n```\n\n" "### Step 2 plan_specs\n```json\n" + json.dumps(plans, ensure_ascii=False, indent=2) + "\n```\n\n" f"### Step 3 shot overlays ({len(shots)}개)\n```json\n" + json.dumps(shots, ensure_ascii=False, indent=2) + "\n```\n\n" "### 도면 PNG 경로 (사진 reference로 사용)\n" + "\n".join(f"- {pid}: {path}" for pid, path in plan_image_paths.items()) + "\n\n" f"### 관련 씬 원문 ({len(ctx.get('scenes', []))}개, 시간/조명 단서)\n" + "\n\n".join(f"【{s['heading']}】\n{s['text']}" for s in ctx.get("scenes", [])) + "\n\n" "## 작업\n" "1) **anchor_clusters**: 각 visual_domain마다 anchor photo 1개씩 (도면 anchor와 동일 plan id 사용 권장).\n" "2) **base_photos**: Step 2 base_plans 각각에 대해 사진 T2I 프롬프트.\n" " - anchor: source_floor_plan_image_path 사용, reference_strategy='plan_png_only'\n" " - non-anchor: target plan PNG + domain anchor photo (multi-ref), reference_strategy='plan_png+domain_anchor_photo'\n" "3) **shot_photos**: 각 step3 shot에 대해 사진 T2I 프롬프트.\n" " - reference_strategy='base_photo+shot_overlay' (matching base photo + shot plan PNG)\n" "4) **모두 인물 0** + 도면 라벨/선이 사진에 번지지 않도록 명시.\n" "JSON만 출력." ) # ────────────────────────────────────────────────────────── # LLM (sanitize 재호출) # ────────────────────────────────────────────────────────── def call_llm_json(client_or_provider: Any, model: str, system: str, user: str, schema: Optional[Dict[str, Any]] = None) -> Dict[str, Any]: """OpenAI 또는 Gemini provider 모두 처리. provider면 provider.call_json 위임 (+ schema).""" if isinstance(client_or_provider, TextProvider): return client_or_provider.call_json(system, user, schema=schema) resp = client_or_provider.chat.completions.create( model=model, messages=[{"role": "system", "content": system}, {"role": "user", "content": user}], response_format={"type": "json_object"}, ) return json.loads(resp.choices[0].message.content) def grep_unsafe(payload: Any, words: List[str]) -> List[str]: text = json.dumps(payload, ensure_ascii=False).lower() return sorted({w for w in words if w.lower() in text}) def grep_scenario(payload: Any, banned: List[str]) -> List[str]: text = json.dumps(payload, ensure_ascii=False) return sorted({w for w in banned if w and w in text}) def call_llm_json_sanitized( client: Any, model: str, system: str, user: str, unsafe_words: List[str], banned_words: List[str], label: str, max_attempts: int = 2, skip_keys_for_banned: Optional[List[str]] = None, schema: Optional[Dict[str, Any]] = None, ) -> Tuple[Dict[str, Any], List[str], List[str]]: """LLM 호출 + sanitize 재호출 (D6). schema는 Gemini 등에 전달. skip_keys_for_banned: 이 키 경로는 banned grep에서 제외 (예: characters[].name 메타). """ last_unsafe: List[str] = [] last_banned: List[str] = [] extra_user = "" payload: Dict[str, Any] = {} for attempt in range(1, max_attempts + 1): payload = call_llm_json(client, model, system, user + extra_user, schema=schema) # banned 검사 시 skip_keys 제거한 사본 check_payload = payload if skip_keys_for_banned: check_payload = _strip_keys(payload, skip_keys_for_banned) last_unsafe = grep_unsafe(payload, unsafe_words) last_banned = grep_scenario(check_payload, banned_words) if not last_unsafe and not last_banned: logger.info("[%s] sanitize OK (attempt=%d)", label, attempt) return payload, last_unsafe, last_banned logger.warning("[%s] sanitize attempt=%d unsafe=%s banned=%s", label, attempt, last_unsafe, last_banned) problems = sorted(set(last_unsafe + last_banned)) extra_user = ( "\n\n## 재호출 — 다음 단어가 출력에 포함되어 있어 모두 제거하고 일반 명사로 대체하세요:\n" + json.dumps(problems, ensure_ascii=False) + "\nCLAUDE.md 절대 규칙: 시나리오 의존 어휘 + 안전 회피 어휘 0건이 되어야 합니다. JSON만 다시 출력." ) logger.error("[%s] sanitize 실패 (attempt 최대 도달). unsafe=%s banned=%s", label, last_unsafe, last_banned) return payload, last_unsafe, last_banned def _strip_keys(obj: Any, key_paths: List[str]) -> Any: """주어진 key path들을 제외한 deep copy. 단순 구현 — top-level/array 1단 path 처리.""" # 예: "characters.name" → 모든 characters[].name 키 제거 if isinstance(obj, dict): new = {} for k, v in obj.items(): new[k] = _strip_keys_inner(k, v, key_paths, [k]) return new if isinstance(obj, list): return [_strip_keys(v, key_paths) for v in obj] return obj def _strip_keys_inner(key: str, value: Any, key_paths: List[str], path_so_far: List[str]) -> Any: """recursive helper.""" cur = ".".join(path_so_far) if cur in key_paths: return None # 또는 ".name" suffix만으로 매치 (배열 안에) for kp in key_paths: # "characters.name" 등 — characters 배열의 각 원소에서 name 제거 parts = kp.split(".") if len(parts) == 2 and parts[0] == path_so_far[-1]: if isinstance(value, list): return [ {k: v for k, v in item.items() if k != parts[1]} if isinstance(item, dict) else item for item in value ] if isinstance(value, dict): return {k: _strip_keys_inner(k, v, key_paths, path_so_far + [k]) for k, v in value.items()} if isinstance(value, list): return [_strip_keys_inner(key, v, key_paths, path_so_far) for v in value] return value # ────────────────────────────────────────────────────────── # 이미지 (multi-ref edit 지원) # ────────────────────────────────────────────────────────── def _decode_or_raise(resp: Any, op: str) -> bytes: if not resp.data: raise RuntimeError(f"{op}: empty response.data") item = resp.data[0] b64 = getattr(item, "b64_json", None) if not b64: url = getattr(item, "url", None) raise RuntimeError(f"{op}: b64_json missing (url={url})") return base64.b64decode(b64) def gen_image(client_or_provider: Any, model: str, prompt: str, size: str, quality: str, out_path: Path, max_retries: int = 3) -> Path: """OpenAI 또는 Gemini provider 모두 처리. provider면 위임.""" if isinstance(client_or_provider, ImageProvider): logger.info("[image gen] %s → %s", client_or_provider.model, out_path.name) return client_or_provider.generate(prompt, size, quality, out_path, max_retries) logger.info("[image gen] %s → %s", model, out_path.name) last_exc: Optional[Exception] = None for attempt in range(1, max_retries + 1): try: resp = client_or_provider.images.generate( model=model, prompt=prompt, size=size, quality=quality, n=1) except Exception as e: last_exc = e logger.warning(" gen attempt=%d failed: %s", attempt, e) if attempt < max_retries: time.sleep(min(2 ** attempt, 8)) continue else: out_path.write_bytes(_decode_or_raise(resp, f"gen({out_path.name})")) logger.info(" saved %d KB (attempt=%d)", out_path.stat().st_size // 1024, attempt) return out_path raise RuntimeError(f"gen_image failed after {max_retries} attempts: {last_exc}") def edit_image(client_or_provider: Any, model: str, ref_paths: List[Path], prompt: str, size: str, quality: str, out_path: Path, max_retries: int = 3) -> Path: """multi-ref edit. OpenAI 또는 Gemini provider 모두 처리.""" if isinstance(client_or_provider, ImageProvider): logger.info("[image edit] %s + refs=%d → %s", client_or_provider.model, len(ref_paths), out_path.name) return client_or_provider.edit(ref_paths, prompt, size, quality, out_path, max_retries) logger.info("[image edit] %s + refs=%d → %s", model, len(ref_paths), out_path.name) last_exc: Optional[Exception] = None for attempt in range(1, max_retries + 1): files = [open(p, "rb") for p in ref_paths] try: resp = client_or_provider.images.edit( model=model, image=files, prompt=prompt, size=size, quality=quality, n=1, ) except Exception as e: last_exc = e for f in files: f.close() logger.warning(" edit attempt=%d failed: %s", attempt, e) if attempt < max_retries: time.sleep(min(2 ** attempt, 8)) continue else: for f in files: f.close() out_path.write_bytes(_decode_or_raise(resp, f"edit({out_path.name})")) logger.info(" saved %d KB (attempt=%d)", out_path.stat().st_size // 1024, attempt) return out_path raise RuntimeError(f"edit_image failed after {max_retries} attempts: {last_exc}") # ────────────────────────────────────────────────────────── # 검증 # ────────────────────────────────────────────────────────── def verify_schema(obj: Any, required_keys: List[str], label: str) -> List[str]: errs = [] if not isinstance(obj, dict): return [f"{label}: not a dict"] for k in required_keys: if k not in obj: errs.append(f"{label}: missing key {k!r}") return errs def verify_reference_integrity(spatial: Dict[str, Any], plans: Dict[str, Any], shots: List[Dict[str, Any]], photos: Optional[Dict[str, Any]]) -> List[str]: errs = [] space_ids = {sg.get("id") for sg in spatial.get("space_groups", [])} plan_ids = {p.get("id") for p in plans.get("base_plans", [])} domains = {sg.get("visual_domain") for sg in spatial.get("space_groups", []) if sg.get("visual_domain")} # anchor_clusters cluster_anchor_ids = [] for c in plans.get("anchor_clusters", []): ap = c.get("anchor_plan_id") if ap not in plan_ids: errs.append(f"anchor_cluster {c.get('id')}: anchor_plan_id {ap!r} not in base_plans") cluster_anchor_ids.append(ap) if c.get("visual_domain") not in domains and domains: errs.append(f"anchor_cluster {c.get('id')}: visual_domain {c.get('visual_domain')!r} not in spatial domains") # base_plans visual_domain for p in plans.get("base_plans", []): if p.get("visual_domain") not in domains and domains: errs.append(f"base_plan {p.get('id')}: visual_domain {p.get('visual_domain')!r} not in spatial domains") # shot base_plan_id + 필수 키 검증 for sh in shots: si, sx = sh.get("scene_index"), sh.get("shot_index") if si is None or sx is None: errs.append(f"shot entry missing scene_index/shot_index: keys={list(sh.keys())}") continue bid = sh.get("base_plan_id") if bid not in plan_ids: errs.append(f"shot S{si}_Shot{sx}: base_plan_id {bid!r} not in base_plans") # photos if photos: for ph in photos.get("base_photos", []): spid = ph.get("source_plan_id") if spid not in plan_ids: errs.append(f"base_photo {ph.get('id')}: source_plan_id {spid!r} not in base_plans") photo_ids = {ph.get("id") for ph in photos.get("base_photos", [])} for ph in photos.get("shot_photos", []): mb = ph.get("matching_base_photo_id") if mb and mb not in photo_ids: errs.append(f"shot_photo {ph.get('id')}: matching_base_photo_id {mb!r} not in base_photos") return errs def verify_canon_coverage(canon: Dict[str, Any], plans: Dict[str, Any]) -> List[str]: """environment_canon의 핵심 단어가 모든 base prompt에 등장하는지 점검.""" keys = [] for cat in ("building", "interior"): d = canon.get(cat) or {} for v in d.values(): if isinstance(v, str) and len(v) >= 4: keys.append(v.split()[0].lower()) # 첫 단어 keys = list({k for k in keys if k.isalpha()}) missing = [] for p in plans.get("base_plans", []): prompt = (p.get("t2i_prompt") or "").lower() absent = [k for k in keys if k and k not in prompt] # 너무 엄격하지 않게 — 50% 이상 누락 시만 기록 if keys and len(absent) > len(keys) * 0.5: missing.append({"plan_id": p.get("id"), "missing": absent[:5]}) return [json.dumps(m, ensure_ascii=False) for m in missing] def verify_no_text_overlay_in_photo_prompts(photos: Dict[str, Any]) -> List[str]: """사진 prompt에 'no people' 또는 'empty' + 'no overlaid text' 또는 'clean photo' 단어 검증.""" errs = [] for ph in (photos.get("base_photos", []) + photos.get("shot_photos", [])): text = (ph.get("t2i_prompt") or "").lower() if not any(w in text for w in ["no people", "empty room", "uninhabited", "no human", "no figures"]): errs.append(f"{ph.get('id')}: missing 'no people'-like instruction") if not any(w in text for w in ["no overlaid text", "no diagram lines", "clean photo", "no architectural label"]): errs.append(f"{ph.get('id')}: missing 'no overlaid text/diagram lines' instruction") return errs # ────────────────────────────────────────────────────────── # Adapter (set_design preview) # ────────────────────────────────────────────────────────── def build_set_design_adapter_preview( spatial: Dict[str, Any], plans: Dict[str, Any], shots: List[Dict[str, Any]], photos: Dict[str, Any], plan_image_paths: Dict[str, Path], base_photo_paths: Dict[str, Path], shot_plan_paths: Dict[str, Path], shot_photo_paths: Dict[str, Path], location_ids: List[str], ) -> Dict[str, Any]: """v4 결과를 set_design checkpoint 호환 구조로 변환 (Phase D). M5 fix: photo path 매칭은 source_plan_id 정확 매칭 (endswith 사용 X). M6 note: 멀티 location 입력 시 base_plans의 covers_location_ids로 분리해야 하나 현재는 단일 location 가정 — 멀티 시 warn. """ # source_plan_id → photo path (정확 매칭, M5 fix) plan_to_photo_path: Dict[str, Path] = {} for ph in (photos or {}).get("base_photos", []): spid = ph.get("source_plan_id") ph_id = ph.get("id") if spid and ph_id and ph_id in base_photo_paths: plan_to_photo_path[spid] = base_photo_paths[ph_id] base_image_entries: List[Dict[str, Any]] = [] for p in plans.get("base_plans", []): pid = p["id"] plan_png = plan_image_paths.get(pid) photo_png = plan_to_photo_path.get(pid) # shot_assignment 기반 covers_shots 계산 covers = [ f"S{sa['scene_index']:02d}_Shot{sa['shot_index']}" for sa in plans.get("shot_assignment", []) if sa.get("base_plan_id") == pid ] base_image_entries.append({ "base_id": f"base_{pid}", "angle_description": f"floor-plan guided {p.get('visual_domain')} base", "image_path": str(photo_png) if photo_png else "", "floor_plan_image_path": str(plan_png) if plan_png else "", "source_plan_id": pid, "visual_domain": p.get("visual_domain"), "anchor_cluster_id": p.get("anchor_cluster_id"), "is_anchor": bool(p.get("is_anchor")), "t2i": p.get("t2i_prompt"), "covers_shots": covers, }) shot_bg_entries: List[Dict[str, Any]] = [] for sh in shots: si = sh.get("scene_index") sx = sh.get("shot_index") if si is None or sx is None: logger.warning("adapter: shot entry skipped (missing scene/shot index): %s", {k: sh.get(k) for k in ("scene_index", "shot_index", "base_plan_id")}) continue try: label = f"S{int(si):02d}_Shot{int(sx)}" except (TypeError, ValueError): logger.warning("adapter: shot entry skipped (bad index types): si=%r sx=%r", si, sx) continue shot_bg_entries.append({ "scene_index": si, "shot_index": sx, "shot_label": label, "ref_type": "exact_floor_plan_background", "ref_id": f"base_{sh.get('base_plan_id')}", "source_plan_id": sh.get("base_plan_id"), "visual_domain": sh.get("visual_domain"), "image_path": str(shot_photo_paths.get(label, "")), "floor_plan_image_path": str(shot_plan_paths.get(label, "")), "t2i": sh.get("t2i_prompt"), "state_changes": json.dumps(sh.get("additions", []), ensure_ascii=False), }) # M6 fix: 단일 location 묶음 + 멀티 location 입력 시 warn if len(location_ids) > 1: logger.warning("adapter preview: %d locations 입력됐으나 단일 location 묶음으로 처리. " "멀티 location 분리는 v4.1 또는 base_plans.covers_location_ids 기반 후속 분리 필요.", len(location_ids)) primary_loc = location_ids[0] if location_ids else "L_unknown" return { "locations": { primary_loc: { "location_name": f"", "floor_plan": { "environment_canon": spatial.get("environment_canon", {}), "anchor_clusters": plans.get("anchor_clusters", []), "space_groups": spatial.get("space_groups", []), }, "base_images": base_image_entries, "shot_backgrounds": shot_bg_entries, } } } # ────────────────────────────────────────────────────────── # Gallery HTML (간단 grid) # ────────────────────────────────────────────────────────── def write_gallery(out_dir: Path, plans: Dict[str, Any], shots: List[Dict[str, Any]], plan_image_paths: Dict[str, Path], base_photo_paths: Dict[str, Path], shot_plan_paths: Dict[str, Path], shot_photo_paths: Dict[str, Path], manifest: Dict[str, Any]) -> Path: rows = [] # base 행 for p in plans.get("base_plans", []): pid = p["id"] plan_png = plan_image_paths.get(pid) photo_png = next((v for k, v in base_photo_paths.items() if k.endswith(pid)), None) rows.append({ "label_id": pid, "label_ko": p.get("label", ""), "domain": p.get("visual_domain", ""), "is_anchor": bool(p.get("is_anchor")), "plan": plan_png.name if plan_png else None, "photo": photo_png.name if photo_png else None, "is_shot": False, }) # shot 행 for sh in shots: label = f"S{sh['scene_index']:02d}_Shot{sh['shot_index']}" rows.append({ "label_id": label, "label_ko": "", "domain": sh.get("visual_domain", ""), "is_anchor": False, "plan": shot_plan_paths.get(label, Path("")).name or None, "photo": shot_photo_paths.get(label, Path("")).name or None, "is_shot": True, }) html_rows = [] for r in rows: kind = "SHOT" if r["is_shot"] else ("ANCHOR" if r["is_anchor"] else "BASE") html_rows.append(f"""
{r['label_id']} {kind} {r['domain']} {r['label_ko']}
{f'' if r["plan"] else "missing"}
PLAN
{f'' if r["photo"] else "missing"}
PHOTO
""") val = manifest.get("validation", {}) badges = ( f'scenario {len(val.get("scenario_word_hits") or [])}' f'unsafe {len(val.get("unsafe_word_hits") or [])}' f'missing {len(val.get("missing_files") or [])}' f'refs {len(val.get("reference_integrity_errors") or [])}' ) html = f""" v4 Gallery — {manifest.get("run_id","")}

v4 Gallery — Plan ↔ Photo

{manifest.get("run_id","")} · text={manifest.get("models",{}).get("text","?")} · image={manifest.get("models",{}).get("image","?")}
{badges}
{"".join(html_rows)}
""" out = out_dir / "gallery.html" out.write_text(html, encoding="utf-8") return out # ────────────────────────────────────────────────────────── # CLI # ────────────────────────────────────────────────────────── def parse_args() -> argparse.Namespace: p = argparse.ArgumentParser(description="v4 — Multi-step floor plan + photo (시나리오 무관)") p.add_argument("--project-id", required=True) p.add_argument("--episode-id", required=True) p.add_argument("--scene-indices", nargs="+", type=int, required=True) p.add_argument("--location-ids", nargs="+", required=True) p.add_argument("--target-shots", nargs="*", default=[], help='format: scene:shot, e.g. "12:4 12:6 25:2"') p.add_argument("--text-provider", default="openai", choices=["openai", "gemini"]) p.add_argument("--image-provider", default="openai", choices=["openai", "gemini"]) p.add_argument("--text-model", default="gpt-5.5") p.add_argument("--image-model", default="gpt-image-2") p.add_argument("--image-size-plan", default="1024x1024") p.add_argument("--image-size-photo", default="1536x1024") p.add_argument("--image-quality", default="high") p.add_argument("--out-name", default=None) p.add_argument("--scenario-banned-words", default="", help='추가 banned (쉼표 구분). 자동 추출과 합쳐짐') p.add_argument("--moderation-precheck", default="sanitize", choices=["warn", "sanitize", "skip"]) p.add_argument("--max-image-retries", type=int, default=3, help="이미지 API 호출 실패 시 재시도 횟수") p.add_argument("--require-domain-anchor", action="store_true", default=True, help="anchor가 없는 domain은 fail.") p.add_argument("--skip-isometric", action="store_true", help="Phase B' isometric 단계 건너뛰기 (단순 흐름)") p.add_argument("--skip-axo", action="store_true", help="Phase B'' axonometric 단계 건너뛰기 (단순 흐름)") p.add_argument("--skip-photo", action="store_true") p.add_argument("--skip-shots", action="store_true") p.add_argument("--skip-adapter", action="store_true") return p.parse_args() # ────────────────────────────────────────────────────────── # 메인 # ────────────────────────────────────────────────────────── def main() -> int: args = parse_args() # B2 fix: provider별 조건부 키 검사 needs_openai = (args.text_provider == "openai" or args.image_provider == "openai") needs_gemini = (args.text_provider == "gemini" or args.image_provider == "gemini") if needs_openai and not os.getenv("OPENAI_API_KEY"): logger.error("OPENAI_API_KEY not set (required for OpenAI provider)"); return 1 if needs_gemini and not os.getenv("GEMINI_API_KEY"): logger.error("GEMINI_API_KEY not set (required for Gemini provider)"); return 1 run_id = args.out_name or time.strftime("run_%Y%m%d_%H%M%S") out_dir = BACKEND / "scripts" / "output" / "floor_plan_v4" / run_id out_dir.mkdir(parents=True, exist_ok=True) photos_dir = out_dir / "photos" photos_dir.mkdir(parents=True, exist_ok=True) logger.info("run_id=%s out=%s", run_id, out_dir) openai_client = OpenAI() if needs_openai else None text_provider = TextProvider(args.text_provider, args.text_model, openai_client) image_provider = ImageProvider(args.image_provider, args.image_model, openai_client) logger.info("text=%s/%s image=%s/%s", args.text_provider, args.text_model, args.image_provider, args.image_model) client = openai_client # 기존 변수명 유지 (이후 호환) t0 = time.time() # 0) 컨텍스트 + banned 자동 추출 ctx = collect_context(args.project_id, args.episode_id, args.scene_indices, args.location_ids) (out_dir / "context.json").write_text( json.dumps(ctx, ensure_ascii=False, indent=2), encoding="utf-8") auto_banned = auto_extract_banned_words(ctx) manual_banned = [w.strip() for w in (args.scenario_banned_words or "").split(",") if w.strip()] banned = sorted(set(auto_banned) | set(manual_banned)) logger.info("scenario banned (auto=%d manual=%d total=%d)", len(auto_banned), len(manual_banned), len(banned)) # ── Phase A ───────────────────────────────────────────── pa_start = time.time() sanitize_mode = args.moderation_precheck def _llm(system, user, label, skip_keys=None, unsafe_words=None, schema=None): """provider 분기 + step별 JSON schema (Gemini는 필수, OpenAI 무시).""" words = unsafe_words if unsafe_words is not None else UNSAFE_WORDS_PLAN if sanitize_mode == "skip": return call_llm_json(text_provider, args.text_model, system, user, schema=schema), [], [] return call_llm_json_sanitized( text_provider, args.text_model, system, user, words, banned, label, max_attempts=2 if sanitize_mode == "sanitize" else 1, skip_keys_for_banned=skip_keys, schema=schema, ) logger.info("=== Step 1: spatial + canon (%s) ===", args.text_model) spatial, s1_unsafe, s1_banned = _llm(SYSTEM_STEP1, msg_step1(ctx), "step1", schema=SCHEMA_STEP1) (out_dir / "step1_spatial.json").write_text( json.dumps(spatial, ensure_ascii=False, indent=2), encoding="utf-8") logger.info("=== Step 2: plan_specs + anchor_clusters ===") plans, s2_unsafe, s2_banned = _llm(SYSTEM_STEP2, msg_step2(ctx, spatial), "step2", schema=SCHEMA_STEP2) (out_dir / "step2_plan_specs.json").write_text( json.dumps(plans, ensure_ascii=False, indent=2), encoding="utf-8") parsed_shots: List[Tuple[int, int]] = [] for s in args.target_shots: try: sc, sh = s.split(":"); parsed_shots.append((int(sc), int(sh))) except Exception: logger.warning("bad target-shot: %r", s) shot_results: List[Dict[str, Any]] = [] s3_unsafe_all: List[str] = [] s3_banned_all: List[str] = [] for si, sx in parsed_shots: logger.info("=== Step 3: S%d_Shot%d ===", si, sx) overlay, u, b = _llm( SYSTEM_STEP3, msg_step3(ctx, spatial, plans, si, sx), f"step3_S{si:02d}_Shot{sx}", skip_keys=["characters.name"], schema=SCHEMA_STEP3, ) s3_unsafe_all += u s3_banned_all += b # LLM이 scene_index/shot_index를 누락/잘못 채우는 경우 강제 보정 # (사용자가 CLI로 명시한 값이 진실) overlay["scene_index"] = si overlay["shot_index"] = sx label = f"S{si:02d}_Shot{sx}" (out_dir / f"step3_shot_{label}.json").write_text( json.dumps(overlay, ensure_ascii=False, indent=2), encoding="utf-8") shot_results.append(overlay) pa_dur = time.time() - pa_start # ── Phase B 도면 이미지 ───────────────────────────────── pb_start = time.time() plan_image_paths: Dict[str, Path] = {} valid_plan_ids: Set[str] = set() plan_id_to_anchor: Dict[str, str] = {} # plan_id → anchor_plan_id (같은 domain) # anchor_clusters 매핑 — M2 fix: 첫 번째 cluster를 우선 (last-wins 회피) anchor_by_domain: Dict[str, str] = {} for c in plans.get("anchor_clusters", []): d = c.get("visual_domain"); aid = c.get("anchor_plan_id") if not (d and aid): continue if d in anchor_by_domain: logger.warning("anchor cluster duplicate domain=%r — keeping first (%s, ignoring %s)", d, anchor_by_domain[d], aid) continue anchor_by_domain[d] = aid for p in plans.get("base_plans", []): d = p.get("visual_domain") a = anchor_by_domain.get(d) if a: plan_id_to_anchor[p["id"]] = a # Step 4: anchor 도면 generate (도메인별 1개 검증) anchors_done: Set[str] = set() # H3 fix: 같은 domain에 anchor가 여러 cluster에서 중복 정의된 경우 검증 domain_anchor_count: Dict[str, int] = {} for c in plans.get("anchor_clusters", []): d = c.get("visual_domain") if d: domain_anchor_count[d] = domain_anchor_count.get(d, 0) + 1 duplicate_domains = [d for d, n in domain_anchor_count.items() if n > 1] if duplicate_domains: logger.warning("duplicate anchor_clusters per domain: %s (마지막 항목이 우선)", duplicate_domains) for d, anchor_id in anchor_by_domain.items(): if not anchor_id: continue plan = next((p for p in plans.get("base_plans", []) if p["id"] == anchor_id), None) if not plan or not plan.get("t2i_prompt"): logger.error("anchor plan missing for domain %s: %s", d, anchor_id) continue out_p = out_dir / f"base_plan_{anchor_id}.png" try: gen_image(image_provider, args.image_model, plan["t2i_prompt"], args.image_size_plan, args.image_quality, out_p, max_retries=args.max_image_retries) plan_image_paths[anchor_id] = out_p valid_plan_ids.add(anchor_id) anchors_done.add(anchor_id) except Exception as e: logger.error("anchor plan gen failed (%s): %s", anchor_id, e) # Step 4.5/4.6: isometric / axonometric 중간 단계 (skip 가능) canon = spatial.get("environment_canon") or {} anchor_isometric_by_domain: Dict[str, Path] = {} anchor_axo_by_domain: Dict[str, Path] = {} if not args.skip_isometric: for d, anchor_id in anchor_by_domain.items(): anchor_plan_path = plan_image_paths.get(anchor_id) if not anchor_plan_path: continue iso_prompt = build_isometric_prompt(canon, d or "interior") out_p = out_dir / f"isometric_{anchor_id}.png" try: edit_image(image_provider, args.image_model, [anchor_plan_path], iso_prompt, args.image_size_plan, args.image_quality, out_p, max_retries=args.max_image_retries) anchor_isometric_by_domain[d] = out_p except Exception as e: logger.error("isometric gen failed (domain=%s, anchor=%s): %s", d, anchor_id, e) if not args.skip_axo and not args.skip_isometric: for d, anchor_id in anchor_by_domain.items(): iso_path = anchor_isometric_by_domain.get(d) if not iso_path: continue axo_prompt = build_axo_semi_photo_prompt(canon, d or "interior") out_p = out_dir / f"axo_{anchor_id}.png" try: edit_image(image_provider, args.image_model, [iso_path], axo_prompt, args.image_size_plan, args.image_quality, out_p, max_retries=args.max_image_retries) anchor_axo_by_domain[d] = out_p logger.info("axonometric semi-photo (domain=%s) ready", d) except Exception as e: logger.error("axo semi-photo failed (domain=%s, anchor=%s): %s", d, anchor_id, e) # Step 5: non-anchor 도면 edit(domain anchor) — BLOCKING-B2 fix anchor_fallback_warnings: List[str] = [] for plan in plans.get("base_plans", []): pid = plan["id"] if pid in anchors_done or not plan.get("t2i_prompt"): continue anchor_id = plan_id_to_anchor.get(pid) anchor_path = plan_image_paths.get(anchor_id) if anchor_id else None out_p = out_dir / f"base_plan_{pid}.png" try: if anchor_path: edit_image(image_provider, args.image_model, [anchor_path], plan["t2i_prompt"], args.image_size_plan, args.image_quality, out_p, max_retries=args.max_image_retries) else: # B2 fix: anchor 없는 domain은 visual identity 깨짐 — manifest에 명시 기록 msg = (f"plan {pid} (domain={plan.get('visual_domain')}): " f"domain anchor missing, falling back to independent generate " f"(visual identity NOT guaranteed)") anchor_fallback_warnings.append(msg) if args.require_domain_anchor: logger.error("ANCHOR FALLBACK BLOCKED (--require-domain-anchor): %s", msg) continue logger.warning("ANCHOR FALLBACK: %s", msg) gen_image(image_provider, args.image_model, plan["t2i_prompt"], args.image_size_plan, args.image_quality, out_p, max_retries=args.max_image_retries) plan_image_paths[pid] = out_p valid_plan_ids.add(pid) except Exception as e: logger.error("base plan failed (%s): %s", pid, e) # Step 6: shot 도면 edit(matching base) shot_plan_paths: Dict[str, Path] = {} if not args.skip_shots: for sh in shot_results: si, sx = sh.get("scene_index"), sh.get("shot_index") label = f"S{si:02d}_Shot{sx}" bid = sh.get("base_plan_id") base_p = plan_image_paths.get(bid) if not base_p or not sh.get("t2i_prompt"): logger.error("shot plan skip %s: missing base or prompt", label); continue out_p = out_dir / f"shot_plan_{label}.png" try: edit_image(image_provider, args.image_model, [base_p], sh["t2i_prompt"], args.image_size_plan, args.image_quality, out_p, max_retries=args.max_image_retries) shot_plan_paths[label] = out_p except Exception as e: logger.error("shot plan edit failed %s: %s", label, e) pb_dur = time.time() - pb_start # ── Phase C 사진 ──────────────────────────────────────── pc_start = time.time() photo_specs: Dict[str, Any] = {} base_photo_paths: Dict[str, Path] = {} shot_photo_paths: Dict[str, Path] = {} s7_unsafe: List[str] = [] s7_banned: List[str] = [] photo_specific_unsafe: List[str] = [] if not args.skip_photo: # Step 7 — BLOCKING-B1 fix: PHOTO 사전으로 sanitize 재호출 logger.info("=== Step 7: photo_specs (PHOTO unsafe 사전 적용) ===") photo_specs, s7_unsafe, s7_banned = _llm( SYSTEM_STEP7, msg_step7(ctx, spatial, plans, shot_results, {pid: str(p) for pid, p in plan_image_paths.items()}), "step7", skip_keys=None, unsafe_words=UNSAFE_WORDS_PHOTO, schema=SCHEMA_STEP7, ) (out_dir / "step7_photo_specs.json").write_text( json.dumps(photo_specs, ensure_ascii=False, indent=2), encoding="utf-8") # PHOTO 사전 한 번 더 체크 (sanitize 후에도 잔존 시 manifest 별도 키로 기록) photo_specific_unsafe = sorted( set(grep_unsafe(photo_specs, UNSAFE_WORDS_PHOTO)) - set(UNSAFE_WORDS_PLAN) ) if photo_specific_unsafe: logger.warning("photo-specific unsafe (post-sanitize): %s", photo_specific_unsafe) # Step 8: anchor photo — Phase B'' axonometric semi-photo + Phase B' isometric을 reference # refs = [axo semi-photo, isometric, anchor plan PNG] — 시점 도약 점진 분할 anchor_photo_by_domain: Dict[str, Path] = {} for ap in photo_specs.get("anchor_clusters", []): d = ap.get("visual_domain"); aid = ap.get("anchor_plan_id") anchor_plan_path = plan_image_paths.get(aid) anchor_iso_path = anchor_isometric_by_domain.get(d) anchor_axo_path = anchor_axo_by_domain.get(d) ph_entry = next( (b for b in photo_specs.get("base_photos", []) if b.get("source_plan_id") == aid), None) if not ph_entry or not anchor_plan_path: logger.error("anchor photo skip domain=%s aid=%s", d, aid); continue # 시점 가까운 ref가 앞쪽 — axo가 가장 가까움 refs = [p for p in (anchor_axo_path, anchor_iso_path, anchor_plan_path) if p] out_p = photos_dir / f"photo_base_{aid}.png" try: edit_image(image_provider, args.image_model, refs, ph_entry["t2i_prompt"], args.image_size_photo, args.image_quality, out_p, max_retries=args.max_image_retries) base_photo_paths[ph_entry["id"]] = out_p anchor_photo_by_domain[d] = out_p logger.info("anchor photo %s: refs=%d (axo=%s, iso=%s, plan=%s)", aid, len(refs), bool(anchor_axo_path), bool(anchor_iso_path), bool(anchor_plan_path)) except Exception as e: logger.error("anchor photo failed %s: %s", aid, e) # Step 9: non-anchor base photo — refs = [target plan PNG, domain isometric, domain anchor photo] for ph in photo_specs.get("base_photos", []): pid = ph.get("source_plan_id") if not pid or ph.get("id") in base_photo_paths: continue target_plan_path = plan_image_paths.get(pid) d = ph.get("visual_domain") anchor_iso_path = anchor_isometric_by_domain.get(d) anchor_photo = anchor_photo_by_domain.get(d) refs = [p for p in (target_plan_path, anchor_iso_path, anchor_photo) if p] if not refs or not ph.get("t2i_prompt"): logger.error("base photo skip %s: no refs", ph.get("id")); continue out_p = photos_dir / f"photo_base_{pid}.png" try: edit_image(image_provider, args.image_model, refs, ph["t2i_prompt"], args.image_size_photo, args.image_quality, out_p, max_retries=args.max_image_retries) base_photo_paths[ph["id"]] = out_p logger.info("base photo %s: refs=%d", ph.get("id"), len(refs)) except Exception as e: logger.error("base photo failed %s: %s", ph.get("id"), e) # Step 10: shot photo # - 같은 base_plan_id를 공유하는 shot들을 (scene_index, shot_index) 시간순으로 정렬 # - 같은 base 위 직전 shot photo를 추가 reference로 (장면 내부 visual chaining) # - H7: matching_base_photo_id 누락 시 source_plan_id로 역검색 if not args.skip_shots: plan_to_photo_id: Dict[str, str] = { ph.get("source_plan_id"): ph.get("id") for ph in photo_specs.get("base_photos", []) if ph.get("source_plan_id") and ph.get("id") } sorted_shot_photos = sorted( photo_specs.get("shot_photos", []), key=lambda x: ( (x.get("source_shot") or {}).get("scene_index") or 0, (x.get("source_shot") or {}).get("shot_index") or 0, ), ) prev_photo_per_base: Dict[str, Path] = {} # source_plan_id → 직전 shot photo for ph in sorted_shot_photos: src = ph.get("source_shot") or {} si, sx = src.get("scene_index"), src.get("shot_index") # B1 fix: si/sx None일 때 ph.id에서 정규식 추출 (이중 prefix 방지) if si is not None and sx is not None: label = f"S{int(si):02d}_Shot{int(sx)}" else: m = re.search(r"S(\d+)_Shot(\d+)", ph.get("id", "")) if m: label = f"S{int(m.group(1)):02d}_Shot{int(m.group(2))}" else: logger.error("shot photo skip — cannot derive label from id=%r src=%s", ph.get("id"), src) continue base_photo_id = ph.get("matching_base_photo_id") base_photo_path = base_photo_paths.get(base_photo_id) if not base_photo_path and ph.get("source_plan_id"): fallback_id = plan_to_photo_id.get(ph["source_plan_id"]) if fallback_id: base_photo_path = base_photo_paths.get(fallback_id) if base_photo_path: logger.warning("shot photo %s: matching_base_photo_id 매칭 실패 → " "source_plan_id로 fallback (%s → %s)", label, base_photo_id, fallback_id) shot_plan_path = shot_plan_paths.get(label) spid = ph.get("source_plan_id") prev_shot_photo_path = prev_photo_per_base.get(spid) if spid else None # 같은 장면 내부 chaining — 직전 shot photo가 있으면 visual identity 강화 reference refs = [p for p in (base_photo_path, shot_plan_path, prev_shot_photo_path) if p] if prev_shot_photo_path: logger.info("shot photo %s: prev-shot chaining (prev=%s)", label, prev_shot_photo_path.name) if not refs or not ph.get("t2i_prompt"): logger.error("shot photo skip %s: no refs", label); continue out_p = photos_dir / f"photo_shot_{label}.png" try: edit_image(image_provider, args.image_model, refs, ph["t2i_prompt"], args.image_size_photo, args.image_quality, out_p, max_retries=args.max_image_retries) shot_photo_paths[label] = out_p if spid: prev_photo_per_base[spid] = out_p except Exception as e: logger.error("shot photo failed %s: %s", label, e) pc_dur = time.time() - pc_start # ── Phase D adapter ──────────────────────────────────── pd_start = time.time() if not args.skip_adapter: adapter = build_set_design_adapter_preview( spatial, plans, shot_results, photo_specs, plan_image_paths, base_photo_paths, shot_plan_paths, shot_photo_paths, args.location_ids, ) (out_dir / "set_design_adapter_preview.json").write_text( json.dumps(adapter, ensure_ascii=False, indent=2), encoding="utf-8") pd_dur = time.time() - pd_start # ── 검증 + manifest ──────────────────────────────────── schema_errs = [] schema_errs += verify_schema(spatial, ["space_groups", "environment_canon"], "step1") schema_errs += verify_schema(plans, ["anchor_clusters", "base_plans"], "step2") if not args.skip_photo and photo_specs: # M1 fix: shot이 활성일 때 shot_photos도 required photo_required = ["anchor_clusters", "base_photos"] if not args.skip_shots and parsed_shots: photo_required.append("shot_photos") schema_errs += verify_schema(photo_specs, photo_required, "step7") ref_errs = verify_reference_integrity(spatial, plans, shot_results, photo_specs if not args.skip_photo else None) canon_missing = verify_canon_coverage(spatial.get("environment_canon", {}), plans) photo_text_errs = [] if not args.skip_photo and photo_specs: photo_text_errs = verify_no_text_overlay_in_photo_prompts(photo_specs) # 누락 파일 expected_files = list(plan_image_paths.values()) + list(base_photo_paths.values()) \ + list(shot_plan_paths.values()) + list(shot_photo_paths.values()) missing_files = [str(p) for p in expected_files if not p.exists()] # banned 합산 (사진 단계는 동일 banned 사전) all_banned = sorted(set(s1_banned + s2_banned + s3_banned_all + s7_banned)) all_unsafe = sorted(set(s1_unsafe + s2_unsafe + s3_unsafe_all + s7_unsafe)) manifest = { "run_id": run_id, "mode": "standalone_experiment", "args": vars(args), "models": {"text": args.text_model, "image": args.image_model}, "stats": { "space_groups": len(spatial.get("space_groups", [])), "base_plans": len(plans.get("base_plans", [])), "isometric_renders": len(anchor_isometric_by_domain), "shot_plans": len(shot_plan_paths), "base_photos": len(base_photo_paths), "shot_photos": len(shot_photo_paths), }, "phase_durations": { "phaseA": round(pa_dur, 1), "phaseB": round(pb_dur, 1), "phaseC": round(pc_dur, 1), "phaseD": round(pd_dur, 1), "total": round(time.time() - t0, 1), }, "validation": { "scenario_word_hits": all_banned, "unsafe_word_hits": all_unsafe, "photo_specific_unsafe_hits": photo_specific_unsafe, "anchor_fallback_warnings": anchor_fallback_warnings, "duplicate_domain_anchors": duplicate_domains, "missing_files": missing_files, "schema_errors": schema_errs, "reference_integrity_errors": ref_errs, "canon_coverage_warnings": canon_missing, "photo_text_overlay_errors": photo_text_errs, }, "set_design_adapter": { "enabled": not args.skip_adapter, "checkpoint_preview_path": "set_design_adapter_preview.json" if not args.skip_adapter else "", }, } (out_dir / "manifest.json").write_text( json.dumps(manifest, ensure_ascii=False, indent=2), encoding="utf-8") # gallery write_gallery(out_dir, plans, shot_results, plan_image_paths, base_photo_paths, shot_plan_paths, shot_photo_paths, manifest) logger.info("=== DONE === %s (총 %.1fs)", out_dir, time.time() - t0) logger.info("validation: scenario=%d unsafe=%d missing=%d ref_errs=%d schema=%d", len(all_banned), len(all_unsafe), len(missing_files), len(ref_errs), len(schema_errs)) return 0 if __name__ == "__main__": sys.exit(main())