#!/usr/bin/env python3 """Background place grouping experiment (dry-run, 2026-05-24). scripts_output/background_place_grouping_experiment// read-only: DB write 0 / network/image 모듈 0 / production code 0. Goal (plan.md): keyword/gap-ledger 방식 대신 place/set/space/state/ structural_version/camera 6축을 명시적으로 분리한 generic 배경 일관성 grouping 방법론 검증. ★★★ Scope (2026-05-24 사용자 명시 지시): 본 실험은 generic background place grouping 방법론 검증이다. L05 (rooftop interior) 는 sample fixture only — 모든 hard-coded L05 데이터 (room whitelist, node ID, project ID 등) 는 SAMPLE_FIXTURE_* prefix 로 격리되어야 하고, generic rule 은 SpaceNode containment/boundary 기반으로 표현되어야 한다. 사용자 standing rule [[feedback-no-scenario-specific-coding]] 참조 — scope leakage 금지 (자세한 violation phrase 목록은 test_*.py 의 TestGenericMethodologyGuard.FORBIDDEN_LEAKAGE_PHRASES). Import scope: - app.core.database.SessionLocal (read-only). - experiment_rooftop_spatial_bg: ShotMeta, load_l05_shots — sample fixture helper only. - build_shot_plans + 다른 production / network / image 모듈 절대 금지. Output 10 files: place_groups.json / set_groups.json / space_nodes.json / state_layers.json / structural_versions.json / shot_bindings.json / generation_unit_plan.json / chain_bg_decomposition.json+tsv / index.html / run_meta.json. """ from __future__ import annotations import argparse import csv import hashlib import html import json import re import sys import uuid from dataclasses import dataclass, field, asdict from datetime import datetime, timezone, timedelta from pathlib import Path from typing import Optional # repo root + env -------------------------------------------------------------- _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)) _SCRIPTS_DIR = Path(__file__).resolve().parent if str(_SCRIPTS_DIR) not in sys.path: sys.path.insert(0, str(_SCRIPTS_DIR)) def _load_backend_env() -> None: 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] if key and key not in _os.environ: _os.environ[key] = value _load_backend_env() from experiment_rooftop_spatial_bg import ( # noqa: E402 ShotMeta, load_l05_shots, ) # ============================================================================ # SAMPLE FIXTURE (L05 rooftop interior) — generic methodology 검증용 표본 only. # Production rules 에 복제 금지. 다른 location 추가 시 동일 contract 로 별도 # fixture 추가하면 됨. # ============================================================================ SAMPLE_FIXTURE_PROJECT_ID = "6cb862d9-590c-4dce-86e6-d10c2977db19" SAMPLE_FIXTURE_EPISODE_ID = "08ad2cd3-3e96-4d84-808f-869ee628473c" SAMPLE_FIXTURE_CANON_ID = "3afbc7a8-b919-4431-a691-0a99057a26ca" SAMPLE_FIXTURE_SHORT_ID = "L05" # legacy aliases — kept for backward compat with helpers/tests that already # reference these names. NEW code should use SAMPLE_FIXTURE_* directly. PROJECT_ID = SAMPLE_FIXTURE_PROJECT_ID EPISODE_ID = SAMPLE_FIXTURE_EPISODE_ID L05_CANON_ID = SAMPLE_FIXTURE_CANON_ID L05_SHORT_ID = SAMPLE_FIXTURE_SHORT_ID DEFAULT_SOURCE_RUN = Path( "scripts_output/rooftop_source_grounding/codex_entry_sanity_gemini_ok" ) DEFAULT_OUTPUT_DIR = Path("scripts_output/background_place_grouping_experiment") PLAN_VERSION = "bpg_w2e" # StateLayer affects enum (plan.md §1-D) STATE_LAYER_AFFECTS_ENUM = { "surface", "furniture", "lighting", "damage", "blood", "emptiness", "weather", } # Geometry-change keyword whitelist (plan.md §1-E) GEOMETRY_CHANGE_KEYWORDS = [ "리모델링", "철거", "폭파", "벽을 부순", "벽을 부수", "가벽 추가", "공사", "재건축", "확장 공사", ] # ---------------------------------------------------------------------------- # SAMPLE FIXTURE — L05 rooftop interior subspace whitelist + node-type sets. # 다른 location 추가 시 동일 형태 fixture 만 새로 정의하면 generic engine 재사용. # generic rule 은 SpaceNode containment/boundary 기반 (LOCATION-AGNOSTIC). # ---------------------------------------------------------------------------- SAMPLE_FIXTURE_L05_SPACE_KEYWORDS: list[tuple[str, str]] = [ ("수리영의 방", "수리영의_방"), ("수리영의_방", "수리영의_방"), ("수리영 방", "수리영의_방"), ("안방", "민숙의_방_안방"), ("침실", "수리영의_방"), # sample variant 기본 침실 = 수리영의_방 ("거실", "거실"), ("식탁", "거실"), ("주방", "주방코너"), ("싱크대", "주방코너"), ("현관", "현관"), ("욕실", "욕실"), ("거울", "욕실"), ] # Sample fixture zone/boundary 노드 — primary_space 결정 시 부모 노드 흡수. # generic rule: node_type ∈ {'zone', 'boundary'} 노드는 contained_in 으로 부모 # room 에 자동 흡수된다 (SpaceNode containment 기반). 아래 set 는 L05 sample 의 # 노드 식별값 — 다른 fixture 는 자기 노드 이름으로 정의. SAMPLE_FIXTURE_L05_ZONE_BOUNDARY_NODES = {"주방코너", "현관"} SAMPLE_FIXTURE_L05_ROOM_NODES = {"거실", "수리영의_방", "민숙의_방_안방", "욕실"} # legacy aliases L05_SPACE_KEYWORDS = SAMPLE_FIXTURE_L05_SPACE_KEYWORDS ZONE_BOUNDARY_NODES = SAMPLE_FIXTURE_L05_ZONE_BOUNDARY_NODES ROOM_NODES = SAMPLE_FIXTURE_L05_ROOM_NODES # State layer keyword (for variant + shot, Codex BLOCKING 1) STATE_LAYER_KEYWORDS: dict[str, list[str]] = { "corpse_marks": ["시신", "시체", "참혹", "쇄골", "어깨가 뜯", "범죄 현장"], "cleaned": ["깨끗하게 정돈", "지나치게 깔끔", "지나치게 깨끗", "감쪽같이", "깨끗해진"], "vandalized": ["어지럽혀", "뒤집힌", "넘어진 가구", "벽면에 그려진", "벽에 칠한", "붓으로 칠한", "거친 붉은 원", "벽면에 나타난", "어두운 붉은 얼룩", "붉은 발자국", "붉은 원형 표식"], "dusk_night": ["황혼", "해 질", "해질", "노을"], # "밤/어둠" 은 너무 광범위 — 제거 "morning_light": ["새벽", "동틀"], # "아침" 도 광범위 "clue_macro": ["구겨진 사진", "근접 증거", "macro_close"], # "클로즈업" 만으론 mirror_close 와 충돌 } # State-priority order (Codex BLOCKING 1) — first match wins after negation check. STATE_CLASSIFICATION_PRIORITY = [ "cleaned", # cleaned overrides everything (사용자 의도) "vandalized", # explicit vandalized/vision marks "corpse_marks", # only if negation 미적용 "dusk_night", "morning_light", "clue_macro", ] # Corpse negation patterns (BLOCKING 1 + W2b extended) # Window 안에서 (corpse/blood/body keyword 주변 80 chars) negation 표현. CORPSE_NEGATION_PATTERNS = [ re.compile(r"시신[^없]{0,20}(없|전혀)"), re.compile(r"시체[^없]{0,20}(없|전혀)"), re.compile(r"피[^없]{0,15}(없|전혀)"), re.compile(r"핏자국[^없]{0,15}(없|전혀)"), re.compile(r"폭력\s*흔적[^없]{0,15}(없|전혀)"), re.compile(r"흔적[^없]{0,15}없"), re.compile(r"흔적을\s*찾지"), re.compile(r"감쪽\s*같이\s*사라"), re.compile(r"사라진"), # 길게 enumerate 후 부정 — "사람, 얼굴, 시신, 피, ... 흔적은 전혀 없다" re.compile(r"시신.{0,30}전혀\s*없"), re.compile(r"시신이나\s*사람\s*형체\s*없"), # W2b BLOCKING 1: 실측 DB prompt 의 명시적 부정 표현 (object-before-prohibition) re.compile(r"(시신|시체|피|혈흔|유혈|신체|폭력).{0,80}절대\s*포함하지\s*않"), re.compile(r"(시신|시체|피|혈흔|유혈|신체|폭력).{0,80}절대\s*넣지\s*않"), re.compile(r"(시신|시체|피|혈흔|유혈|신체|폭력).{0,80}포함하지\s*않"), re.compile(r"(시신|시체|피|혈흔|유혈|신체|폭력).{0,80}넣지\s*않"), re.compile(r"(시신|시체|피|혈흔|유혈|신체|폭력).{0,80}등장하지\s*않"), re.compile(r"(시신|시체|피|혈흔|유혈|신체|폭력).{0,80}보이지\s*않"), # W2c IMPORTANT 1: prohibition-before-object (절대/금지/안 된다 가 앞에 오는 형태) re.compile(r"절대\s*(시신|시체|피|혈흔|유혈|신체|폭력).{0,15}넣지\s*않"), re.compile(r"절대\s*(시신|시체|피|혈흔|유혈|신체|폭력).{0,15}포함하지\s*않"), re.compile(r"(시신|시체|피|혈흔|유혈|신체|폭력).{0,15}보이면\s*안\s*된"), re.compile(r"(시신|시체|피|혈흔|유혈|신체|폭력)\s*묘사\s*금지"), re.compile(r"(시신|시체|피|혈흔|유혈|신체|폭력)\s*금지"), ] # W2b IMPORTANT 1: wrist/body context — character mark 로서 vandalized 제외. WRIST_BODY_MARK_CONTEXTS = [ re.compile(r"손목.{0,15}(붉은|원형|표식|마크|문신)"), re.compile(r"팔.{0,15}(붉은|원형|표식|마크|문신)"), re.compile(r"문신.{0,15}(붉은|원형|표식)"), re.compile(r"몸.{0,10}(붉은|원형|표식|문신)"), re.compile(r"(붉은|원형)\s*표식.{0,15}(본다|들여다본다|손목|문신|팔)"), ] # W2c IMPORTANT 2: unit_id slug sanitizer (alphanumeric + Hangul + '_'). _SLUG_SAFE_RE = re.compile(r"[^0-9A-Za-z_가-힣]+") def _slug_id(value: str) -> str: """Return a stable slug suitable for unit_id / file id. Keeps ASCII alnum + underscore + Hangul. Other chars collapse to '_'. """ if not value: return "_" slug = _SLUG_SAFE_RE.sub("_", value).strip("_") return slug or "_" def _has_background_vandalized_context(text: str) -> bool: """vandalized 후보일 때 surface/structure context (wall/floor/furniture/ explicit damage action) 가 있는지 확인. 일반 location keyword (거실/침실 등) 만 있으면 BG context 로 인정하지 않는다 — wrist mark 가 거실 안에서 발생해도 그건 character mark 이지 BG state 가 아니기 때문. """ surface_contexts = [ "벽면", "벽에", "바닥", "장판", "마룻바닥", "가구", "식탁 위", "침대 위", "선반 위", "표면", "어지럽혀", "뒤집힌", "넘어진", ] return any(c in text for c in surface_contexts) def _is_wrist_body_mark(text: str) -> bool: return any(p.search(text) for p in WRIST_BODY_MARK_CONTEXTS) # W2c BLOCKING 3: golden state oracle — runtime classifier 실측 케이스. # 각 튜플 = (case_label, input_text, expected_state). # W2d IMPORTANT 1: scenario-neutral text only. 특정 location/character/scene # 어휘 금지. sample fixture 특화 케이스는 SAMPLE_FIXTURE_L05_ORACLE_CASES 로 # 분리. STATE_ORACLE_CASES: list[tuple[str, str, str]] = [ ("negation_jeoldae_pohamhaji", ("저녁 무렵 작은 실내, 황혼빛이 창문을 통해 들어온다." " 인물, 얼굴, 시신, 피가 묻은 신체, 노골적인 유혈 장면은" " 절대 포함하지 않는다."), "dusk_night"), ("negation_jeoldae_nehji", ("실내 wide 시점. 인물, 식별 가능한 얼굴, 시신," " 노골적인 혈흔은 절대 넣지 않는다."), "normal"), ("negation_prohibition_before", "절대 시신을 넣지 않는다. 평범한 작은 실내.", "normal"), ("negation_simple_jeonhyeo_obs", ("사람, 얼굴, 시신, 피, 노골적인 폭력 흔적은 전혀 없다." " 평범한 자연광 실내."), "normal"), ("cleaned_explicit", "지나치게 깨끗하게 정돈된 방안, 시신은 보이지 않는다.", "cleaned"), ("real_corpse_unmasked", "어깨가 뜯어진 참혹한 모습으로 주저앉아 있는 시신, 흥건한 핏자국.", "corpse_marks"), ("wrist_mark_not_vandalized", "인물의 손목에 또렷한 붉은 원형 표식이 보인다. 실내 공간 안에서.", "normal"), ("wall_mark_vandalized", "벽면에 그려진 거친 붉은 원형 표식이 깊게 새겨져 있다.", "vandalized"), ] # W2d IMPORTANT 1: sample-fixture-specific oracle cases (L05 옥탑방 / 수리영 등). # 본 케이스 list 는 compute_acceptance_metrics 의 acceptance gate 에 포함되지 # 않는다 — generic STATE_ORACLE_CASES 만 gate. sample-specific 은 별도 진단용. SAMPLE_FIXTURE_L05_ORACLE_CASES: list[tuple[str, str, str]] = [ ("l05_negation_with_location", ("저녁 무렵 옥탑방 거실, 황혼빛. 인물, 얼굴, 시신, 피가 묻은 신체," " 노골적인 유혈 장면은 절대 포함하지 않는다."), "dusk_night"), ("l05_wrist_mark_with_character", "엄마의 손목에 또렷한 붉은 원형 표식이 보인다. 거실에서.", "normal"), ] def classify_state_layer(text: str) -> str: """Unified state classifier (Codex BLOCKING 1 + W2b extensions). Priority: cleaned > vandalized(bg-context, not wrist mark) > corpse(negation guard) > lighting > clue_macro > normal. """ if not text: return "normal" # 1. cleaned override (corpse keyword 와 동시 등장해도 우선). if any(k in text for k in STATE_LAYER_KEYWORDS["cleaned"]): return "cleaned" # 2. vandalized vision — W2b IMPORTANT 1: wrist/body context 제외. if any(k in text for k in STATE_LAYER_KEYWORDS["vandalized"]): if _is_wrist_body_mark(text) and not _has_background_vandalized_context(text): # wrist/body mark only → vandalized 아님. fall through. pass else: return "vandalized" # 3. corpse — negation guard. if any(k in text for k in STATE_LAYER_KEYWORDS["corpse_marks"]): negated = any(p.search(text) for p in CORPSE_NEGATION_PATTERNS) if not negated: return "corpse_marks" # negated — fall through to lighting/normal. # 4. lighting layers. if any(k in text for k in STATE_LAYER_KEYWORDS["dusk_night"]): return "dusk_night" if any(k in text for k in STATE_LAYER_KEYWORDS["morning_light"]): return "morning_light" # 5. clue macro (single-purpose). if any(k in text for k in STATE_LAYER_KEYWORDS["clue_macro"]): return "clue_macro" return "normal" # W2e BLOCKING 1: primary resolver patterns are now built dynamically from the # spec's keyword whitelist. No sample vocabulary is embedded at module level. # # Generic framing templates — '{kw}' placeholder is filled by re.escape(keyword) # at build time. Order matters: longer/more-specific templates first. PRIMARY_FRAMING_TEMPLATES: list[str] = [ r"카메라는[^.]*?({kw})", r"중앙의?\s*({kw})", r"안쪽의?\s*({kw})", r"({kw})\s*안쪽", r"({kw})\s*내부", r"({kw})을?\s*촬영한", r"({kw})을?\s*바라본다", ] def _build_primary_space_patterns(spec: "SampleFixtureSpec" ) -> list[re.Pattern]: """Build primary-space regex patterns from the spec's keyword whitelist. Sort by descending keyword length so longer phrases match first. Escape Korean keywords and allow optional internal whitespace by replacing literal spaces with `\\s*`. """ keywords = sorted( {kw for kw, _node in spec.space_keyword_whitelist}, key=len, reverse=True, ) if not keywords: return [] escaped = [ re.escape(kw).replace(r"\ ", r"\s*") for kw in keywords ] alt = "(?:" + "|".join(escaped) + ")" patterns: list[re.Pattern] = [] for tmpl in PRIMARY_FRAMING_TEMPLATES: patterns.append(re.compile(tmpl.format(kw=alt))) return patterns def _normalize_primary_node_from_spec(raw_keyword: str, spec: "SampleFixtureSpec" ) -> Optional[str]: """Resolve a matched raw keyword back to a node_id using the spec's whitelist. Whitespace-insensitive comparison. """ cleaned = raw_keyword.replace(" ", "") for kw, node in spec.space_keyword_whitelist: if kw.replace(" ", "") == cleaned: return node return None def extract_primary_and_secondary_spaces( text: str, spec: Optional["SampleFixtureSpec"] = None, ) -> tuple[Optional[str], list[str]]: """W2d BLOCKING 1: spec-driven (location-agnostic) primary/secondary 분리. primary = camera-direction first match (zone/boundary 는 부모 room 흡수). secondary = primary 외 매칭된 모든 노드. spec=None 이면 SampleFixtureSpec global 미사용 → empty 결과 (generic 호출자가 반드시 spec 을 넘겨야 한다는 contract). legacy 호출자 (decompose_l05_variant) 는 build_sample_fixture_l05_spec() 자동 주입. """ if not text: return None, [] if spec is None: return None, [] keyword_whitelist = spec.space_keyword_whitelist room_node_ids = spec.room_node_ids zone_boundary_node_ids = spec.zone_boundary_node_ids # zone/boundary → parent room 매핑 (spec.space_nodes.contained_in 사용). zone_to_parent: dict[str, str] = {} for seed in spec.space_nodes: if seed.node_id in zone_boundary_node_ids and seed.contained_in: zone_to_parent[seed.node_id] = seed.contained_in # W2e BLOCKING 1: patterns built dynamically from spec keywords. primary_patterns = _build_primary_space_patterns(spec) raw_primary: Optional[str] = None for pat in primary_patterns: m = pat.search(text) if m: raw_primary = m.group(1) break primary: Optional[str] = None if raw_primary: primary = _normalize_primary_node_from_spec(raw_primary, spec) # zone/boundary 면 부모 room 흡수. if primary in zone_boundary_node_ids and primary in zone_to_parent: primary = zone_to_parent[primary] # all matched nodes (room + zone + boundary), order preserved. matched: list[str] = [] for kw, node in keyword_whitelist: if kw in text and node not in matched: matched.append(node) # secondary = matched 에서 primary 와 zone/boundary 의 부모 (primary 자체) 제외. secondary: list[str] = [ n for n in matched if n != primary and not ( n in zone_boundary_node_ids and zone_to_parent.get(n) == primary ) ] # fallback: primary 추출 실패 시 첫 매칭 room 으로. if primary is None and matched: for n in matched: if n in room_node_ids: primary = n secondary = [m for m in matched if m != n] break else: # 모든 매칭이 zone/boundary 면 부모 room 으로 흡수. if matched: first_parent = zone_to_parent.get(matched[0]) if first_parent: primary = first_parent secondary = [ m for m in matched if not (m in zone_boundary_node_ids and zone_to_parent.get(m) == first_parent) ] return primary, secondary # Camera view family detection CAMERA_HEIGHT_PATTERNS = [ (re.compile(r"약\s*0\.5\s*m|0\.5미터|바닥\s*가까운|낮은\s*위치"), "floor_close"), (re.compile(r"약\s*1\.3\s*m|1\.3m|앉은\s*눈높이.*선\s*눈높이"), "low_eye_level"), (re.compile(r"약\s*1\.6\s*m|1\.6미터|사람\s*눈높이|standing\s*eye-level"), "standing_eye_level"), ] CAMERA_FRAMING_PATTERNS = [ (re.compile(r"문턱|doorway"), "doorway_wide"), (re.compile(r"식탁\s*가까|식탁\s*위|table\s*close"), "eye_level_table_close"), (re.compile(r"거울\s*정면|mirror\s*close|얼굴을\s*씻"), "mirror_close"), (re.compile(r"클로즈업|close-up|매크로|macro"), "macro_close"), (re.compile(r"광각|wide|전체|대각선"), "eye_level_wide"), ] # ----------------------------------------------------------------------------- # Dataclasses (plan.md §1) # ----------------------------------------------------------------------------- @dataclass class PlaceGroup: place_group_id: str label: str member_location_ids: list[str] physical_place_key: str continuity_scope: str evidence: list[str] = field(default_factory=list) @dataclass class SetGroup: set_group_id: str parent_place_group_id: str label: str continuity_policy: str member_space_nodes: list[str] = field(default_factory=list) structural_versions: list[str] = field(default_factory=list) @dataclass class SpaceNode: node_id: str label: str node_type: str # room|zone|boundary|exterior set_group_id: str contained_in: Optional[str] connected_to: list[str] visibility_to: list[str] active_status: str # active|needs_decision|inactive evidence: list[str] = field(default_factory=list) @dataclass class StateLayer: layer_id: str label: str affects: list[str] geometry_preserving: bool severity: str applied_shots: list[str] = field(default_factory=list) evidence: list[str] = field(default_factory=list) @dataclass class StructuralVersion: version_id: str reason: str geometry_changes: list[str] = field(default_factory=list) starts_at_scene: Optional[int] = None evidence: list[str] = field(default_factory=list) @dataclass class ShotBindingCandidate: shot_id: str loc_id: str set_group_id: str space_node_candidates: list[str] chosen_policy_candidate: str state_layers: list[str] structural_version: str camera_view_family_candidate: Optional[str] unresolved_reason: Optional[str] = None @dataclass class MasterUnit: unit_id: str space_node: str camera_view: str state_layer: str = "normal" @dataclass class DerivedUnit: unit_id: str based_on_master: str state_layer: Optional[str] = None camera_view: Optional[str] = None @dataclass class BackgroundGenerationUnitPlan: set_group_id: str structural_version: str master_units: list[MasterUnit] = field(default_factory=list) derived_state_units: list[DerivedUnit] = field(default_factory=list) derived_camera_units: list[DerivedUnit] = field(default_factory=list) rejected_cross_product: list[dict] = field(default_factory=list) @dataclass class ChainBgVariantDecomposition: """Generic-named chain_bg variant decomposition (W2e IMPORTANT 2).""" variant_id: str space_nodes: list[str] state_layer: str camera_view: str primary_space: Optional[str] = None # Codex BLOCKING 2 visible_secondary_nodes: list[str] = field(default_factory=list) multi_space: bool = False # legacy — see has_secondary_spaces has_secondary_spaces: bool = False # W2b IMPORTANT 3 rename reject_reason: Optional[str] = None violates_continuity: bool = False # legacy alias — old call sites/tests still use L05BDecomposition. L05BDecomposition = ChainBgVariantDecomposition # ---------------------------------------------------------------------------- # W2c BLOCKING 1: sample fixture vs generic engine 구조 분리. # SampleFixtureSpec = fixture data 한 묶음. generic builder 는 spec input. # ---------------------------------------------------------------------------- @dataclass class PlaceGroupSeed: place_group_id: str label: str member_location_ids: list[str] physical_place_key: str continuity_scope: str evidence: list[str] = field(default_factory=list) @dataclass class SetGroupSeed: set_group_id: str parent_place_group_id: str label: str continuity_policy: str member_space_nodes: list[str] = field(default_factory=list) structural_versions: list[str] = field(default_factory=list) @dataclass class SpaceNodeSeed: node_id: str label: str node_type: str set_group_id: str contained_in: Optional[str] connected_to: list[str] visibility_to: list[str] active_status: str evidence: list[str] = field(default_factory=list) @dataclass class SampleFixtureSpec: """All sample-specific data for a single fixture (e.g. L05 rooftop).""" fixture_id: str project_id: str episode_id: str canon_id: str location_short_id: str source_run_path: Path place_groups: list[PlaceGroupSeed] set_groups: list[SetGroupSeed] space_nodes: list[SpaceNodeSeed] space_keyword_whitelist: list[tuple[str, str]] zone_boundary_node_ids: set[str] room_node_ids: set[str] # W2d BLOCKING 2: weak character hints (character_name → candidate room ids). weak_room_hints: dict[str, list[str]] = field(default_factory=dict) # W2d BLOCKING 3: default camera per space_node for master_unit seeding. default_camera_by_space_node: dict[str, str] = field(default_factory=dict) # W2e IMPORTANT 3: bible filename inside source_run_path. source_bible_filename: str = "gemini_rooftop_bible.json" # ----------------------------------------------------------------------------- # PlaceGroup / SetGroup / SpaceNode builders # ----------------------------------------------------------------------------- # ---------------------------------------------------------------------------- # Sample fixture loader (L05 rooftop interior) — SAMPLE-SPECIFIC. # 다른 location 추가 시 build_sample_fixture__spec() 새로 정의. # ---------------------------------------------------------------------------- def build_sample_fixture_l05_spec() -> SampleFixtureSpec: """Construct the L05 rooftop interior sample fixture spec. This function is sample-specific — all hard-coded L05 labels, node IDs, keywords live here. Generic engine (build_*_from_spec) reads the spec and derives PlaceGroup / SetGroup / SpaceNode via location-agnostic rules. """ sg_id = "sg_rooftop_interior" pg_id = "pg_rooftop_villa" return SampleFixtureSpec( fixture_id="l05_rooftop_interior", project_id=SAMPLE_FIXTURE_PROJECT_ID, episode_id=SAMPLE_FIXTURE_EPISODE_ID, canon_id=SAMPLE_FIXTURE_CANON_ID, location_short_id=SAMPLE_FIXTURE_SHORT_ID, source_run_path=DEFAULT_SOURCE_RUN, place_groups=[ PlaceGroupSeed( place_group_id=pg_id, label="옥탑 빌라 (수리영/민숙 거주지)", member_location_ids=["L04", "L05"], physical_place_key="rooftop_villa@incheon_pyeongdong", continuity_scope="single_building", evidence=[ "L04 (옥상 외부) + L05 (옥탑방 내부) 같은 빌라 옥상", "bible.condition_age 인천 변두리 오래된 빌라 옥탑방", ], ), ], set_groups=[ SetGroupSeed( set_group_id=sg_id, parent_place_group_id=pg_id, label="옥탑방 내부 세트", continuity_policy="geometry_locked", member_space_nodes=[ "거실", "주방코너", "현관", "수리영의_방", "민숙의_방_안방", "욕실", ], structural_versions=["sv_initial"], ), ], space_nodes=[ SpaceNodeSeed( node_id="거실", label="거실 (living kitchen)", node_type="room", set_group_id=sg_id, contained_in=None, connected_to=["주방코너", "현관", "수리영의_방", "민숙의_방_안방", "욕실"], visibility_to=["주방코너", "현관"], active_status="active", evidence=["bible.sub_spaces.거실"], ), SpaceNodeSeed( node_id="주방코너", label="주방 영역 (sink + stove)", node_type="zone", set_group_id=sg_id, contained_in="거실", connected_to=["거실"], visibility_to=["거실"], active_status="active", evidence=["bible.sub_spaces.주방 영역 — 거실과 인접/통합"], ), SpaceNodeSeed( node_id="현관", label="현관 (entry)", node_type="boundary", set_group_id=sg_id, contained_in="거실", connected_to=["거실"], visibility_to=["거실"], active_status="active", evidence=["bible.doors_windows 현관 철문"], ), SpaceNodeSeed( node_id="수리영의_방", label="수리영의 방", node_type="room", set_group_id=sg_id, contained_in=None, connected_to=["거실"], visibility_to=[], active_status="active", evidence=["bible.sub_spaces.수리영의 방"], ), SpaceNodeSeed( node_id="민숙의_방_안방", label="민숙의 방 (안방)", node_type="room", set_group_id=sg_id, contained_in=None, connected_to=["거실"], visibility_to=[], active_status="needs_decision", evidence=["bible.sub_spaces.민숙의 방 — 14 shots 안에서 직접 evidence 약함"], ), SpaceNodeSeed( node_id="욕실", label="욕실", node_type="room", set_group_id=sg_id, contained_in=None, connected_to=["거실"], visibility_to=[], active_status="active", evidence=["bible.sub_spaces.욕실"], ), ], space_keyword_whitelist=list(SAMPLE_FIXTURE_L05_SPACE_KEYWORDS), zone_boundary_node_ids=set(SAMPLE_FIXTURE_L05_ZONE_BOUNDARY_NODES), room_node_ids=set(SAMPLE_FIXTURE_L05_ROOM_NODES), weak_room_hints={ "수리영": ["수리영의_방"], "엄마": ["민숙의_방_안방"], "민숙": ["민숙의_방_안방"], "침대": ["민숙의_방_안방"], # weak only with character pair }, default_camera_by_space_node={ "거실": "eye_level_wide@standing_eye_level", "수리영의_방": "doorway_wide@standing_eye_level", "민숙의_방_안방": "doorway_wide@standing_eye_level", "욕실": "mirror_close@standing_eye_level", }, source_bible_filename="gemini_rooftop_bible.json", ) # ---------------------------------------------------------------------------- # Generic engine — location-agnostic. Consumes SampleFixtureSpec. # Bible param 은 향후 spec field 로 결합 가능 (현재는 단순 통과). # ---------------------------------------------------------------------------- def build_place_groups_from_spec(spec: SampleFixtureSpec) -> list[PlaceGroup]: return [PlaceGroup(**asdict(seed)) for seed in spec.place_groups] def build_set_groups_from_spec(spec: SampleFixtureSpec, place_groups: list[PlaceGroup] ) -> list[SetGroup]: valid_pg_ids = {pg.place_group_id for pg in place_groups} out: list[SetGroup] = [] for seed in spec.set_groups: if seed.parent_place_group_id not in valid_pg_ids: continue out.append(SetGroup(**asdict(seed))) return out def build_space_nodes_from_spec(spec: SampleFixtureSpec, set_groups: list[SetGroup] ) -> list[SpaceNode]: valid_sg_ids = {sg.set_group_id for sg in set_groups} out: list[SpaceNode] = [] for seed in spec.space_nodes: if seed.set_group_id not in valid_sg_ids: continue out.append(SpaceNode(**asdict(seed))) return out # ---------------------------------------------------------------------------- # Legacy thin wrappers — delegate to spec-driven engine using sample fixture. # 신규 코드는 *_from_spec 사용. 기존 호출자 (tests + main) 호환 유지용. # ---------------------------------------------------------------------------- def build_place_groups(bible: dict) -> list[PlaceGroup]: return build_place_groups_from_spec(build_sample_fixture_l05_spec()) def build_set_groups(bible: dict, place_groups: list[PlaceGroup]) -> list[SetGroup]: return build_set_groups_from_spec( build_sample_fixture_l05_spec(), place_groups, ) def build_space_nodes(bible: dict, set_groups: list[SetGroup]) -> list[SpaceNode]: return build_space_nodes_from_spec( build_sample_fixture_l05_spec(), set_groups, ) # ----------------------------------------------------------------------------- # StateLayer / StructuralVersion builders # ----------------------------------------------------------------------------- def build_state_layers(bible: dict, shots: list[ShotMeta]) -> list[StateLayer]: """Deterministic StateLayer set. shots is optional (for applied_shots field). state_variants bible 에서 강하게 도출되는 4가지 (corpse/cleaned/vandalized/normal) + lighting layer 2개 (dusk_night / morning_light) + 1 weather/oblique (empty_vision). """ layers = [ StateLayer( layer_id="sl_normal", label="normal", affects=["lighting"], geometry_preserving=True, severity="none", evidence=["bible default state"], ), StateLayer( layer_id="sl_corpse_marks", label="corpse_marks", affects=["blood", "damage", "surface"], geometry_preserving=True, severity="high", evidence=["bible.state_variants 참혹한 범죄 현장"], ), StateLayer( layer_id="sl_cleaned", label="cleaned", affects=["surface", "furniture"], geometry_preserving=True, severity="low", evidence=["bible.state_variants 감쪽같이 청소된 상태"], ), StateLayer( layer_id="sl_vandalized", label="vandalized", affects=["surface", "furniture", "damage"], geometry_preserving=True, severity="medium", evidence=["bible.state_variants 침입자 난장판"], ), StateLayer( layer_id="sl_dusk_night", label="dusk_night", affects=["lighting", "weather"], geometry_preserving=True, severity="low", ), StateLayer( layer_id="sl_morning_light", label="morning_light", affects=["lighting"], geometry_preserving=True, severity="low", ), StateLayer( layer_id="sl_empty_vision", label="empty_vision", affects=["emptiness", "lighting"], geometry_preserving=True, severity="low", evidence=["빈 집 + 환각 vision marks"], ), ] # applied_shots — W2b BLOCKING 3: classify_state_layer single winner 사용. # raw keyword scan 대신 unified classifier 결과로 채워서 shot_binding 과 정합. if shots: winners_by_shot: dict[str, str] = {} for s in shots: text = f"{s.shot_description}\n{s.scene_summary}" winners_by_shot[s.label] = classify_state_layer(text) for layer in layers: layer.applied_shots = sorted( shot_id for shot_id, winner in winners_by_shot.items() if winner == layer.label ) return layers def build_structural_versions(bible: dict, shots: list[ShotMeta]) -> list[StructuralVersion]: """Default = single sv_initial. Geometry-change keyword 가 발견되면 추가 version.""" versions = [StructuralVersion( version_id="sv_initial", reason="initial state — no structural change detected", geometry_changes=[], starts_at_scene=None, evidence=[], )] haystack_chunks: list[str] = [] for unk in bible.get("unknowns", []) or []: haystack_chunks.append(str(unk)) for sv in bible.get("state_variants", []) or []: haystack_chunks.append(json.dumps(sv, ensure_ascii=False)) for s in shots or []: haystack_chunks.append(f"{s.shot_description}\n{s.scene_summary}") haystack = "\n".join(haystack_chunks) triggered: list[str] = [ kw for kw in GEOMETRY_CHANGE_KEYWORDS if kw in haystack ] if triggered: versions.append(StructuralVersion( version_id="sv_remodel_pending", reason="geometry-change keyword detected — manual review required", geometry_changes=triggered, starts_at_scene=None, evidence=triggered, )) return versions # ----------------------------------------------------------------------------- # L05B variant decomposer (plan.md §3) # ----------------------------------------------------------------------------- def decompose_chain_bg_variant(variant_id: str, prompt_text: str, spec: Optional["SampleFixtureSpec"] = None, ) -> L05BDecomposition: """Decompose a chain_bg variant prompt into 3 axes + primary/secondary spaces. W2d BLOCKING 1: spec-driven generic decomposer. fixture-specific data 는 spec 으로만 들어온다. spec=None 이면 caller 가 명시적으로 spec 을 안 제공한 것으로 처리 (decompose_l05_variant 같은 legacy wrapper 가 자동 주입). Deterministic substring + regex matching. No LLM, no fuzzy/synonym. """ text = prompt_text or "" if spec is None: # caller error; return empty placeholder decomposition rather than guess. return L05BDecomposition( variant_id=variant_id, space_nodes=["unknown"], state_layer="normal", camera_view="unknown", primary_space=None, visible_secondary_nodes=[], multi_space=False, has_secondary_spaces=False, reject_reason="no spec provided", violates_continuity=False, ) # 1. space_nodes — primary + secondary (BLOCKING 2) primary_space, secondary_nodes = extract_primary_and_secondary_spaces( text, spec=spec, ) # legacy field `space_nodes` 유지: primary + secondary 합집합 (dedup). matched_spaces: list[str] = [] if primary_space: matched_spaces.append(primary_space) for n in secondary_nodes: if n not in matched_spaces: matched_spaces.append(n) if not matched_spaces: matched_spaces = ["unknown"] multi_space = len(matched_spaces) >= 2 # 2. state_layer — unified classifier (BLOCKING 1) state_layer = classify_state_layer(text) # 3. camera_view — height + framing height_tag: Optional[str] = None for pat, tag in CAMERA_HEIGHT_PATTERNS: if pat.search(text): height_tag = tag break framing_tag: Optional[str] = None for pat, tag in CAMERA_FRAMING_PATTERNS: if pat.search(text): framing_tag = tag break if framing_tag and height_tag: camera_view = f"{framing_tag}@{height_tag}" elif framing_tag: camera_view = framing_tag elif height_tag: camera_view = height_tag else: camera_view = "unknown" # 4. continuity violation — primary 가 2개 이상 room 일 때만 (BLOCKING 2). # secondary mention 은 visibility constraint 로만 보관, reject 아님. violates_continuity = False reject_reason: Optional[str] = None # primary 가 둘 이상 room 으로 잡힌 경우는 현재 추출기 디자인상 0건이지만, # 안전망: primary + secondary 가 모두 room (zone/boundary 아님) 인 경우만 # 이중 master 의도로 간주. secondary_rooms = [n for n in secondary_nodes if n in spec.room_node_ids] primary_is_room = primary_space in spec.room_node_ids # primary 자체 0 또는 unknown 인데 다중 room 매칭이면 위반 후보. if (not primary_is_room) and len(secondary_rooms) >= 2: violates_continuity = True reject_reason = ( f"primary_space 결정 실패 + 다중 room secondary " f"({secondary_rooms}) — 위반 후보, manual review" ) return L05BDecomposition( variant_id=variant_id, space_nodes=matched_spaces, state_layer=state_layer, camera_view=camera_view, primary_space=primary_space, visible_secondary_nodes=secondary_nodes, multi_space=multi_space, # legacy has_secondary_spaces=len(secondary_nodes) > 0, reject_reason=reject_reason, violates_continuity=violates_continuity, ) # legacy alias — old tests still call decompose_l05_variant. 신규 코드는 generic # decompose_chain_bg_variant(spec=...) 사용. def decompose_l05_variant(variant_id: str, prompt_text: str) -> L05BDecomposition: """Legacy alias for the L05 sample fixture. Use the generic `decompose_chain_bg_variant(spec=...)` for new code.""" return decompose_chain_bg_variant( variant_id, prompt_text, spec=build_sample_fixture_l05_spec(), ) # ----------------------------------------------------------------------------- # Continuity grouping helper (Codex IMPORTANT 3: geometry_preserving guard) # ----------------------------------------------------------------------------- # StateLayer registry — geometry_preserving lookup. build_state_layers 가 # 채워주는 것보다, 모든 state 가 preserving=True default 라 inline 등록. _STATE_LAYER_PRESERVING: dict[str, bool] = { "sl_normal": True, "sl_corpse_marks": True, "sl_cleaned": True, "sl_vandalized": True, "sl_dusk_night": True, "sl_morning_light": True, "sl_empty_vision": True, } def continuity_group_for(space_node_id: str, structural_version_id: str, state_layer_id: Optional[str] = None) -> str: """Continuity group key = (space_node, structural_version). state_layer 는 무시 (geometry preserving 가정). 다른 state 라도 같은 geometry 이면 같은 group. Codex IMPORTANT 3 guard: state_layer_id 가 주어졌고 registry 에 등록된 geometry_preserving=False 면 assertion (StructuralVersion 승격 필요). """ if state_layer_id is not None: preserving = _STATE_LAYER_PRESERVING.get(state_layer_id, True) if not preserving: raise AssertionError( f"state_layer={state_layer_id} is geometry_preserving=False — " "must be promoted to StructuralVersion, not continuity group" ) return f"cg::{structural_version_id}::{space_node_id}" # ----------------------------------------------------------------------------- # ShotBindingCandidate builder (rollup) # ----------------------------------------------------------------------------- ROLLUP_DECISIONS = { "bedroom_unresolved": ( "수리영의_방 vs 민숙의_방_안방 분류 — 14 shots 의 bedroom 구분이" " keyword 만으로 결정 안 되는 그룹. 사용자 결정 또는 plan_v3 룰 확장." ), "manual_review_needed": ( "shot description framing keyword 부족 — 카메라 anchor 사용자 결정." ), "minsook_active_decision": ( "민숙의_방_안방 SpaceNode 를 active 로 승격할지 사용자 결정 필요." ), "multi_space_wide_policy": ( "L05B05 같은 두 공간 한 wide 프레임 variant 허용/금지 정책 결정." ), "floor_close_clue_category": ( "L05B02 같은 0.5m floor close (증거품 macro) 를 derived_camera_unit" " 으로 인정할지 별도 detail_clue 카테고리로 분리할지." ), } def _detect_state_layers_for_shot(shot: ShotMeta) -> list[str]: """Codex BLOCKING 1: unified classifier — variant + shot 공통. classify_state_layer 가 단일 winner 반환. sl_