#!/usr/bin/env python3 # -*- coding: utf-8 -*- """Shared preflight helpers for G4.5a canary scripts (Wave 3). All 4 metric scripts (`g4_5a_camera_frame_consistency.py`, `g4_5a_fg_bg_shared_anchor.py`, `g4_5a_primary_framing.py`, `g4_5a_view_mixing_extension.py`) import the helpers from this module and apply them identically. The token-count script (`g4_5a_token_count.py`) does NOT use these helpers — per PR4-B4 it reads prompt files, not CP, and must remain CP-free (no `compute_render_strategy_snapshot_hash` import). Trap #1 silent-absorb ban (G4.3 R3R4-B2 + G4.4 carry): - `or [] / or {}` 패턴 금지. - `data.scenes` / `t2i_variations` / `t2i_prompt` / `render_prompt_card` / `render_strategy.spatial_consistency` 누락은 measurement_failures 로 적재 후 caller 가 exit 1. - `compute_render_strategy_snapshot_hash()` 호출 실패도 measurement_failures 로 적재 (silent skip 금지). Trap #3 — CP shape `scene["render_prompt_card"]["render_strategy"]` (NOT flat). G4.2 carry — `detail_steps.py` authoritative. Trap #8 — module-level constants single source. Imports happen in `render_prompt_card_imports()` only — no inline duplication of the 9-entry close-framing keyword tuple, 7-entry interaction verb tuple, etc. PR4-B4 — token script CP-free: this module is NOT imported by `g4_5a_token_count.py`. """ from __future__ import annotations import json import sys from pathlib import Path from typing import Any # Ensure backend/ is on sys.path so canary scripts can import the production # render_prompt_card module without altering PYTHONPATH at the shell level. _BACKEND_PATH = Path(__file__).resolve().parents[2] / "backend" if str(_BACKEND_PATH) not in sys.path: sys.path.insert(0, str(_BACKEND_PATH)) def render_prompt_card_imports() -> dict: """Return the G4.5a + G4.3-cross module symbols Wave 1-A exposes. Centralised to give every canary script a single import boundary; if Wave 1-A renames a constant, only this module needs adjustment. Trap #8 carry — module-level constants direct import; NO inline tuple duplication of the 9-entry close-framing keywords / 7-entry interaction verbs / 8-entry frame-edge tokens / etc. RO-11 binding — G4.5a reuses G4.3 `_ID_BODY_PART_TRIGGERS` directly. NO `_SPATIAL_BODY_PART_TRIGGERS` alias. """ from app.core.steps.render_prompt_card import ( # noqa: PLC0415 _ID_BODY_PART_TRIGGERS, _SPATIAL_CAMERA_HIGH_TOKENS, _SPATIAL_CAMERA_HIP_TOKENS, _SPATIAL_CAMERA_LOW_TOKENS, _SPATIAL_FG_BG_SEPARATION_TOKENS, _SPATIAL_FRAME_EDGE_POSITION_TOKENS, _SPATIAL_FRAMING_CLOSE_KEYWORDS, _SPATIAL_INTERACTION_VERBS, _SPATIAL_SHARED_ANCHOR_KEYWORDS, compute_render_strategy_snapshot_hash, ) return { "_ID_BODY_PART_TRIGGERS": _ID_BODY_PART_TRIGGERS, "_SPATIAL_CAMERA_HIGH_TOKENS": _SPATIAL_CAMERA_HIGH_TOKENS, "_SPATIAL_CAMERA_HIP_TOKENS": _SPATIAL_CAMERA_HIP_TOKENS, "_SPATIAL_CAMERA_LOW_TOKENS": _SPATIAL_CAMERA_LOW_TOKENS, "_SPATIAL_FG_BG_SEPARATION_TOKENS": _SPATIAL_FG_BG_SEPARATION_TOKENS, "_SPATIAL_FRAME_EDGE_POSITION_TOKENS": ( _SPATIAL_FRAME_EDGE_POSITION_TOKENS ), "_SPATIAL_FRAMING_CLOSE_KEYWORDS": _SPATIAL_FRAMING_CLOSE_KEYWORDS, "_SPATIAL_INTERACTION_VERBS": _SPATIAL_INTERACTION_VERBS, "_SPATIAL_SHARED_ANCHOR_KEYWORDS": _SPATIAL_SHARED_ANCHOR_KEYWORDS, "compute_render_strategy_snapshot_hash": ( compute_render_strategy_snapshot_hash ), } def _load_scenes_or_fail(cp: dict, measurement_failures: list) -> list | None: """Validate CP envelope shape and return scenes list (or None on failure). Caller must check None and fail-fast (sys.exit(1)) — measurement_failures is appended in-place. Trap #1 — silent-absorb ban (NO `.get(..., default)` fallback). """ if "data" not in cp: measurement_failures.append("cp.data key missing") return None data = cp["data"] if not isinstance(data, dict): measurement_failures.append("cp.data not dict") return None if "scenes" not in data: measurement_failures.append("cp.data.scenes key missing") return None scenes = data["scenes"] if not isinstance(scenes, list): measurement_failures.append("cp.data.scenes not list") return None return scenes def _validate_scene_preflight( scene: dict, si: int | None, shi: int | None, measurement_failures: list, ) -> list | None: """Validate per-scene shape — render_prompt_card / render_strategy / spatial_consistency / variations. Returns variations list when valid; None otherwise (caller continues loop). Per spec §5.3 + Trap #1 silent-absorb ban + Trap #3 CP shape: - render_prompt_card must be dict. - render_prompt_card.render_strategy must be dict. - render_prompt_card.render_strategy.spatial_consistency must be dict. - compute_render_strategy_snapshot_hash(rpc) must succeed (no silent skip). - t2i_variations must be present, list, AND non-empty (R3-B1 carry — empty must fail, never silently absorbed). """ if "render_prompt_card" not in scene: measurement_failures.append( f"s{si}_sh{shi}: render_prompt_card key missing" ) return None rpc = scene["render_prompt_card"] if not isinstance(rpc, dict): measurement_failures.append( f"s{si}_sh{shi}: render_prompt_card not dict" ) return None if "render_strategy" not in rpc: measurement_failures.append( f"s{si}_sh{shi}: render_prompt_card.render_strategy key missing" ) return None rs = rpc["render_strategy"] if not isinstance(rs, dict): measurement_failures.append( f"s{si}_sh{shi}: render_prompt_card.render_strategy not dict" ) return None if "spatial_consistency" not in rs: measurement_failures.append( f"s{si}_sh{shi}: render_prompt_card.render_strategy." f"spatial_consistency key missing" ) return None sc = rs["spatial_consistency"] if not isinstance(sc, dict): measurement_failures.append( f"s{si}_sh{shi}: render_prompt_card.render_strategy." f"spatial_consistency not dict" ) return None try: # Lazy import — avoid circular boundary; see render_prompt_card_imports. from app.core.steps.render_prompt_card import ( # noqa: PLC0415 compute_render_strategy_snapshot_hash, ) compute_render_strategy_snapshot_hash(rpc) except Exception as exc: # noqa: BLE001 — surface ALL failure modes measurement_failures.append( f"s{si}_sh{shi}: render_strategy hash failed: {exc}" ) return None if "t2i_variations" not in scene: measurement_failures.append( f"s{si}_sh{shi}: t2i_variations key missing" ) return None variations = scene["t2i_variations"] if not isinstance(variations, list): measurement_failures.append( f"s{si}_sh{shi}: t2i_variations not list" ) return None if not variations: # R3-B1 carry — empty must fail (Trap #1 silent-absorb ban). measurement_failures.append( f"s{si}_sh{shi}: t2i_variations empty" ) return None return variations # Sentinel for explicit-None semantics — Trap #1 / PR-fix-iter-1-1 strict. # `dict.get(key, _DEFAULT_SENTINEL)` is the ONLY allowed `.get(...)` default # pattern in canary scripts. Any other default (`[] / {} / 0 / "" / None / # False`) is silent-absorb. _DEFAULT_SENTINEL = object() def load_pinning_from_args(args: Any) -> dict: """Construct pinning dict from --config JSON OR individual CLI flags. Returns dict with 7 keys (G4.5a pinning block): pid / scene_index_list / shot_index_list_per_scene / model_routing / prompt_source_mode / card_commit_hash / render_strategy_card_snapshot_hash Trap #1 — NO silent defaults; caller must run validate_pinning() afterwards. `_DEFAULT_SENTINEL` is the only allowed default — `None` round-trip is enforced explicitly in validate_pinning(). """ if getattr(args, "config", None): cfg_path = Path(args.config) if not cfg_path.exists(): print( f"[g4_5a_common] ERROR: --config file not found: {cfg_path}", file=sys.stderr, ) sys.exit(1) try: with cfg_path.open("r", encoding="utf-8") as fp: cfg = json.load(fp) except Exception as exc: # noqa: BLE001 print( f"[g4_5a_common] ERROR: --config parse error: {exc}", file=sys.stderr, ) sys.exit(1) # Explicit dict access — Trap #1 ban applies. We use `_DEFAULT_SENTINEL` # so absent keys round-trip as None and validate_pinning() rejects them. def _cfg_or_none(key: str) -> Any: val = cfg.get(key, _DEFAULT_SENTINEL) return None if val is _DEFAULT_SENTINEL else val return { "pid": _cfg_or_none("pid"), "scene_index_list": _cfg_or_none("scene_index_list"), "shot_index_list_per_scene": _cfg_or_none( "shot_index_list_per_scene" ), "model_routing": _cfg_or_none("model_routing"), "prompt_source_mode": _cfg_or_none("prompt_source_mode"), "card_commit_hash": _cfg_or_none("card_commit_hash"), "render_strategy_card_snapshot_hash": _cfg_or_none( "render_strategy_card_snapshot_hash" ), } try: scene_index_list = ( json.loads(args.scene_index_list) if getattr(args, "scene_index_list", None) else None ) shot_index_list_per_scene = ( json.loads(args.shot_index_list_per_scene) if getattr(args, "shot_index_list_per_scene", None) else None ) except Exception as exc: # noqa: BLE001 print( f"[g4_5a_common] ERROR: failed to parse list/dict JSON args: {exc}", file=sys.stderr, ) sys.exit(1) return { "pid": getattr(args, "pid", None), "scene_index_list": scene_index_list, "shot_index_list_per_scene": shot_index_list_per_scene, "model_routing": getattr(args, "model_routing", None), "prompt_source_mode": getattr(args, "prompt_source_mode", None), "card_commit_hash": getattr(args, "card_commit_hash", None), "render_strategy_card_snapshot_hash": getattr( args, "render_strategy_card_snapshot_hash", None ), } def validate_pinning(pinning: dict) -> None: """Fail-fast on missing pinning fields. Trap #1 — NO silent defaults.""" required = ( "pid", "scene_index_list", "shot_index_list_per_scene", "model_routing", "prompt_source_mode", "card_commit_hash", "render_strategy_card_snapshot_hash", ) # Explicit key existence check (Trap #1 — `.get(..., None)` is silent-absorb # if the key is absent, so we use `in` membership + `is None` value test). missing = [] for k in required: if k not in pinning: missing.append(k) continue v = pinning[k] if v is None or v == "": missing.append(k) if missing: print( f"[g4_5a_common] ERROR: pinning missing required field(s): " f"{missing}. Provide via --config or individual flags. " f"NO silent defaults.", file=sys.stderr, ) sys.exit(1) if not isinstance(pinning["scene_index_list"], list): print( "[g4_5a_common] ERROR: pinning.scene_index_list must be list[int]", file=sys.stderr, ) sys.exit(1) if not isinstance(pinning["shot_index_list_per_scene"], dict): print( "[g4_5a_common] ERROR: pinning.shot_index_list_per_scene " "must be dict", file=sys.stderr, ) sys.exit(1) def build_pinning_scene_set(pinning: dict) -> set[tuple[int, int]]: """Cartesian (scene_index, shot_index) tuples from pinning block. Trap #4 — pinning enumerates EXACT (si, shi) tuples. A scene_index without a corresponding shot_index_list entry (or with non-list / empty list) is a contract violation, not a silent default. """ out: set[tuple[int, int]] = set() shot_map = pinning["shot_index_list_per_scene"] for si in pinning["scene_index_list"]: si_int = int(si) # Try str then int key — pinning JSON uses str keys; CLI passes int. # `_DEFAULT_SENTINEL` is the only allowed default — explicit None round # trip + AppError-style fail-fast. raw_shots = shot_map.get(str(si_int), _DEFAULT_SENTINEL) if raw_shots is _DEFAULT_SENTINEL: raw_shots = shot_map.get(si_int, _DEFAULT_SENTINEL) if raw_shots is _DEFAULT_SENTINEL: print( f"[g4_5a_common] ERROR: pinning.scene_index_list contains " f"scene {si_int} but shot_index_list_per_scene has no entry. " f"NO silent default.", file=sys.stderr, ) sys.exit(1) if not isinstance(raw_shots, list): print( f"[g4_5a_common] ERROR: pinning.shot_index_list_per_scene" f"[{si_int}] must be list[int], got " f"{type(raw_shots).__name__}.", file=sys.stderr, ) sys.exit(1) if not raw_shots: print( f"[g4_5a_common] ERROR: pinning.shot_index_list_per_scene" f"[{si_int}] is empty — pinned scene must enumerate " f">=1 shot.", file=sys.stderr, ) sys.exit(1) for shi in raw_shots: out.add((si_int, int(shi))) return out def load_cp_manifest(cp_root: Path) -> dict: """Load `manifest.json` under cp_root. fail-fast on missing/parse error.""" manifest_path = cp_root / "manifest.json" if not manifest_path.exists(): print( f"[g4_5a_common] ERROR: manifest.json not found at " f"{manifest_path}", file=sys.stderr, ) sys.exit(1) try: with manifest_path.open("r", encoding="utf-8") as fp: return json.load(fp) except Exception as exc: # noqa: BLE001 print( f"[g4_5a_common] ERROR: manifest.json parse error: {exc}", file=sys.stderr, ) sys.exit(1) def resolve_cp_root(args: Any) -> Path: """Resolve cp_root from --config or --cp-root flag. Fail-fast on missing. Trap #1 — NO silent defaults. """ cp_root_str: str | None = None if getattr(args, "config", None): with Path(args.config).open("r", encoding="utf-8") as fp: cfg = json.load(fp) cfg_val = cfg.get("cp_root", _DEFAULT_SENTINEL) if cfg_val is not _DEFAULT_SENTINEL: cp_root_str = cfg_val if not cp_root_str: cp_root_str = getattr(args, "cp_root", None) if not cp_root_str: print( "[g4_5a_common] ERROR: --cp-root (or config.cp_root) required", file=sys.stderr, ) sys.exit(1) cp_root = Path(cp_root_str) if not cp_root.exists(): print( f"[g4_5a_common] ERROR: --cp-root not found: {cp_root}", file=sys.stderr, ) sys.exit(1) return cp_root def emit_output(out: dict, args: Any) -> None: """Write JSON output to args.output ('-' for stdout).""" if args.output == "-": json.dump(out, sys.stdout, ensure_ascii=False, indent=2, sort_keys=True) sys.stdout.write("\n") else: out_path = Path(args.output) out_path.parent.mkdir(parents=True, exist_ok=True) with out_path.open("w", encoding="utf-8") as fp: json.dump(out, fp, ensure_ascii=False, indent=2, sort_keys=True)