#!/usr/bin/env python3 """Rooftop Spatial BG Experiment — standalone (plan v2 APPROVED 2026-05-23). plan_v2: scripts_output/rooftop_spatial_bg_experiment/plan_v2.md phase: C (dry-run) — collect + space_bible + camera_slot + prompts + cost + index stub. production pipeline 0 수정, DB write 0, API 호출 0. """ from __future__ import annotations import argparse import base64 import html import json import re import sys import time import uuid from dataclasses import dataclass, field from datetime import datetime from pathlib import Path from typing import Optional # repo root setup _REPO_ROOT = Path(__file__).resolve().parents[2] _BACKEND_ROOT = _REPO_ROOT / "backend" if str(_BACKEND_ROOT) not in sys.path: sys.path.insert(0, str(_BACKEND_ROOT)) def _load_backend_env() -> None: """backend/.env 를 os.environ 에 주입 — pydantic-settings 의 env_file='.env' 가 cwd 상대라 repo root 에서 실행되면 못 찾는 문제 해결. production cwd=backend 와 동등 효과.""" import os as _os # noqa: PLC0415 env_path = _BACKEND_ROOT / ".env" if not env_path.exists(): return for raw in env_path.read_text(encoding="utf-8").splitlines(): line = raw.strip() if not line or line.startswith("#") or "=" not in line: continue key, value = line.split("=", 1) key = key.strip() value = value.strip() if (value.startswith('"') and value.endswith('"')) or ( value.startswith("'") and value.endswith("'") ): value = value[1:-1] # 기존 env 우선 (사용자 명시 export 보호). if key and key not in _os.environ: _os.environ[key] = value _load_backend_env() from app.core.database import SessionLocal # noqa: E402 from app.models.project import EntityCanon, ImageAsset, SceneStill # noqa: E402 # --------------------------------------------------------------------------- # Constants — plan v2 §1, §2 # --------------------------------------------------------------------------- PROJECT_ID = "6cb862d9-590c-4dce-86e6-d10c2977db19" EPISODE_ID = "08ad2cd3-3e96-4d84-808f-869ee628473c" L05_CANON_ID = "3afbc7a8-b919-4431-a691-0a99057a26ca" L05_SHORT_ID = "L05" DEFAULT_OUTPUT_DIR = Path("scripts_output/rooftop_spatial_bg_experiment") # Codex review v2: score 기반 classifier (priority 만으론 evidence 강도 측정 불가). # strong evidence = 그 sub_space 만 가리키는 명시적 단어. weak = 다른 sub_space 와도 호환. # shot_description hit = +3, scene_summary hit = +1 (shot 우선). SUB_SPACE_KEYWORDS: dict[str, dict[str, list[str]]] = { "bathroom": { "strong": ["욕실", "화장실"], "weak": ["거울"], # 거실에도 거울 있을 수 있으나 옥탑방 컨텍스트에선 욕실 hint }, "bedroom": { "strong": ["침실", "침대", "베개", "이불", "매트리스", "수리영의 방", "민숙의 방"], # "방 안" 단독은 약 evidence — 옥탑방 거실도 "방 안" 으로 표현될 수 있음. "weak": ["방안", "방 안", "방바닥", "방안의"], }, "entry_corner": { # 위치 명시만 strong. "현관문" 은 거실 framing 안에서도 보이므로 weak. "strong": ["현관 안쪽", "현관 안에서", "신발장 옆"], "weak": ["현관문", "신발 벗", "진입로"], }, "main_room": { "strong": ["거실", "식탁", "싱크대", "식사 공간", "소파", "TV", "텔레비전", "뉴스"], "weak": ["주방"], }, } # shot_description vs scene_summary, strong vs weak keyword 4 tier: # shot strong = SHOT_WEIGHT * STRONG_MULT = 3.0 # shot weak = SHOT_WEIGHT * WEAK_MULT = 1.5 # summary strong = SUMMARY_WEIGHT * STRONG_MULT = 1.0 # summary weak = SUMMARY_WEIGHT * WEAK_MULT = 0.5 SHOT_WEIGHT = 3.0 SUMMARY_WEIGHT = 1.0 STRONG_MULT = 1.0 WEAK_MULT = 0.5 STRONG_WEIGHT = SHOT_WEIGHT * STRONG_MULT # 호환 alias (cleaned downgrade 용) SUB_SPACE_MIN_SCORE = 2.0 # weak-only (shot weak 1.5 + summary weak 0.5) 이상 채택 SUB_SPACE_TIE_MARGIN = 1.5 # 1위 - 2위 < margin → tie → manual_review_needed # state classifier — score + negation guard (Codex Important C). STATE_KEYWORDS: dict[str, dict[str, list[str]]] = { "corpse_marks": { "strong": ["시신", "시체", "어깨가 뜯", "쇄골", "참혹한 모습", "주저앉아 있는"], "weak": ["피", "핏자국", "흥건한"], }, "vandalized": { # 손목/문신 근처의 "붉은 원형 표식" 은 캐릭터 mark 라 vandalized 가 아님 → negation guard. "strong": [ "어지럽혀", "어지러진", "뒤집힌", "넘어진 가구", "벽면에 붉은", "벽면에 거친", "벽에 칠한", "벽면에 그려진", "붓으로 칠한", ], "weak": ["붉은 원"], }, "cleaned": { "strong": ["깨끗하게 정돈", "지나치게 깔끔", "치워", "흔적을 찾지 못해", "흔적이 없"], "weak": ["깨끗", "정돈"], }, "empty": { "strong": ["빈 집", "아무도 없", "혼자 깨어"], "weak": ["빈 공간"], }, "normal": { # 기본값 — 어느 다른 state 도 strong 으로 안 잡힐 때만 "strong": [], "weak": [], }, } # state 가 cleaned + corpse_marks 동시 강하면 cleaned 우선 (negation 의도). CLEANED_OVERRIDES_CORPSE = True # vandalized 후보를 무력화하는 character-mark 컨텍스트 (손목/문신/표식을 본다). VANDALIZED_NEGATION_CONTEXTS = [ r"손목.{0,15}(붉은|원형|표식)", r"문신.{0,15}(붉은|원형|표식)", r"표식을\s*본다", r"손목\s*안쪽", ] # corpse_marks 무력화 — "시신/피/흔적 없는/없이/지워" 같은 negation. CORPSE_NEGATION_CONTEXTS = [ r"시신.{0,5}없", r"시체.{0,5}없", r"피.{0,5}없", r"핏자국.{0,5}없", r"흔적.{0,5}없", r"흔적을\s*찾지", ] STATE_MIN_SCORE = 1.5 # summary strong (1.0) + summary weak (0.5) 또는 shot weak (1.5) 단독이면 채택 TIME_KEYWORDS: dict[str, dict[str, list[str]]] = { "morning": {"strong": ["아침"], "weak": ["새벽", "해 뜨"]}, "dusk": {"strong": ["해질", "해 질", "황혼", "노을", "어스름"], "weak": []}, "night": {"strong": ["밤", "어두운", "어둠"], "weak": ["불 꺼"]}, "day": {"strong": [], "weak": ["햇살", "햇빛", "낮", "대낮"]}, # default fallback } TIME_MIN_SCORE = 1.0 # summary strong 1.0 이상 — weak-only fallback 은 day 로 # Camera slot pool — (sub_space, slot_label, slot_description). plan §3 자유 구도 금지. # Codex Blocking 3: metric (m, mm) 제거 — qualitative 표현만. CAMERA_SLOT_POOL: list[tuple[str, str, str]] = [ ("main_room", "eye_level_wide", "거실 중앙 standing eye-level, wide framing, 식탁/주방 정면, 거실 전체"), ("main_room", "eye_level_table_close", "식탁 옆 seated height, normal lens framing, 식탁 위 small props 중심"), ("main_room", "eye_level_couch_pov", "소파 옆 standing eye-level, 거실 안쪽 응시"), ("bedroom", "eye_level_doorway_wide", "방 입구 문턱 standing eye-level, wide framing, 방 안 전체"), ("bedroom", "low_floor_bed_close", "바닥 가까운 낮은 앵글, 침대 옆 시점, 침구/바닥 중심"), ("bathroom", "eye_level_mirror_close", "욕실 standing eye-level, 거울 정면, 거울 + 벽 일부"), ("entry_corner", "eye_level_inner_view", "현관 안쪽 standing eye-level, 거실 방향 응시, 신발/진입로 + 거실 일부"), ] # framing 키워드 → 선호 slot tag (없으면 sub_space 의 첫 slot) FRAMING_KEYWORD_MAP: dict[str, list[str]] = { "wide": ["eye_level_wide", "eye_level_doorway_wide"], "close": ["eye_level_table_close", "low_floor_bed_close", "eye_level_mirror_close"], } WIDE_TEXT_HINTS = r"전체|벽|방\s*안|거실\s*전|중앙" CLOSE_TEXT_HINTS = r"클로즈|근접|손목|얼굴|식탁\s*위|거울\s*속|벽면" GPT_IMAGE_2_UNIT_COST = 0.04 GPT5_VISION_UNIT_COST = 0.02 # --------------------------------------------------------------------------- # Data classes # --------------------------------------------------------------------------- @dataclass class ShotMeta: still_id: str scene_index: int shot_index: int scene_summary: str shot_description: str visible_short_ids: list[str] dependent_scene_id: Optional[str] raw_visible_json: str @property def label(self) -> str: return f"S{self.scene_index}_S{self.shot_index}" @dataclass class ShotPlan: shot: ShotMeta sub_space: str # main_room|bedroom|bathroom|entry_corner|manual_review_needed state: str # normal|corpse_marks|cleaned|empty|vandalized time: str # morning|day|dusk|night camera_slot: str # slot_label or "fallback_unknown" manual_review_needed: bool classification_notes: list[str] = field(default_factory=list) # --------------------------------------------------------------------------- # DB read-only loaders # --------------------------------------------------------------------------- def load_l05_shots(session) -> list[ShotMeta]: rows = ( session.query(SceneStill) .filter(SceneStill.episode_id == EPISODE_ID) .filter(SceneStill.is_selected.is_(True)) .filter(SceneStill.visible_entities_json.like(f"%{L05_SHORT_ID}%")) .order_by(SceneStill.scene_index, SceneStill.shot_index) .all() ) out: list[ShotMeta] = [] for r in rows: try: visible = json.loads(r.visible_entities_json or "[]") except json.JSONDecodeError: visible = [] visible_short_ids = [ v.get("short_id") for v in visible if isinstance(v, dict) and v.get("short_id") ] out.append(ShotMeta( still_id=r.id, scene_index=int(r.scene_index or 0), shot_index=int(r.shot_index or 0), scene_summary=(r.scene_summary or "").strip(), shot_description=(r.shot_description or "").strip(), visible_short_ids=visible_short_ids, dependent_scene_id=r.dependent_scene_id, raw_visible_json=r.visible_entities_json or "[]", )) return out def load_existing_chain_bg_prompts(session) -> list[str]: rows = ( session.query(ImageAsset) .filter(ImageAsset.project_id == PROJECT_ID) .filter(ImageAsset.asset_type == "chain_bg") .filter(ImageAsset.entity_id == L05_CANON_ID) .all() ) return [(a.prompt_used or "").strip() for a in rows if a.prompt_used] def load_l05_canon_description(session) -> str: canon = session.query(EntityCanon).filter(EntityCanon.id == L05_CANON_ID).one_or_none() if canon is None: return "" return (canon.description or "").strip() # --------------------------------------------------------------------------- # Classifiers # --------------------------------------------------------------------------- def _score_keywords( keyword_map: dict[str, dict[str, list[str]]], shot_description: str, scene_summary: str, ) -> dict[str, dict]: """label 별 score + evidence 누적 — 4 tier (shot×{strong,weak}, summary×{strong,weak}).""" out: dict[str, dict] = {} for label, kw in keyword_map.items(): strong = kw.get("strong", []) or [] weak = kw.get("weak", []) or [] evidence: list[str] = [] score = 0.0 for kws, mult, kind in ((strong, STRONG_MULT, "strong"), (weak, WEAK_MULT, "weak")): for k in kws: if k in shot_description: score += SHOT_WEIGHT * mult evidence.append(f"shot/{kind}/{k}") elif k in scene_summary: score += SUMMARY_WEIGHT * mult evidence.append(f"summary/{kind}/{k}") out[label] = {"score": round(score, 2), "evidence": evidence} return out def _pick_top(scored: dict[str, dict], min_score: int, tie_margin: int = 1) -> tuple[str, dict, list[str]]: """returns (chosen_label, chosen_dict, notes). 조건 미달 시 ('', {}, notes).""" sortable = sorted(scored.items(), key=lambda kv: kv[1]["score"], reverse=True) if not sortable: return "", {}, ["no labels"] top_label, top_val = sortable[0] if top_val["score"] < min_score: return "", top_val, [ f"top={top_label} score={top_val['score']} < min={min_score}" ] if len(sortable) >= 2: second_val = sortable[1][1]["score"] if top_val["score"] - second_val < tie_margin: return "", top_val, [ f"tie top={top_label}({top_val['score']}) vs {sortable[1][0]}({second_val})" ] return top_label, top_val, [] def classify_sub_space(shot: ShotMeta) -> tuple[str, str, list[str]]: """returns (label, evidence_str, notes). label='manual_review_needed' for low/tie score.""" scored = _score_keywords(SUB_SPACE_KEYWORDS, shot.shot_description, shot.scene_summary) chosen, top, notes = _pick_top(scored, SUB_SPACE_MIN_SCORE, SUB_SPACE_TIE_MARGIN) if not chosen: score_brief = ", ".join(f"{k}={v['score']}" for k, v in scored.items() if v["score"] > 0) return "manual_review_needed", score_brief or "no hits", notes return chosen, ",".join(top["evidence"]), [ f"sub_space score={top['score']} evidence={top['evidence']}" ] def classify_state(shot: ShotMeta) -> tuple[str, str, list[str]]: """returns (label, evidence_str, notes). negation guard + cleaned priority.""" # 1. negation guards 먼저 — character mark / "시신 없는" 같은 표현 무력화 text_full = f"{shot.shot_description}\n{shot.scene_summary}" vandalized_blocked = any(re.search(p, text_full) for p in VANDALIZED_NEGATION_CONTEXTS) corpse_blocked = any(re.search(p, text_full) for p in CORPSE_NEGATION_CONTEXTS) scored = _score_keywords(STATE_KEYWORDS, shot.shot_description, shot.scene_summary) if vandalized_blocked and "vandalized" in scored: scored["vandalized"] = { "score": 0.0, "evidence": ["BLOCKED_BY_NEGATION (character mark)"], } if corpse_blocked and "corpse_marks" in scored: scored["corpse_marks"] = { "score": 0.0, "evidence": ["BLOCKED_BY_NEGATION (시신/피 없는 표현)"], } # 2. cleaned overrides corpse_marks (negation 의도) if CLEANED_OVERRIDES_CORPSE: cleaned_score = scored.get("cleaned", {}).get("score", 0) corpse_score = scored.get("corpse_marks", {}).get("score", 0) if cleaned_score >= STRONG_WEIGHT and corpse_score > 0: # corpse score 깎음 (cleaned 우위) scored["corpse_marks"] = { "score": max(0, corpse_score - STRONG_WEIGHT), "evidence": scored["corpse_marks"]["evidence"] + ["DOWNGRADED_BY_CLEANED"], } # 3. 기본값 normal — 다른 어느 state 도 strong 이상 안 잡혔으면 normal max_other = max( (v["score"] for k, v in scored.items() if k != "normal"), default=0, ) if max_other < STATE_MIN_SCORE: return "normal", "no strong state evidence", [ "state=normal (fallback — no other state hit STATE_MIN_SCORE)" ] # 4. pick top non-normal non_normal = {k: v for k, v in scored.items() if k != "normal"} chosen, top, notes = _pick_top(non_normal, STATE_MIN_SCORE, tie_margin=1) if not chosen: return "normal", "fallback (low/tie)", notes + ["fallback to normal"] return chosen, ",".join(top["evidence"]), [ f"state score={top['score']} evidence={top['evidence']}" ] def classify_time(shot: ShotMeta) -> tuple[str, str, list[str]]: scored = _score_keywords(TIME_KEYWORDS, shot.shot_description, shot.scene_summary) # time 은 약하게 — 1점이라도 있으면 채용. 동률 시 priority: night > dusk > morning > day. time_priority = ["night", "dusk", "morning", "day"] by_score = sorted(scored.items(), key=lambda kv: (-kv[1]["score"], time_priority.index(kv[0]))) top_label, top_val = by_score[0] if top_val["score"] < TIME_MIN_SCORE: return "day", "fallback (no time keyword)", ["time=day fallback"] return top_label, ",".join(top_val["evidence"]), [ f"time score={top_val['score']} evidence={top_val['evidence']}" ] def assign_camera_slot(shot: ShotMeta, sub_space: str) -> tuple[str, list[str]]: notes: list[str] = [] if sub_space == "manual_review_needed": return "fallback_unknown", ["sub_space=manual_review_needed → slot 결정 불가"] candidates = [(label, desc) for (sp, label, desc) in CAMERA_SLOT_POOL if sp == sub_space] if not candidates: return "fallback_unknown", [f"sub_space={sub_space} 의 camera_slot 정의 부재"] text = f"{shot.scene_summary} {shot.shot_description}" framing_hits: list[str] = [] if re.search(r"\bwide\b", text, re.IGNORECASE) or re.search(WIDE_TEXT_HINTS, text): framing_hits.append("wide") if re.search(r"\bclose\b", text, re.IGNORECASE) or re.search(CLOSE_TEXT_HINTS, text): framing_hits.append("close") for fr in framing_hits: for preferred in FRAMING_KEYWORD_MAP.get(fr, []): for label, _desc in candidates: if label == preferred: notes.append(f"framing 키워드 '{fr}' → {label}") return label, notes default_label = candidates[0][0] notes.append(f"framing hit 없음 → 기본 {default_label}") return default_label, notes # --------------------------------------------------------------------------- # Plan / outputs # --------------------------------------------------------------------------- def build_shot_plans(shots: list[ShotMeta]) -> list[ShotPlan]: plans: list[ShotPlan] = [] for s in shots: sub, sub_ev, sub_notes = classify_sub_space(s) state, st_ev, st_notes = classify_state(s) time, t_ev, t_notes = classify_time(s) slot, slot_notes = assign_camera_slot(s, sub) notes = [ f"sub_space={sub} (evidence={sub_ev or 'no match'})", *sub_notes, f"state={state} (evidence={st_ev or 'fallback'})", *st_notes, f"time={time} (evidence={t_ev or 'fallback'})", *t_notes, *slot_notes, ] manual = sub == "manual_review_needed" or slot == "fallback_unknown" plans.append(ShotPlan( shot=s, sub_space=sub, state=state, time=time, camera_slot=slot, manual_review_needed=manual, classification_notes=notes, )) return plans def build_space_bible(plans: list[ShotPlan], canon_desc: str, existing_prompts: list[str]) -> dict: used_sub_spaces = sorted({p.sub_space for p in plans}) bible: dict = { "location_short_id": L05_SHORT_ID, "location_name": "옥탑방 내부", "canon_description": canon_desc, "classifier_weights": { "shot_weight": SHOT_WEIGHT, "summary_weight": SUMMARY_WEIGHT, "strong_mult": STRONG_MULT, "weak_mult": WEAK_MULT, "sub_space_min_score": SUB_SPACE_MIN_SCORE, "sub_space_tie_margin": SUB_SPACE_TIE_MARGIN, "state_min_score": STATE_MIN_SCORE, }, "sub_spaces": {}, "fixed_geometry_source": "entity_canon.description + 기존 chain_bg prompts (텍스트)", "existing_chain_bg_prompt_count": len(existing_prompts), "notes": [ "v1: dry-run 단계 — fixed geometry 자동 추출 미수행 (텍스트 키워드 기반 sub_space 분류만).", "Codex Important 1: metric 추정 금지. qualitative impression 만.", "Codex Blocking 2: plate 단계 인체 0. 시신 silhouette 도 금지.", ], } for sub in used_sub_spaces: related = [p for p in plans if p.sub_space == sub] bible["sub_spaces"][sub] = { "shot_count": len(related), "shot_labels": [p.shot.label for p in related], "camera_slots_used": sorted({p.camera_slot for p in related}), "confidence": "low" if sub == "manual_review_needed" else "high", } return bible def build_base_plate_prompt(sub_space: str, slot_label: str, slot_desc: str, canon_desc: str) -> str: # Codex Blocking 3: numeric metric (m, mm, 1024x768) 제거. qualitative 표현만. return ( f"# Base plate — {sub_space} :: {slot_label}\n" f"# Camera slot: {slot_desc}\n" f"# Location canon description: {canon_desc}\n\n" "Photorealistic, documentary-grade interior reference photograph of a small Korean " "rooftop apartment interior, west-coast town setting. Empty room. No people, no " "characters, no animals, no human silhouettes. Time: daytime, normal/clean state. " f"Sub-space: {sub_space}. Camera framing: {slot_desc}. Lens impression: standard wide-normal, " "eye-level human standing height unless the slot description specifies otherwise. " "Focus on architectural geometry (walls, doors, windows) and fixed furniture. " "Lighting: soft natural daylight. Style: shot on a mirrorless camera, slight grain, " "no cinematic LUT, no story moment, no narrative props." ) def build_shot_plate_prompt(plan: ShotPlan, base_plate_filename: str) -> str: state_overlay = { "normal": "Same daytime, normal/clean baseline. No additional alteration.", "corpse_marks": ( "Dark blood stain on the floor near the wall (NO body, NO silhouette). " "Disturbed bedding scattered. Knocked-over chair. Heavy shadows. " "ABSOLUTELY NO human figure, NO corpse, NO body parts." ), "cleaned": ( "Room cleaned and tidied. Furniture restored to neutral arrangement. " "No stains, no debris." ), "empty": ( "Empty room, signs of recent abandonment (open cabinet, unfinished cup on the table). " "NO people, NO animals." ), "vandalized": ( "Room ransacked: knocked-over furniture, scattered papers, dark red circular symbol " "painted on the wall. NO people, NO animals." ), }.get(plan.state, "Normal baseline.") time_overlay = { "morning": "Cold pale morning light entering through a single small window.", "day": "Soft warm midday sunlight through the window, gentle interior shadows.", "dusk": "Late-afternoon orange light, long shadows, fading warmth.", "night": "Night interior, lamp light or moonlight only, deep shadow regions, low key.", }.get(plan.time, "Soft natural light.") return ( f"# Source base plate: {base_plate_filename}\n" f"# Shot: {plan.shot.label} (still_id={plan.shot.still_id})\n" f"# Sub-space: {plan.sub_space}\n" f"# Camera slot: {plan.camera_slot}\n" f"# State: {plan.state}\n" f"# Time: {plan.time}\n\n" "I2I edit instruction — keep the spatial geometry, walls, doors, windows, and fixed " "furniture of the source base plate IDENTICAL. Edit only the lighting, state, and " "environmental traces as described. ABSOLUTELY NO human figures, NO character " "silhouettes, NO body parts, NO living animals. Environmental marks only.\n\n" f"State overlay: {state_overlay}\n" f"Time/lighting overlay: {time_overlay}\n\n" "Scene context (for atmospheric reference only, do NOT draw people or events literally): " f"{plan.shot.scene_summary}\n" f"Shot description (for atmospheric reference only): {plan.shot.shot_description}" ) # --------------------------------------------------------------------------- # Cost estimate # --------------------------------------------------------------------------- def estimate_cost(unique_base_plates: int, shot_plate_count: int, plan_lvm: bool, retry_factor: float = 1.2) -> dict: raw_image_calls = (unique_base_plates + shot_plate_count) * retry_factor raw_lvm_calls = (unique_base_plates + shot_plate_count) * (1.0 if plan_lvm else 0.0) image_calls = round(raw_image_calls, 2) lvm_calls = round(raw_lvm_calls, 2) return { "unique_base_plates": unique_base_plates, "shot_plate_count": shot_plate_count, "retry_factor": retry_factor, "image_calls_est": image_calls, "lvm_calls_est": lvm_calls, "usd_estimate": round( image_calls * GPT_IMAGE_2_UNIT_COST + lvm_calls * GPT5_VISION_UNIT_COST, 3 ), "unit_costs": { "gpt-image-2_per_image_usd": GPT_IMAGE_2_UNIT_COST, "gpt-5_vision_per_call_usd": GPT5_VISION_UNIT_COST, }, } # --------------------------------------------------------------------------- # OpenAI client + image generate + LVM wrapper (Phase D / F) # --------------------------------------------------------------------------- def _make_openai_client(): """production _resolve_openai_client 와 동일 패턴. api_key 는 settings.openai_api_key 명시 전달 (bare OpenAI() 는 .env 인식 X). timeout 은 settings.llm_timeout_image_gen. """ from openai import OpenAI # noqa: PLC0415 from app.core.config import settings # noqa: PLC0415 return OpenAI( api_key=settings.openai_api_key, timeout=float(settings.llm_timeout_image_gen), ) def _append_jsonl(path: Path, payload: dict) -> None: with path.open("a", encoding="utf-8") as f: f.write(json.dumps(payload, ensure_ascii=False)) f.write("\n") def generate_base_plate( *, client, image_model: str, prompt: str, out_path: Path, size: str = "1536x864", quality: str = "high", max_attempts: int = 2, cost_log_path: Optional[Path] = None, llm_meta_log_path: Optional[Path] = None, plate_id: str = "", ) -> dict: """base plate text-only generate. ref 없음 (production background_render text-only path 패턴).""" info: dict = { "plate_id": plate_id, "model": image_model, "size": size, "quality": quality, "status": "failed", "attempts": 0, "final_error": None, } last_exc: Optional[Exception] = None for attempt in range(1, max_attempts + 1): info["attempts"] = attempt t0 = time.perf_counter() try: resp = client.images.generate( model=image_model, prompt=prompt, size=size, quality=quality, n=1, ) b64 = resp.data[0].b64_json if resp and resp.data else None if not b64: raise RuntimeError("empty b64 response") out_path.write_bytes(base64.b64decode(b64)) ms = int((time.perf_counter() - t0) * 1000) info["status"] = "ok" info["png_path"] = str(out_path) info["latency_ms"] = ms if cost_log_path is not None: _append_jsonl(cost_log_path, { "stage": "base_plate", "plate_id": plate_id, "model": image_model, "size": size, "quality": quality, "status": "ok", "latency_ms": ms, }) if llm_meta_log_path is not None: _append_jsonl(llm_meta_log_path, { "kind": "image_generate", "stage": "base_plate", "plate_id": plate_id, "model": image_model, "prompt_chars": len(prompt), "size": size, "quality": quality, "latency_ms": ms, "status": "ok", }) return info except Exception as exc: last_exc = exc ms = int((time.perf_counter() - t0) * 1000) if cost_log_path is not None: _append_jsonl(cost_log_path, { "stage": "base_plate", "plate_id": plate_id, "model": image_model, "size": size, "quality": quality, "status": "error", "latency_ms": ms, "error": str(exc)[:200], "attempt": attempt, }) if attempt >= max_attempts: info["final_error"] = str(exc)[:300] return info time.sleep(2 * attempt) if last_exc is not None: info["final_error"] = str(last_exc)[:300] return info # realized_spatial_card strict JSON schema (plan v2 §4 Stage 6) REALIZED_SPATIAL_CARD_SCHEMA: dict = { "type": "object", "additionalProperties": False, "properties": { "plate_id": {"type": "string"}, "plate_kind": {"type": "string", "enum": ["base", "shot"]}, "space_key": {"type": "string"}, "camera_slot": {"type": "string"}, "fixed_objects": { "type": "array", "items": { "type": "object", "additionalProperties": False, "properties": { "label": {"type": "string"}, "bbox_xyxy_norm": { "type": "array", "items": {"type": "number"}, "minItems": 4, "maxItems": 4, }, "zone_label": {"type": "string"}, "confidence": {"type": "number"}, "confidence_band": {"type": "string", "enum": ["trusted", "weak", "unknown"]}, }, "required": ["label", "bbox_xyxy_norm", "zone_label", "confidence", "confidence_band"], }, }, "depth_zones": { "type": "object", "additionalProperties": False, "properties": { "foreground": {"type": "string"}, "midground": {"type": "string"}, "background": {"type": "string"}, }, "required": ["foreground", "midground", "background"], }, "camera_impression": { "type": "object", "additionalProperties": False, "properties": { "view_text": {"type": "string"}, "eye_level_qualitative": { "type": "string", "enum": ["low", "eye_level", "high", "unknown"], }, "framing_qualitative": { "type": "string", "enum": ["close", "normal", "wide", "unknown"], }, }, "required": ["view_text", "eye_level_qualitative", "framing_qualitative"], }, "placement_zones": { "type": "array", "items": { "type": "object", "additionalProperties": False, "properties": { "label": {"type": "string"}, "bbox_xyxy_norm": { "type": "array", "items": {"type": "number"}, "minItems": 4, "maxItems": 4, }, "zone_label": {"type": "string"}, "usable_for": {"type": "string"}, }, "required": ["label", "bbox_xyxy_norm", "zone_label", "usable_for"], }, }, "not_visible": {"type": "array", "items": {"type": "string"}}, "overall_confidence": {"type": "number"}, "overall_confidence_band": { "type": "string", "enum": ["trusted", "weak", "unknown"], }, }, "required": [ "plate_id", "plate_kind", "space_key", "camera_slot", "fixed_objects", "depth_zones", "camera_impression", "placement_zones", "not_visible", "overall_confidence", "overall_confidence_band", ], } LVM_INSTRUCTION_TEMPLATE = ( "당신은 이미지에 실제로 보이는 것만 evidence 로 기술하는 시각 검증자입니다. " "추측 금지. 보이지 않는 것은 `not_visible` 리스트에 적으세요. " "fixed_objects 의 bbox_xyxy_norm 은 0.0~1.0 사이 normalized coordinate. " "zone_label 은 3x3 grid: foreground|midground|background_x_left|center|right (예: 'midground_center'). " "confidence_band: >=0.75 trusted, 0.5~0.75 weak, <0.5 unknown. " "camera_impression 은 metric (mm, m) 표기 금지 — qualitative tag (eye_level/low/high, close/normal/wide) 만. " "\n\nplate metadata: plate_id={plate_id}, plate_kind={plate_kind}, " "space_key={space_key}, camera_slot={camera_slot}. " "이 값을 응답의 동일 필드에 그대로 echo." ) def call_lvm_realized_card( *, image_bytes: bytes, plate_id: str, plate_kind: str, space_key: str, camera_slot: str, lvm_model: str, max_attempts: int = 2, cost_log_path: Optional[Path] = None, llm_meta_log_path: Optional[Path] = None, ) -> dict: """LVM realized_spatial_card 호출. _call_openai_vision 패턴 모방 + custom schema.""" import socket as _socket # noqa: PLC0415 import urllib.error as _urlerr # noqa: PLC0415 import urllib.request as _urlreq # noqa: PLC0415 from app.core.config import settings # noqa: PLC0415 api_key = settings.openai_api_key if not api_key: raise RuntimeError("OPENAI_API_KEY 미설정 — LVM 호출 불가") text_prompt = LVM_INSTRUCTION_TEMPLATE.format( plate_id=plate_id, plate_kind=plate_kind, space_key=space_key, camera_slot=camera_slot, ) b64 = base64.b64encode(image_bytes).decode("ascii") body: dict = { "model": lvm_model, "input": [{ "type": "message", "role": "user", "content": [ {"type": "input_text", "text": text_prompt}, {"type": "input_image", "image_url": f"data:image/png;base64,{b64}"}, ], }], "text": { "format": { "type": "json_schema", "name": "realized_spatial_card", "strict": True, "schema": REALIZED_SPATIAL_CARD_SCHEMA, } }, "store": False, } # gpt-5* 는 non-default temperature 거부 (ref_image_pipeline.py:120 기록). # 다른 모델만 명시 — gpt-5 family 면 default 만 사용. if not lvm_model.startswith("gpt-5"): body["temperature"] = 0.1 req = _urlreq.Request( "https://api.openai.com/v1/responses", data=json.dumps(body).encode("utf-8"), headers={ "Authorization": f"Bearer {api_key}", "Content-Type": "application/json", }, method="POST", ) timeout = settings.llm_timeout_validation last_exc: Optional[Exception] = None for attempt in range(1, max_attempts + 1): t0 = time.perf_counter() try: with _urlreq.urlopen(req, timeout=timeout) as resp: payload = json.loads(resp.read().decode("utf-8")) ms = int((time.perf_counter() - t0) * 1000) # Responses API: output[0].content[0].text → JSON string (strict schema 적용). output = payload.get("output", []) text_block = None for item in output: for c in item.get("content", []) or []: if c.get("type") in ("output_text", "text") and c.get("text"): text_block = c["text"] break if text_block: break if not text_block: raise RuntimeError(f"empty LVM text response: {json.dumps(payload)[:200]}") card = json.loads(text_block) if cost_log_path is not None: _append_jsonl(cost_log_path, { "stage": "lvm", "plate_id": plate_id, "model": lvm_model, "status": "ok", "latency_ms": ms, }) if llm_meta_log_path is not None: _append_jsonl(llm_meta_log_path, { "kind": "lvm_call", "stage": "lvm", "plate_id": plate_id, "model": lvm_model, "latency_ms": ms, "status": "ok", }) return {"status": "ok", "card": card, "latency_ms": ms, "attempts": attempt} except (_urlerr.HTTPError, _urlerr.URLError, _socket.timeout, RuntimeError, json.JSONDecodeError) as exc: last_exc = exc ms = int((time.perf_counter() - t0) * 1000) if cost_log_path is not None: _append_jsonl(cost_log_path, { "stage": "lvm", "plate_id": plate_id, "model": lvm_model, "status": "error", "latency_ms": ms, "error": str(exc)[:200], "attempt": attempt, }) if attempt >= max_attempts: return {"status": "failed", "error": str(exc)[:300], "attempts": attempt} time.sleep(2 * attempt) return {"status": "failed", "error": str(last_exc)[:300] if last_exc else "unknown", "attempts": max_attempts} # --------------------------------------------------------------------------- # Webserver (background, 사용자 추가 요구) # --------------------------------------------------------------------------- def _find_free_port(start: int, attempts: int = 6, bind: str = "127.0.0.1") -> Optional[int]: import socket as _socket # noqa: PLC0415 for offset in range(attempts): port = start + offset s = _socket.socket(_socket.AF_INET, _socket.SOCK_STREAM) try: s.setsockopt(_socket.SOL_SOCKET, _socket.SO_REUSEADDR, 1) s.bind((bind, port)) s.close() return port except OSError: continue finally: try: s.close() except Exception: pass return None def start_static_webserver(run_dir: Path, port: int, bind: str = "127.0.0.1") -> dict: """python -m http.server 백그라운드 launch. run_dir 만 serve. Codex Important 2: Popen 직후 짧은 sleep + proc.poll() — bind/race/import 실패 catch. """ import subprocess # noqa: PLC0415 free = _find_free_port(port, attempts=6, bind=bind) if free is None: return {"status": "no_free_port", "tried_from": port} log_path = run_dir / "_serve.log" log_fh = log_path.open("w", encoding="utf-8") proc = subprocess.Popen( # noqa: S603 — 사용자 명시 요구 [sys.executable, "-m", "http.server", str(free), "--bind", bind], cwd=str(run_dir), stdout=log_fh, stderr=log_fh, ) # bind/race/import 실패 검출 — 0.6s 후 alive 확인. time.sleep(0.6) rc = proc.poll() if rc is not None: try: log_tail = log_path.read_text(encoding="utf-8", errors="replace")[-400:] except Exception: log_tail = "" return { "status": "died_early", "pid": proc.pid, "port": free, "bind": bind, "exit_code": rc, "log": str(log_path), "log_tail": log_tail, } return { "status": "started", "pid": proc.pid, "port": free, "bind": bind, "log": str(log_path), "cwd": str(run_dir), } # --------------------------------------------------------------------------- # Writers # --------------------------------------------------------------------------- def _safe_tsv_cell(v) -> str: return str(v).replace("\t", " ").replace("\r", " ").replace("\n", "\\n") def write_tsv(path: Path, rows: list[list]) -> None: with path.open("w", encoding="utf-8") as f: for r in rows: f.write("\t".join(_safe_tsv_cell(c) for c in r)) f.write("\n") def render_run_index_html( run_dir: Path, plans: list[ShotPlan], base_plate_map: dict, run_meta: dict, base_results: Optional[dict] = None, lvm_results: Optional[dict] = None, ) -> None: """Enhanced HTML report — base/shot plate 이미지 inline + LVM 요약 + manual_review/failed 표.""" base_results = base_results or {} lvm_results = lvm_results or {} def _img_or_placeholder(rel_path: str, alt: str) -> str: abs_path = run_dir / rel_path if abs_path.exists(): return ( f'' f'{html.escape(alt)}' ) return ( f'
' f"미생성 ({html.escape(rel_path)})
" ) # --- Base plate section --- base_cards = [] for (sub, slot), info in base_plate_map.items(): plate_id = f"base__{sub}__{slot}" rel_png = f"base_plates/{sub}__{slot}.png" rel_prompt = f"prompts/base/{sub}__{slot}.txt" rel_lvm = f"realized_spatial_cards/{plate_id}.json" base_info = base_results.get(plate_id, {}) lvm_info = lvm_results.get(plate_id, {}) gen_status = base_info.get("status", "not_generated") lvm_status = lvm_info.get("status", "not_run") lvm_card = lvm_info.get("card") if lvm_info else None if lvm_card: fixed = [f["label"] for f in lvm_card.get("fixed_objects", [])[:5]] band = lvm_card.get("overall_confidence_band", "?") not_vis = lvm_card.get("not_visible", [])[:3] framing = lvm_card.get("camera_impression", {}).get("framing_qualitative", "?") lvm_summary = ( f"band={band} | framing={framing} | " f"fixed={', '.join(fixed) or '(none)'} | " f"not_visible={', '.join(not_vis) or '(none)'}" ) elif lvm_status == "failed": lvm_summary = f'FAILED: {html.escape((lvm_info.get("error") or "")[:100])}' else: lvm_summary = "(LVM 미실행)" gen_badge = { "ok": 'OK', "failed": f'FAILED: {html.escape((base_info.get("final_error") or "")[:80])}', "not_generated": '미생성', }.get(gen_status, gen_status) base_cards.append(f"""
{_img_or_placeholder(rel_png, plate_id)}
{html.escape(plate_id)}
sub_space: {html.escape(sub)} · camera_slot: {html.escape(slot)}
slot_desc: {html.escape(info['slot_desc'])}
generate: {gen_badge} {'· ' + str(base_info.get('latency_ms', '?')) + 'ms' if base_info.get('latency_ms') else ''}
prompt: {html.escape(rel_prompt)}
LVM: {lvm_summary} {' · card' if lvm_card else ''}
""") base_section_html = "".join(base_cards) if base_cards else "

(base plate plan 없음)

" # --- Shot plate section (이번 wave 는 placeholder. shots stage 진입 시 채움) --- shot_cards = [] plate_ready_shots = [p for p in plans if not p.manual_review_needed] for p in plate_ready_shots: rel_png = f"shot_plates/{p.shot.label}.png" rel_prompt = f"prompts/shot/{p.shot.label}.txt" plate_id = f"shot__{p.shot.label}" rel_lvm = f"realized_spatial_cards/{plate_id}.json" lvm_info = lvm_results.get(plate_id, {}) lvm_card = lvm_info.get("card") if lvm_info else None if lvm_card: band = lvm_card.get("overall_confidence_band", "?") framing = lvm_card.get("camera_impression", {}).get("framing_qualitative", "?") lvm_summary = f"band={band} | framing={framing}" else: lvm_summary = "(LVM 미실행)" shot_cards.append(f"""
{_img_or_placeholder(rel_png, p.shot.label)}
{html.escape(p.shot.label)}
sub_space: {html.escape(p.sub_space)} · camera_slot: {html.escape(p.camera_slot)}
state: {html.escape(p.state)} · time: {html.escape(p.time)}
summary: {html.escape(p.shot.scene_summary[:90])}
shot_desc: {html.escape(p.shot.shot_description[:90])}
prompt: {html.escape(rel_prompt)}
LVM: {lvm_summary}{' · card' if lvm_card else ''}
""") shot_section_html = "".join(shot_cards) if shot_cards else "

(plate_ready shot 없음)

" # --- manual_review / skipped / failed 표 (사용자 요구 #4) --- manual_rows = "".join( f"{html.escape(p.shot.label)}" f"{html.escape(p.sub_space)}" f"{html.escape(p.camera_slot)}" f"{html.escape(p.state)}" f"{html.escape(p.time)}" f"{html.escape(' | '.join(p.classification_notes[:3]))}" f"{html.escape(p.shot.scene_summary[:90])}" for p in plans if p.manual_review_needed ) failed_rows = "".join( f"{html.escape(pid)}" f"base" f"{html.escape(str(info.get('attempts', '?')))}" f"{html.escape((info.get('final_error') or '')[:200])}" for pid, info in base_results.items() if info.get("status") == "failed" ) + "".join( f"{html.escape(pid)}" f"lvm" f"{html.escape(str(info.get('attempts', '?')))}" f"{html.escape((info.get('error') or '')[:200])}" for pid, info in lvm_results.items() if info.get("status") == "failed" ) cost_pre = html.escape(json.dumps(run_meta.get("cost_estimate", {}), ensure_ascii=False, indent=2)) args_pre = html.escape(json.dumps(run_meta.get("args", {}), ensure_ascii=False, indent=2)) serve_info = run_meta.get("serve_info") or {} serve_pre = html.escape(json.dumps(serve_info, ensure_ascii=False, indent=2)) if serve_info else "" body = f""" Rooftop Spatial BG — {html.escape(run_meta['run_id'])}

