#!/usr/bin/env python3 # -*- coding: utf-8 -*- """Shared preflight helpers for G4.3 canary scripts (R3R4-B2 carry). All 4 metric scripts (`g4_3_body_part_focus.py`, `g4_3_close_framing_face_forbidden.py`, `g4_3_reproduction_surface.py`, `g4_3_demographic_descriptor_present.py`) import the helpers from this module and apply them identically. The token-count script (`g4_3_token_count.py`) does NOT use these helpers — it reads prompt files, not CP. NO silent absorb (Override R3R4-B2 / O-3 / O-4 / O-5): - `or [] / or {}` 패턴 금지. - `data.scenes` / `t2i_variations` / `t2i_prompt` / `render_prompt_card` / `id_policy` 누락은 measurement_failures 로 적재 후 caller 가 exit 1. - `compute_id_policy_snapshot_hash()` 호출 실패도 measurement_failures 로 적재 (silent skip 금지). """ 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)) # NOTE: deferred import — caller may set sys.path differently before importing # this module. We expose the symbol via a function rather than a module-level # import to keep the import surface small and predictable. def render_prompt_card_imports() -> dict: """Return the 6 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. """ from app.core.steps.render_prompt_card import ( # noqa: PLC0415 _ID_AGE_BANDS, _ID_BODY_PART_TRIGGERS, _ID_CLOSE_FACE_FORBIDDEN_PHRASES, _ID_ETHNICITY_COMPONENTS, _ID_REPRODUCTION_SURFACES, compute_id_policy_snapshot_hash, ) return { "_ID_AGE_BANDS": _ID_AGE_BANDS, "_ID_BODY_PART_TRIGGERS": _ID_BODY_PART_TRIGGERS, "_ID_CLOSE_FACE_FORBIDDEN_PHRASES": _ID_CLOSE_FACE_FORBIDDEN_PHRASES, "_ID_ETHNICITY_COMPONENTS": _ID_ETHNICITY_COMPONENTS, "_ID_REPRODUCTION_SURFACES": _ID_REPRODUCTION_SURFACES, "compute_id_policy_snapshot_hash": compute_id_policy_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. NO silent absorb. """ data = cp.get("data") if not isinstance(data, dict): measurement_failures.append("data missing or not dict") return None scenes = data.get("scenes") if not isinstance(scenes, list): measurement_failures.append("data.scenes missing or 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 / id_policy / variations. Returns variations list when valid; None otherwise (caller continues loop). Per spec §5.3 + Override R3R4-B2: - render_prompt_card must be dict. - render_prompt_card.id_policy must be dict. - compute_id_policy_snapshot_hash(rpc) must succeed (no silent skip). - t2i_variations must be list (may be empty — empty list is valid CP shape; per-variation validation happens in caller loop). """ rpc = scene.get("render_prompt_card") if not isinstance(rpc, dict): measurement_failures.append( f"s{si}_sh{shi}: render_prompt_card missing or not dict" ) return None id_policy = rpc.get("id_policy") if not isinstance(id_policy, dict): measurement_failures.append( f"s{si}_sh{shi}: render_prompt_card.id_policy missing or not dict" ) return None try: # Import lazily — see render_prompt_card_imports() rationale. from app.core.steps.render_prompt_card import ( # noqa: PLC0415 compute_id_policy_snapshot_hash, ) compute_id_policy_snapshot_hash(rpc) except Exception as exc: # noqa: BLE001 — surface ALL failure modes measurement_failures.append( f"s{si}_sh{shi}: id_policy hash failed: {exc}" ) return None variations = scene.get("t2i_variations") if not isinstance(variations, list): measurement_failures.append( f"s{si}_sh{shi}: t2i_variations missing or not list" ) return None return variations def load_pinning_from_args(args: Any) -> dict: """Construct pinning dict from --config JSON OR individual CLI flags. Returns dict with 7 keys (G4.3 pinning block): pid / scene_index_list / shot_index_list_per_scene / model_routing / prompt_source_mode / card_commit_hash / id_policy_card_snapshot_hash NO silent defaults — caller must run validate_pinning() afterwards. """ if getattr(args, "config", None): cfg_path = Path(args.config) if not cfg_path.exists(): print( f"[g4_3_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_3_common] ERROR: --config parse error: {exc}", file=sys.stderr) sys.exit(1) return { "pid": cfg.get("pid"), "scene_index_list": cfg.get("scene_index_list"), "shot_index_list_per_scene": cfg.get("shot_index_list_per_scene"), "model_routing": cfg.get("model_routing"), "prompt_source_mode": cfg.get("prompt_source_mode"), "card_commit_hash": cfg.get("card_commit_hash"), "id_policy_card_snapshot_hash": cfg.get("id_policy_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_3_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), "id_policy_card_snapshot_hash": getattr(args, "id_policy_card_snapshot_hash", None), } def validate_pinning(pinning: dict) -> None: """Fail-fast on missing pinning fields. NO silent defaults.""" required = ( "pid", "scene_index_list", "shot_index_list_per_scene", "model_routing", "prompt_source_mode", "card_commit_hash", "id_policy_card_snapshot_hash", ) missing = [k for k in required if pinning.get(k) in (None, "")] if missing: print( f"[g4_3_common] ERROR: pinning missing required field(s): {missing}. " f"Provide via --config or individual flags. NO silent defaults.", file=sys.stderr, ) sys.exit(1) if not isinstance(pinning["scene_index_list"], list): print( "[g4_3_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_3_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. Strict — 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) raw_shots = shot_map.get(str(si_int)) if raw_shots is None: raw_shots = shot_map.get(si_int) if raw_shots is None: print( f"[g4_3_common] ERROR: pinning.scene_index_list contains scene " f"{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_3_common] ERROR: pinning.shot_index_list_per_scene" f"[{si_int}] must be list[int], got {type(raw_shots).__name__}.", file=sys.stderr, ) sys.exit(1) if not raw_shots: print( f"[g4_3_common] ERROR: pinning.shot_index_list_per_scene" f"[{si_int}] is empty — pinned scene must enumerate >=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_3_common] ERROR: manifest.json not found at {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_3_common] ERROR: manifest.json parse error: {exc}", file=sys.stderr, ) sys.exit(1) def is_close_framing(scene: dict) -> tuple[bool, str | None]: """Return (cf_flag, error_message). Reads `scene["render_prompt_card"]["background_binding"]`. The G4.2 lift stores close-framing flag as either: - bb["mode"] == "skipped_close_framing", OR - bb["close_framing_skips_background_ref"] is True error_message is None on success; a human-readable string when CP shape is malformed (caller appends to measurement_failures — NO silent absorb). """ rpc = scene.get("render_prompt_card") if not isinstance(rpc, dict): return (False, "render_prompt_card missing or not dict") bb = rpc.get("background_binding") if not isinstance(bb, dict): return (False, "render_prompt_card.background_binding missing or not dict") if bb.get("mode") == "skipped_close_framing": return (True, None) if bb.get("close_framing_skips_background_ref") is True: return (True, None) return (False, None)