"""OutdoorShotGroundingStep — 샷 grounding (W22 직행 체인 ③, order 21.79). 캐논 자산이 있는 그룹의 야외 선택 샷마다 존/앵커/카메라 접지를 산출한다. data.groups[] = { "map_asset_id": str, "shots": {"_": {"status": "ok", "ground": {...}, "attempts": n} | {"status": "failed", "error": str}}, } 캐논 결측 그룹 = {"skipped": str} (직행 비활성 — 기존 라인 fallback, 비실패) 샷 매핑 규칙은 outdoor_direct_common 이 SOT (W1 과 동일 계약). 설계: docs/w22-outdoor-canon-direct-compose-design-20260709/design.md """ import json import logging from pathlib import Path from typing import Any, Dict, Optional from app.core.step_runner import StepRunner logger = logging.getLogger(__name__) SCHEMA_VERSION = 1 PROMPT_VERSION = "1" class OutdoorShotGroundingStep(StepRunner): """샷 grounding (W22 ③) — 맵+범례+샷+씬 전문 → 존/앵커/카메라.""" def _config_hash(self) -> str: import hashlib import json as _json from app.core.config import settings payload = { "background_mode": settings.background_mode, "model": settings.openai_model, "schema_version": SCHEMA_VERSION, "prompt_version": PROMPT_VERSION, "outdoor_direct_compose_enabled": True, } return hashlib.sha256( _json.dumps(payload, sort_keys=True).encode("utf-8") ).hexdigest()[:16] def _load_prev_checkpoint(self, step_id: str) -> Optional[Dict[str, Any]]: from app.core.config import settings cp = ( Path(settings.projects_dir) / self.project_id / "checkpoints" / "episodes" / self.episode_id / step_id / "manifest.json" ) if cp.exists(): try: return json.loads(cp.read_text(encoding="utf-8")) except Exception as exc: # noqa: BLE001 logger.warning( "outdoor_shot_grounding: %s 로드 실패: %s", step_id, exc ) return None def _execute(self, mode: str = "resume") -> Dict[str, Any]: from app.core.config import settings def _empty() -> Dict[str, Any]: return { "applicable_count": 0, "completed_count": 0, "failed_count": 0, "schema_version": SCHEMA_VERSION, "config_hash": self._config_hash(), "data": {"groups": {}}, } if settings.background_mode not in {"on", "floor_plan_anchored"}: return _empty() if not getattr(settings, "outdoor_direct_compose_enabled", False): return _empty() spec_cp = self._load_prev_checkpoint("outdoor_place_spec") spec_groups = (spec_cp or {}).get("data", {}).get("groups", {}) or {} canon_cp = self._load_prev_checkpoint("outdoor_place_canon") canon_groups = (canon_cp or {}).get("data", {}).get("groups", {}) or {} target: Dict[str, Dict[str, Any]] = {} skipped: Dict[str, Any] = {} for gid, entry in spec_groups.items(): if not isinstance(entry, dict) or not entry.get("spec"): continue canon = canon_groups.get(gid) or {} map_path = canon.get("map_png_path") or "" if canon.get("status") != "ok" or not map_path: # 캐논 결측 = 이 그룹 직행 비활성 (기존 라인 fallback) — 비실패 skipped[gid] = {"skipped": "canon assets missing"} continue if not Path(map_path).exists(): skipped[gid] = {"skipped": f"map png not found: {map_path}"} continue target[gid] = { "spec": entry["spec"], "outdoor_loc_ids": entry.get("outdoor_loc_ids") or [], "scene_indices": entry.get("scene_indices") or [], "map_png_path": map_path, "map_asset_id": canon.get("map_asset_id") or "", } if not target and not skipped: logger.info("outdoor_shot_grounding: 대상 그룹 없음 — no-op") return _empty() # ── 공통 입력 로드 ──────────────────────────────────────── scene_save_cp = self._load_prev_checkpoint("scene_save") scene_texts_all: Dict[int, str] = {} for seg in (scene_save_cp or {}).get("data", {}).get("segments", []) or []: si = seg.get("scene_index") if isinstance(si, int): # 씬 원문 전문 — 절대 자르지 않는다 scene_texts_all[si] = seg.get("text") or "" director_cp = self._load_prev_checkpoint("scene_director") scene_primary: Dict[int, str] = {} for sc in (director_cp or {}).get("data", {}).get("scenes", []) or []: si = sc.get("scene_index") primary = sc.get("primary_location", "") or "" if si is not None and primary: scene_primary[int(si)] = primary from app.modules.pipeline.outdoor_direct_common import ( build_selected_keys, build_shot_loc_map, filter_group_shots, ) selected_keys = build_selected_keys( self._load_prev_checkpoint("shot_selection") ) shot_loc_by_key = build_shot_loc_map( self._load_prev_checkpoint("shot_validator") ) staging_cp = self._load_prev_checkpoint("shot_staging") staging_shots = (staging_cp or {}).get("data", {}).get("shots", []) or [] from app.core.creator_corrections import project_corrections_block corrections_block = project_corrections_block(self.project_id) from app.modules.pipeline.outdoor_shot_grounding import ( run_outdoor_shot_grounding_shot, ) # ── 그룹→샷 grounding (샷 단위 실패 격리) ───────────────── results: Dict[str, Any] = dict(skipped) applicable = 0 completed = 0 failed = 0 for gid in sorted(target): info = target[gid] loc_ids = set(info["outdoor_loc_ids"]) group_shots = filter_group_shots( staging_shots, scene_indices=info["scene_indices"], loc_ids=loc_ids, scene_primary=scene_primary, shot_loc_by_key=shot_loc_by_key, selected_keys=selected_keys, ) try: map_png = Path(info["map_png_path"]).read_bytes() except OSError as exc: # 파일이 사이에 삭제/권한 오류 — 그룹 격리 (Codex MINOR) logger.warning( "outdoor_shot_grounding: group=%s 맵 read 실패: %s", gid, exc ) results[gid] = {"skipped": f"map read failed: {exc}"} continue shots_out: Dict[str, Any] = {} for sh in group_shots: si = sh.get("scene_index") shi = sh.get("shot_index") key = f"{si}_{shi}" applicable += 1 try: out = run_outdoor_shot_grounding_shot( spec=info["spec"], shot=sh, scene_text=scene_texts_all.get(int(si), ""), map_png=map_png, creator_corrections_block=corrections_block, prompt_version=PROMPT_VERSION, project_config=self.project_config, opik_metadata=self.build_opik_metadata(), ) shots_out[key] = { "status": "ok", "ground": out["ground"], "attempts": out["attempts"], } completed += 1 except Exception as exc: # noqa: BLE001 logger.exception( "outdoor_shot_grounding: %s (group=%s) 실패", key, gid ) shots_out[key] = {"status": "failed", "error": str(exc)} failed += 1 results[gid] = { "map_asset_id": info["map_asset_id"], "shots": shots_out, } return { "applicable_count": applicable, "completed_count": completed, "failed_count": failed, "schema_version": SCHEMA_VERSION, "config_hash": self._config_hash(), "data": {"groups": results}, }