Rooftop Spatial BG Experiment

run_id {html.escape(run_meta['run_id'])} · plan v2 · phase {html.escape(run_meta.get('phase', '?'))}

stages: {html.escape(','.join(run_meta.get('stages_run', [])))} · base_plates: {len(base_plate_map)} · shots: {len(plans)} · manual_review: {sum(1 for p in plans if p.manual_review_needed)}

{('

webserver: http://' + html.escape(serve_info.get('bind', '?')) + ':' + str(serve_info.get('port', '?')) + '/ (PID ' + str(serve_info.get('pid', '?')) + ', log: ' + html.escape(str(serve_info.get('log', ''))) + ')

') if serve_info.get('status') == 'started' else ''}

1. Base plates ({len(base_plate_map)})

{base_section_html}

2. Shot plates ({len(plate_ready_shots)})

{shot_section_html}

3. Realized spatial cards (LVM, {sum(1 for v in lvm_results.values() if v.get('status') == 'ok')})

{('' + ''.join( '' f'' f'' f'' f'' f'' f'' f'' f'' f'' f'' f'' '' for pid, res in lvm_results.items() if res.get("status") == "ok" ) + '
plate_idkindspace_keycamera_slotbandframingeye_levelfixed (top)placement_zonesnot_visiblejson
{html.escape(pid)}{html.escape((res.get("card") or {}).get("plate_kind", "?"))}{html.escape((res.get("card") or {}).get("space_key", "?"))}{html.escape((res.get("card") or {}).get("camera_slot", "?"))}{html.escape((res.get("card") or {}).get("overall_confidence_band", "?"))}{html.escape(((res.get("card") or {}).get("camera_impression") or {}).get("framing_qualitative", "?"))}{html.escape(((res.get("card") or {}).get("camera_impression") or {}).get("eye_level_qualitative", "?"))}{html.escape(", ".join(f["label"] for f in ((res.get("card") or {}).get("fixed_objects") or [])[:5]) or "(none)")}{html.escape(", ".join(z["label"] for z in ((res.get("card") or {}).get("placement_zones") or [])[:3]) or "(none)")}{html.escape(", ".join(((res.get("card") or {}).get("not_visible") or [])[:3]) or "(none)")}json
') if any(r.get("status") == "ok" for r in lvm_results.values()) else '

(LVM 미실행 또는 결과 0)

'}

4. Final scene context cards (Phase G — 미구현)

아직 생성 안 됨. base+shot+LVM 완료 + 사용자 검토 통과 후 별도 wave 에서 추가 예정.

5. Manual review / skipped ({sum(1 for p in plans if p.manual_review_needed)})

{manual_rows or ''}
shotsub_spacecamera_slotstatetime classification_notessummary
(없음)

6. Failed

{failed_rows or ''}
plate_idkindattemptserror
(없음)

7. Cost estimate

{cost_pre}

8. Run args

{args_pre}
""" (run_dir / "index.html").write_text(body, encoding="utf-8") def write_index_html_stub(run_dir: Path, plans: list[ShotPlan], base_plate_map: dict, run_meta: dict) -> None: shot_rows = "".join( "" f"{html.escape(p.shot.label)}" f"{html.escape(p.sub_space)}" f"{html.escape(p.camera_slot)}" f"{html.escape(p.state)}" f"{html.escape(p.time)}" f"{'⚠️ manual_review' if p.manual_review_needed else ''}" f"{html.escape(p.shot.scene_summary[:80])}" "" for p in plans ) base_rows = "".join( "" f"{html.escape(sub)}" f"{html.escape(slot)}" f"{html.escape(info['slot_desc'])}" "" for (sub, slot), info in base_plate_map.items() ) cost_pre = html.escape(json.dumps(run_meta["cost_estimate"], ensure_ascii=False, indent=2)) args_pre = html.escape(json.dumps(run_meta["args"], ensure_ascii=False, indent=2)) body = f""" Rooftop Spatial BG Experiment — {html.escape(run_meta['run_id'])} (dry-run)

Rooftop Spatial BG Experiment — dry-run

run_id {html.escape(run_meta['run_id'])} · plan v2 · phase C dry-run

API 호출 0. Plate / LVM card 미생성. Phase D+ 이후 wave.

Base plate plan ({len(base_plate_map)})

{base_rows}
sub_spacecamera_slotdesc

Shot plan ({len(plans)})

{shot_rows}
shotsub_spacecamera_slotstatetime reviewsummary

Cost estimate

{cost_pre}

Args

{args_pre}
""" (run_dir / "index.html").write_text(body, encoding="utf-8") def write_dry_run_outputs(run_dir: Path, plans: list[ShotPlan], bible: dict, base_plate_map: dict, run_meta: dict) -> None: run_dir.mkdir(parents=True, exist_ok=True) (run_dir / "prompts" / "base").mkdir(parents=True, exist_ok=True) (run_dir / "prompts" / "shot").mkdir(parents=True, exist_ok=True) (run_dir / "space_bible.json").write_text( json.dumps(bible, ensure_ascii=False, indent=2), encoding="utf-8" ) header = [ "shot_label", "scene_index", "shot_index", "still_id", "sub_space", "camera_slot", "state", "time", "manual_review_needed", "dependent_scene_id", "visible_short_ids", "scene_summary", "shot_description", "classification_notes", ] rows: list[list] = [header] for p in plans: rows.append([ p.shot.label, p.shot.scene_index, p.shot.shot_index, p.shot.still_id, p.sub_space, p.camera_slot, p.state, p.time, "true" if p.manual_review_needed else "false", p.shot.dependent_scene_id or "", ",".join(p.shot.visible_short_ids), p.shot.scene_summary, p.shot.shot_description, " | ".join(p.classification_notes), ]) write_tsv(run_dir / "shot_background_plan.tsv", rows) for (sub, slot), info in base_plate_map.items(): filename = f"{sub}__{slot}.txt" (run_dir / "prompts" / "base" / filename).write_text(info["prompt"], encoding="utf-8") for p in plans: path = run_dir / "prompts" / "shot" / f"{p.shot.label}.txt" if p.manual_review_needed: body = ( f"# Shot: {p.shot.label}\n" f"# manual_review_needed=true — sub_space={p.sub_space}, slot={p.camera_slot}\n" "# 자동 생성 불가. classification_notes:\n" + "\n".join(f"# - {n}" for n in p.classification_notes) ) else: base_filename = f"{p.sub_space}__{p.camera_slot}.png" body = build_shot_plate_prompt(p, base_filename) path.write_text(body, encoding="utf-8") (run_dir / "run_meta.json").write_text( json.dumps(run_meta, ensure_ascii=False, indent=2), encoding="utf-8" ) (run_dir / "cost_log.jsonl").write_text("", encoding="utf-8") (run_dir / "llm_call_meta.jsonl").write_text("", encoding="utf-8") write_index_html_stub(run_dir, plans, base_plate_map, run_meta) # --------------------------------------------------------------------------- # CLI # --------------------------------------------------------------------------- VALID_STAGES = {"plan", "base", "lvm", "shots", "shots_lvm", "context"} def parse_args() -> argparse.Namespace: ap = argparse.ArgumentParser( description="Rooftop Spatial BG Experiment (standalone, plan v2 — Phase C/D/F)" ) ap.add_argument("--run-id", default=None, help="명시 안 하면 {YYYYMMDD_HHMM}_{short_uuid} 자동") # --generate 활성 시 base/lvm/shots stage 의 실제 API 호출 허용. ap.add_argument("--generate", action="store_true", help="image / LVM API 호출 허용 (Phase D+)") ap.add_argument("--stages", default="plan,base,lvm", help=( "comma-separated. plan(dry-run only) | base | lvm | shots | " "shots_lvm | context. default 'plan,base,lvm' — Codex base-first 권고. " "shots 포함 시 base+lvm 산출물 fail-fast 검증." )) ap.add_argument("--skip-lvm", action="store_true", help="LVM pass 건너뛰기 (stages 에 lvm 있어도 skip)") ap.add_argument("--max-shots", type=int, default=14) ap.add_argument("--max-plates", type=int, default=30, help="base+shot plate 총량 cap (Phase D+ 진입 시 강제)") ap.add_argument("--space-keys", default="", help="comma-separated sub_space 필터 (예: 'main_room,bedroom')") ap.add_argument("--shots", default="", help="comma-separated shot label 필터 (예: 'S5_S2,S12_S4')") ap.add_argument("--resume", default=None, help="이전 run_id 의 plate / lvm 재사용 (--run-id 와 함께 사용)") ap.add_argument("--estimate-only", action="store_true", help="cost estimate + run_meta 출력 후 exit (파일 출력 0)") ap.add_argument("--image-backend", choices=["openai", "gemini"], default="openai", help="default openai (production 일치, plan §5)") ap.add_argument("--image-model", default="gpt-image-2", help="openai backend 의 모델 (default gpt-image-2)") ap.add_argument("--image-size", default="1536x864", help="openai images.generate size (default 1536x864 production 일치)") ap.add_argument("--image-quality", default="high", help="openai images.generate quality (default high)") ap.add_argument("--lvm-model", default="gpt-5", help="LVM 모델 (default gpt-5 vision)") ap.add_argument("--allow-fallback-generate", action="store_true", help="fallback_unknown slot 도 generate 허용 (default OFF, Codex Important 4)") ap.add_argument("--output-base", default=str(DEFAULT_OUTPUT_DIR), help="run dir 의 부모 디렉토리") # webserver options (사용자 추가 요구, Codex Q1 OK with default 127.0.0.1). ap.add_argument("--no-serve", action="store_true", help="HTML webserver 기동 안 함") ap.add_argument("--serve-port", type=int, default=8765, help="webserver 포트 (in-use 면 +1..+5 fallback)") ap.add_argument("--bind", default="127.0.0.1", help="webserver bind 주소. 0.0.0.0 명시 시 외부 노출 경고.") return ap.parse_args() def _parse_stages(arg: str) -> list[str]: out = [] for s in arg.split(","): s = s.strip() if not s: continue if s not in VALID_STAGES: raise SystemExit( f"ERROR: --stages 의 '{s}' 는 유효하지 않음. 허용: {sorted(VALID_STAGES)}" ) out.append(s) if not out: out = ["plan"] return out def main() -> int: args = parse_args() stages = _parse_stages(args.stages) needs_api = any(s in {"base", "lvm", "shots", "shots_lvm"} for s in stages) if needs_api and not args.generate: print( "ERROR: stages 에 base/lvm/shots 가 있으면 --generate 명시 필요 (API 호출 허용).", file=sys.stderr, ) return 2 if args.bind == "0.0.0.0": print( "WARNING: --bind 0.0.0.0 — unauthenticated local network server. " "산출물(prompt/LVM card 시나리오 텍스트) 외부 노출 위험. " "127.0.0.1 권장 (Codex Q1).", file=sys.stderr, ) run_id = args.run_id or f"{datetime.now():%Y%m%d_%H%M}_{uuid.uuid4().hex[:6]}" output_base = Path(args.output_base).resolve() run_dir = output_base / run_id with SessionLocal() as session: canon_desc = load_l05_canon_description(session) if not canon_desc: print(f"ERROR: EntityCanon id={L05_CANON_ID} 부재.", file=sys.stderr) return 3 shots = load_l05_shots(session) if not shots: print("ERROR: L05 visible selected shot 0건.", file=sys.stderr) return 4 existing_chain_bg_prompts = load_existing_chain_bg_prompts(session) if args.shots: wanted = {s.strip() for s in args.shots.split(",") if s.strip()} shots = [s for s in shots if s.label in wanted] if args.max_shots and len(shots) > args.max_shots: shots = shots[: args.max_shots] plans = build_shot_plans(shots) if args.space_keys: wanted_subs = {s.strip() for s in args.space_keys.split(",") if s.strip()} plans = [p for p in plans if p.sub_space in wanted_subs] bible = build_space_bible(plans, canon_desc, existing_chain_bg_prompts) base_plate_map: dict = {} for p in plans: if p.manual_review_needed: continue key = (p.sub_space, p.camera_slot) if key in base_plate_map: continue slot_desc = next( (desc for sp, lbl, desc in CAMERA_SLOT_POOL if sp == p.sub_space and lbl == p.camera_slot), "(slot desc 부재)", ) base_plate_map[key] = { "sub_space": p.sub_space, "slot_label": p.camera_slot, "slot_desc": slot_desc, "prompt": build_base_plate_prompt(p.sub_space, p.camera_slot, slot_desc, canon_desc), } shot_plate_count = sum(1 for p in plans if not p.manual_review_needed) total_plates = len(base_plate_map) + shot_plate_count if args.max_plates and total_plates > args.max_plates: print( f"ERROR: planned plates ({total_plates} = base {len(base_plate_map)} + shot {shot_plate_count}) " f"> --max-plates {args.max_plates}. abort.", file=sys.stderr, ) return 5 cost = estimate_cost( unique_base_plates=len(base_plate_map), shot_plate_count=shot_plate_count, plan_lvm=not args.skip_lvm and ("lvm" in stages or "shots_lvm" in stages), ) run_meta = { "run_id": run_id, "created_at": datetime.now().isoformat(timespec="seconds"), "plan_version": "v2", "phase": f"stages={','.join(stages)}", "stages_run": stages, "project_id": PROJECT_ID, "episode_id": EPISODE_ID, "location_short_id": L05_SHORT_ID, "location_canon_id": L05_CANON_ID, "shot_count": len(plans), "manual_review_needed_count": sum(1 for p in plans if p.manual_review_needed), "unique_base_plates": len(base_plate_map), "cost_estimate": cost, "args": { "generate": args.generate, "stages": stages, "image_backend": args.image_backend, "image_model": args.image_model, "image_size": args.image_size, "image_quality": args.image_quality, "lvm_model": args.lvm_model, "max_shots": args.max_shots, "max_plates": args.max_plates, "space_keys_filter": args.space_keys, "shots_filter": args.shots, "resume": args.resume, "skip_lvm": args.skip_lvm, "allow_fallback_generate": args.allow_fallback_generate, "no_serve": args.no_serve, "serve_port": args.serve_port, "bind": args.bind, }, } if args.estimate_only: print(json.dumps(run_meta, ensure_ascii=False, indent=2)) return 0 # plan stage 산출물 (TSV, prompts, json, stub HTML) — 항상 수행. write_dry_run_outputs(run_dir, plans, bible, base_plate_map, run_meta) # shots 또는 shots_lvm stage 진입 전 prerequisite check (Codex 추가 권고). base_dir = run_dir / "base_plates" if ("shots" in stages or "shots_lvm" in stages) and not any(base_dir.glob("*.png")): print( "ERROR: stages 에 shots 가 있는데 base_plates/*.png 없음. " "base+lvm 먼저 실행 → 사람이 확인 → shots 진입 (Codex base-first).", file=sys.stderr, ) return 6 cost_log_path = run_dir / "cost_log.jsonl" llm_meta_log_path = run_dir / "llm_call_meta.jsonl" base_plates_dir = run_dir / "base_plates" realized_cards_dir = run_dir / "realized_spatial_cards" base_plates_dir.mkdir(parents=True, exist_ok=True) realized_cards_dir.mkdir(parents=True, exist_ok=True) base_results: dict = {} lvm_results: dict = {} if "base" in stages: print(f"[base] generating {len(base_plate_map)} plate(s)...", flush=True) client = _make_openai_client() for (sub, slot), info in base_plate_map.items(): plate_id = f"base__{sub}__{slot}" out_path = base_plates_dir / f"{sub}__{slot}.png" if args.resume and out_path.exists(): base_results[plate_id] = {"status": "ok", "png_path": str(out_path), "attempts": 0, "model": args.image_model, "latency_ms": 0, "resumed": True} print(f" · {plate_id}: resumed") continue res = generate_base_plate( client=client, image_model=args.image_model, prompt=info["prompt"], out_path=out_path, size=args.image_size, quality=args.image_quality, cost_log_path=cost_log_path, llm_meta_log_path=llm_meta_log_path, plate_id=plate_id, ) base_results[plate_id] = res print(f" · {plate_id}: {res['status']} ({res.get('latency_ms', '?')}ms)") if "lvm" in stages and not args.skip_lvm: targets = [(pid, info) for pid, info in base_results.items() if info.get("status") == "ok"] # Codex Blocking 2: lvm 단독 실행 — base PNG 디스크 fallback (--resume 없어도 OK, # 단 base stage 결과 0 이면 fail-fast). if not targets: for (sub, slot), info in base_plate_map.items(): plate_id = f"base__{sub}__{slot}" png = base_plates_dir / f"{sub}__{slot}.png" if png.exists() and plate_id not in base_results: base_results[plate_id] = {"status": "ok", "png_path": str(png), "resumed_for_lvm": True} targets.append((plate_id, base_results[plate_id])) if not targets: print( "ERROR: stages 에 lvm 있는데 base PNG 0장 (base_results=0 + base_plates/*.png 0). " "base stage 먼저 실행하거나 base_plates/ 채워 두세요.", file=sys.stderr, ) return 7 print(f"[lvm] running on {len(targets)} base plate(s)...", flush=True) for plate_id, info in targets: png = Path(info["png_path"]) sub_slot = plate_id.removeprefix("base__") sub, slot = sub_slot.split("__", 1) res = call_lvm_realized_card( image_bytes=png.read_bytes(), plate_id=plate_id, plate_kind="base", space_key=sub, camera_slot=slot, lvm_model=args.lvm_model, cost_log_path=cost_log_path, llm_meta_log_path=llm_meta_log_path, ) lvm_results[plate_id] = res if res.get("status") == "ok" and res.get("card"): (realized_cards_dir / f"{plate_id}.json").write_text( json.dumps(res["card"], ensure_ascii=False, indent=2), encoding="utf-8" ) print(f" · {plate_id}: {res['status']}") # NOTE: shots stage 는 별도 wave 에서 구현. 현재는 prerequisite check 만. if "shots" in stages or "shots_lvm" in stages: print( "NOTE: shots stage 는 아직 구현 안 됨 (이번 wave 는 base+lvm 까지). " "base 확인 후 별도 wave 에서 shots 추가 예정.", file=sys.stderr, ) # serve_info 를 run_meta 에 추가하기 전에 webserver launch (선택) serve_info = None if not args.no_serve: serve_info = start_static_webserver(run_dir, args.serve_port, bind=args.bind) run_meta["serve_info"] = serve_info # Enhanced HTML (이미지 + LVM 카드 inline) render_run_index_html(run_dir, plans, base_plate_map, run_meta, base_results=base_results, lvm_results=lvm_results) (run_dir / "run_meta.json").write_text( json.dumps(run_meta, ensure_ascii=False, indent=2), encoding="utf-8" ) # Codex Important 3: partial failure → exit code 다르게. base_failed = sum(1 for v in base_results.values() if v.get("status") == "failed") lvm_failed = sum(1 for v in lvm_results.values() if v.get("status") == "failed") any_failed = base_failed + lvm_failed > 0 status_word = "COMPLETED_WITH_FAILURES" if any_failed else "OK" print(f"\n{status_word}: run 완료 → {run_dir}") print(f" shots={len(plans)}, manual_review={run_meta['manual_review_needed_count']}, " f"base_plates={len(base_plate_map)}, est USD={cost['usd_estimate']}") if any_failed: print(f" failed: base={base_failed}, lvm={lvm_failed}") print(f" index: {run_dir / 'index.html'}") if serve_info and serve_info.get("status") == "started": url = f"http://{serve_info['bind']}:{serve_info['port']}/index.html" print(f" webserver: {url} (PID {serve_info['pid']}, log {serve_info['log']})") print(f" stop: kill {serve_info['pid']}") elif serve_info and serve_info.get("status") == "died_early": print(f" webserver: DIED_EARLY exit={serve_info.get('exit_code')} log_tail={serve_info.get('log_tail', '')[:200]}", file=sys.stderr) elif serve_info and serve_info.get("status") == "no_free_port": print(f" webserver: NOT STARTED (no free port from {serve_info['tried_from']})", file=sys.stderr) return 8 if any_failed else 0 if __name__ == "__main__": raise SystemExit(main())