"""구간의 **샷 catalog** — 실제 체크포인트에서 동적으로 만든다. ★유료 0. ## 왜 샷인가 production 반복축(`entity_filter._appearance_count`)은 `shot_count` 를 **먼저** 본다. 씬 수로 세면 같은 씬의 서로 다른 샷에 반복 등장한 대상이 1회로 줄어 기존 축이 바뀐다 (Codex 2026-08-31). ## 왜 모델에게 숫자를 안 물어보나 「몇 번 나왔나」를 숫자로 받으면 되짚을 수가 없다. 대신 **그 구간의 실제 샷 ID 를 주고 고르게** 하고, 세는 것은 코드가 한다. ID 는 새로 지어내지 않는다 — `shot_validator` 체크포인트가 이미 `scene_index` + `shot_index` 를 갖고 있다. ## 상태가 셋인 까닭 `shot_appearance_ids: []` 하나로는 **「원문엔 있는데 이 샷들엔 안 보인다」**와 **「모델이 못 붙였다」**가 둘 다 0회가 된다. 그래서 상태를 함께 받는다. bound_complete ids ≥ 1 · catalog ID 만 · unique → 고유 ID 수를 센다 not_in_catalog_shots ids 는 **정확히 빈 배열** → 0회 (명시 판정) unresolved 못 정함 → **미확정** (false 아님) ★나체 `[]` 는 금지다. """ from __future__ import annotations from typing import Any, Dict, List, Sequence, Tuple #: 결속 상태 — ★셋이다. 빈 배열 하나로 뭉치면 두 뜻이 합쳐진다. BOUND_COMPLETE = "bound_complete" NOT_IN_CATALOG = "not_in_catalog_shots" BIND_UNRESOLVED = "unresolved" BIND_STATES = (BOUND_COMPLETE, NOT_IN_CATALOG, BIND_UNRESOLVED) def shot_id(scene_index: Any, shot_index: Any) -> str: """안정 샷 ID. ★`shot_validator` 가 이미 갖고 있는 두 좌표로만 만든다.""" s = str(scene_index or "").strip() t = str(shot_index or "").strip() if not s or not t: raise ValueError(f"샷 좌표가 비었다 (scene={s!r} shot={t!r})") return f"s{s}#{t}" def build_catalog(shot_scenes: Sequence[Dict[str, Any]], segment_ids: Sequence[str]) -> List[Dict[str, Any]]: """이 구간에 속한 샷들의 catalog. ★**설명을 자르지 않는다**. 임의 절단은 구별점을 없애 **잘못된 샷 결속**을 만들고, 그 결함은 validator 가 못 잡는다 (Codex). 상한을 넘으면 자르지 말고 구간을 더 나눈다. Args: shot_scenes: `shot_validator` 체크포인트의 `scenes[]` 그대로. segment_ids: 이 구간의 씬 id (`scene-`). """ want = set() for sid in segment_ids: s = str(sid) want.add(s.split("-", 1)[1] if s.startswith("scene-") else s) out: List[Dict[str, Any]] = [] seen: set = set() for sc in shot_scenes or (): idx = str((sc or {}).get("scene_index") or "").strip() if idx not in want: continue for sh in (sc or {}).get("shots") or (): sid = shot_id(idx, (sh or {}).get("shot_index")) if sid in seen: raise ValueError(f"샷 ID 가 겹친다: {sid}") seen.add(sid) row: Dict[str, Any] = { "id": sid, "scene_id": f"scene-{idx}", # ★검증된 **전문**. 앞 N자 자르기 금지. "description": str((sh or {}).get("description") or ""), } chars = (sh or {}).get("characters") if chars: # ★이미 구조화돼 있는 것만 그대로 옮긴다 — 새로 안 짓는다. row["characters"] = chars out.append(row) return out def catalog_ids(catalog: Sequence[Dict[str, Any]]) -> List[str]: return [str(c.get("id")) for c in catalog or ()] def scene_of(catalog: Sequence[Dict[str, Any]]) -> Dict[str, str]: """샷 ID → 그 샷이 속한 씬. ★씬 밖 결속을 잡는 데 쓴다.""" return {str(c.get("id")): str(c.get("scene_id")) for c in catalog or ()} def patch_schema_with_shot_ids(schema: Dict[str, Any], catalog: Sequence[Dict[str, Any]] ) -> Dict[str, Any]: """`shot_appearance_ids` 를 **이 호출의 catalog 로** 좁힌다 (runtime enum). ★고정 목록을 팩에 안 박는다 — 원고마다 다르고, 박으면 다른 원고에서 거짓말이 된다. """ import copy ids = catalog_ids(catalog) out = copy.deepcopy(schema) props = out["properties"]["rows"]["items"]["properties"] node = props.get("shot_appearance_ids") if not isinstance(node, dict): raise KeyError("schema 에 `shot_appearance_ids` 가 없다") node["uniqueItems"] = True if ids: node["items"] = {"type": "string", "enum": ids} node.pop("maxItems", None) else: # ★★catalog 가 비면 **문을 닫는다** (Codex 2026-08-31). 앞 판은 # `{"type": "string"}` 만 남겨서 **아무 문자열이나** 통과했다 — # 「그 호출의 ID 만」이 아니었다. 빈 enum 은 provider 가 거절할 수 # 있으니 `maxItems: 0` 으로 닫는다. node["items"] = {"type": "string"} node["maxItems"] = 0 return out def verify_binding(status: Any, ids: Any, catalog: Sequence[Dict[str, Any]], row_scenes: Sequence[str]) -> Tuple[str, List[str], str]: """한 행의 샷 결속을 검사한다. Returns: `(상태, 받아들인 ID, 사유)`. 사유는 **뺀 ID 들의 까닭**이다 — 행은 살고 (`salvage_problems` 에 남는다), 검증된 ID 가 0 이면 상태만 `unresolved` 다. ★`row_scenes` 는 그 행의 occurrence 가 있는 씬들이다. 샷이 그 밖이면 fail-closed — 이름·substring 으로 추정하지 않는다. """ st = str(status or "") got = list(ids or []) if st not in BIND_STATES: return BIND_UNRESOLVED, [], f"모르는 결속 상태 {st!r} — {BIND_STATES}" known = set(catalog_ids(catalog)) where = scene_of(catalog) if st == NOT_IN_CATALOG: if got: return BIND_UNRESOLVED, [], ( f"`{NOT_IN_CATALOG}` 인데 ID 가 {len(got)}개 있다 — " "「이 샷들엔 안 보인다」와 어긋난다") return st, [], "" if st == BIND_UNRESOLVED: # ★★감사용 ID 도 **같은 검사**를 거친다 (Codex 2026-08-31). # 앞 판은 여기서 검사를 아예 안 타서 catalog 밖 ID·중복·씬 밖 ID 가 # **사유 없이 조용히 사라졌다**. 「행째 격리」라고 보고해 놓고 # 실제로는 삼키고 있었다. # ★상태는 `unresolved` 로 두고, 어긋난 ID 는 **사유와 함께 뺀다**. keep, why = _check_ids(got, known, where, row_scenes) return st, keep, why # bound_complete if not got: return BIND_UNRESOLVED, [], ( f"`{BOUND_COMPLETE}` 인데 ID 가 없다 — 나체 빈 배열은 금지다") keep, why = _check_ids(got, known, where, row_scenes) if not keep: # ★검증된 결속이 하나도 없다 — 「붙였다」는 말이 서지 않는다 return BIND_UNRESOLVED, [], why # ★★검증된 것만 남긴다 — 하나가 어긋났다고 나머지 여섯을 버리지 않는다 # (2026-09-02 실측: 씬 밖 결속 한 개로 장소 셋·인물·부분 둘이 **행째** 사라졌다). # 뺀 것은 `why` 로 올라가 행의 `salvage_problems` 에 남는다. return st, keep, why def _check_ids(got: Sequence[Any], known: set, where: Dict[str, str], row_scenes: Sequence[str]) -> Tuple[List[str], str]: """샷 ID 목록을 검사한다. ★`bound_complete` 든 `unresolved` 든 **같다**. Returns: `(검증된 ID, 뺀 사유들)`. ★첫 어긋남에서 서지 않는다 — 어긋난 ID 는 하나씩 사유를 적어 빼고, 검증된 것은 남긴다.""" seen: set = set() dropped: List[str] = [] for x in got: x = str(x) if x not in known: dropped.append(f"catalog 밖 샷 ID {x!r}") continue if x in seen: dropped.append(f"샷 ID 가 겹친다 {x!r}") continue if row_scenes and where.get(x) not in set(row_scenes): dropped.append(f"샷 {x!r} 은 {where.get(x)!r} 것인데 이 행의 출현은 " f"{sorted(set(row_scenes))} 다 — 씬 밖 결속이다") continue seen.add(x) return sorted(seen), "; ".join(dropped)