#!/usr/bin/env python3
"""`key_bg_elements[].state` 가 **제작자가 본 근거**에 뒷받침되나 (감사 C-2).

## 첫 판은 폐기했다 — 재는 도구에 결정적 오류가 셋 있었다 (Codex 반증)

    ① **옛 체크포인트를 섞었다.** `config_hash` 를 안 보고 모든 CP 를
       모아, 30표본이 **22 에피소드 · 계약 5판 · 2026-04~08** 에 걸쳤다.
       현행 system+schema 로 다시 계산해 일치하는 것은 **6 에피소드
       638샷**뿐이고 그 30개 중 현행은 **4건**이었다.
       ★`clean/dirty` 5건은 **전부 5~6월 구 CP** 라 「현행 10/10
        unsupported」의 근거가 아니었다. 발명 경계 규칙보다 이전 판이다.
    ② **판정자에게 제작자가 본 근거를 덜 줬다.** 실제 `shot_staging`
       호출은 같은 batch(10개)의 **선택된 샷 묘사 전부** + 씬 원문 +
       제작자 정정을 본다. 내 도구는 **씬 원문만** 보냈다.
       그래서 유일한 도달 사례도 거짓이었다 — `worn bench (worn)` 은
       같은 batch 의 S31sh5 묘사 「**낡은 벤치**에 나란히 앉아…」가
       근거였다.
    ③ **층화 정규식이 축을 잘못 잡았다.** `\\b(on|off|…)\\b` 가 전치사
       `on` 을 잡아 「Fixed prominently **on** the workroom wall」을
       on/off 표본에 넣었고, `dark`(밝기)도 on/off 로 셌다. 5건 중 3건이
       그 축이 아니었다.

## 그래서 이 판은 이렇게 잰다

* **현행 effective hash 와 일치하는 CP 만** — system+schema+project_config
  로 스텝과 **같은 식**으로 다시 계산한다
* 판정자에게 **제작자가 본 것을 그대로** 준다 — 그 batch 의 선택된 샷
  묘사 전부 + 씬 원문(★자르지 않는다) + 제작자 정정
* 축은 **상태 토큰**으로 가른다(쉼표로 끊어 낱말이 그 자체로 상태일 때만)
* `state` 만 본다 — `camera_use`·`lighting_mood`·`orientation` 제외
* 판정은 gemini-3.1-pro + grok **각각**, 합의로 꾸미지 않는다
* **dry plan 을 사람이 먼저 본다.** 그 전에 유료 호출 0

## 이 조사의 결론 (2026-08-28 · Codex 확인 후 종결)

> 현행 계약이 clean/dirty·wet/dry 같은 근거 없는 상태를 **강제로
> 생산한다**는 C-2 가설은 **반증됐다.** 현행 6 CP 에 그 두 축은 0건이고,
> 여섯 축에 걸린 52건(비어 있지 않은 state **1,344개**의 3.9%) 중 표본
> 17건은 Codex 가 대상 샷 묘사·씬 원문과 직접 대조해 **전부 직접 근거
> 또는 행동상 필연**임을 확인했다(독립 근거 단위 14개). **현행 모든
> state 가 안전하다고 일반화하지 않는다.** 추가 VLM 호출과 코드 수정
> 없이 C-2 를 닫는다. 새 주행·최종 갤러리에서 근거 없는 state 가 실제로
> 발견되면 그 샷으로 재개한다.

★이 도구는 「제작자가 본 것 그대로」가 **아니다** — 실제 입력에는 Beat·
 인물·camera flow·등록 인물·t2i_context·제작자 정정이 더 있고 여기는
 묘사+씬 원문만 보낸다. 현행 6 CP 는 배치 순서가 같고 active correction
 이 0이라 위 수치는 뒤집히지 않지만, **일반적 결론에는 못 쓴다.**

usage:  probe_state_support.py [N]        # 기본 dry — 유료 호출 0
        probe_state_support.py --run [N]  # 실제로 gemini+grok 판정을 산다
"""
from __future__ import annotations

import collections
import hashlib
import json
import pathlib
import random
import re
import sys

sys.path.insert(0, str(pathlib.Path(__file__).resolve().parents[2]))

ROOT = pathlib.Path(__file__).resolve().parents[3]
OUT = ROOT / "artifact" / "20260828_state_support"
BATCH_SIZE = 10          # shot_staging.py:25 와 같아야 한다

# 상태 토큰 — 쉼표로 끊은 조각이 **그 자체로** 이 상태일 때만 그 축이다.
# (★전치사 `on`, 밝기 `dark` 를 배제한다 — 첫 판이 여기서 틀렸다)
AXIS_TOKENS = {
    "open/closed": {"open", "closed", "ajar", "shut", "locked", "unlocked",
                    "half-open", "fully open", "wide open"},
    "on/off": {"on", "off", "lit", "unlit", "switched on", "switched off",
               "powered", "powered off", "glowing"},
    "clean/dirty": {"clean", "dirty", "stained", "grimy", "dusty", "spotless",
                    "soiled", "smudged"},
    "wet/dry": {"wet", "damp", "dry", "soaked", "puddled", "moist"},
    "empty/occupied": {"empty", "occupied", "full", "vacant", "crowded",
                       "bare"},
    "damaged/intact": {"broken", "cracked", "damaged", "intact", "torn",
                       "shattered", "worn", "chipped"},
}


def _axis_of(state: str):
    """쉼표 조각이 **그 자체로** 상태 낱말일 때만 그 축으로 센다."""
    for piece in re.split(r"[,;/]", state):
        w = piece.strip().lower().rstrip(".")
        for ax, toks in AXIS_TOKENS.items():
            if w in toks:
                return ax
    return None


def _current_effective_hash(project_config) -> str:
    """스텝(`shot_staging_step._config_hash`)과 **같은 식**으로 계산한다."""
    from app.core.step_runner import compute_config_hash
    from app.modules.prompt_loader import load_prompt, load_schema

    h = hashlib.sha256()
    h.update(load_prompt("shot_staging", "system").encode("utf-8"))
    h.update(b"\x00")
    h.update(json.dumps(
        load_schema("shot_staging", "schema"), sort_keys=True).encode("utf-8"))
    h.update(b"\x00")
    h.update(compute_config_hash(project_config or {}).encode("utf-8"))
    return h.hexdigest()[:16]


def _episode_inputs(epi_dir: pathlib.Path):
    """제작자가 본 것 — 씬 원문 · 선택된 샷 묘사 · 제작자 정정."""
    scenes, desc, order = {}, {}, []
    for step in ("scene_save", "scene_segmentation"):
        f = epi_dir / step / "manifest.json"
        if not f.exists():
            continue
        try:
            dd = json.loads(f.read_text(encoding="utf-8"))["data"]
        except Exception:
            continue
        for s in (dd.get("segments") or []):
            if isinstance(s, dict) and s.get("scene_index") is not None:
                if s.get("text"):
                    scenes.setdefault(int(s["scene_index"]), s["text"])
        if scenes:
            break
    f = epi_dir / "shot_validator" / "manifest.json"
    if f.exists():
        try:
            dd = json.loads(f.read_text(encoding="utf-8"))["data"]
            for sc in (dd.get("scenes") or []):
                si = sc.get("scene_index")
                for sh in (sc.get("shots") or []):
                    desc[(si, sh.get("shot_index"))] = sh.get("description", "")
        except Exception:
            pass
    f = epi_dir / "shot_selection" / "manifest.json"
    if f.exists():
        try:
            dd = json.loads(f.read_text(encoding="utf-8"))["data"]
            for sc in (dd.get("scenes") or []):
                si = sc.get("scene_index")
                for shi in (sc.get("selected_shot_indices") or []):
                    order.append((si, shi))
        except Exception:
            pass
    return scenes, desc, order


def collect():
    rows = []
    kept = skipped = 0
    for m in sorted(ROOT.glob(
            "projects/*/checkpoints/episodes/*/shot_staging/manifest.json")):
        try:
            d = json.loads(m.read_text(encoding="utf-8"))
        except Exception:
            continue
        pc = d.get("project_config_snapshot") or d.get("project_config") or {}
        # ★현행 계약과 일치하는 CP 만 — 옛 판을 지금 규칙으로 재면 안 된다
        if d.get("config_hash") != _current_effective_hash(pc):
            skipped += 1
            continue
        kept += 1
        epi_dir = m.parents[1]
        scenes, desc, order = _episode_inputs(epi_dir)
        idx = {k: i for i, k in enumerate(order)}
        shots = (d.get("data") or {}).get("shots") or []
        for sh in shots:
            if not isinstance(sh, dict):
                continue
            si, shi = sh.get("scene_index"), sh.get("shot_index")
            key = (si, shi)
            if key not in idx or si not in scenes:
                continue
            # 그 샷이 실린 batch(10개) 전체 — 제작자가 한 호출에서 본 것
            b0 = (idx[key] // BATCH_SIZE) * BATCH_SIZE
            batch = order[b0:b0 + BATCH_SIZE]
            for el in (sh.get("key_bg_elements") or []):
                if not isinstance(el, dict):
                    continue
                st = (el.get("state") or "").strip()
                ax = _axis_of(st)
                if not st or ax is None:
                    continue
                rows.append({
                    "epi": epi_dir.name[:8], "scene_index": si,
                    "shot_index": shi, "element": el.get("element", ""),
                    "state": st, "axis": ax,
                    "batch": [{"scene_index": a, "shot_index": b,
                               "description": desc.get((a, b), ""),
                               "scene_text": scenes.get(a, "")}
                              for a, b in batch],
                })
    print(f"현행 계약과 일치하는 shot_staging CP {kept} · 건너뛴 옛 CP {skipped}")
    return rows


SYSTEM = (
    "You check whether a stated fact about a film set is supported by the "
    "material the planner actually saw. You are given every shot brief the "
    "planner received in the same request (each with its scene text), and "
    "one background element with the state the planner wrote for one of "
    "those shots. Decide ONLY whether that material — or an action it "
    "describes — supports the state. Do not judge whether it is a good idea."
)
SCHEMA = {
    "type": "object",
    "properties": {
        "verdict": {"type": "string", "enum": [
            "supported", "action_derived", "unsupported"]},
        "evidence_quote": {"type": "string"},
        "reason_en": {"type": "string"},
    },
    "required": ["verdict", "evidence_quote", "reason_en"],
    "additionalProperties": False,
}


def _brief(r) -> str:
    seen, parts = set(), []
    for b in r["batch"]:
        tag = f"[씬{b['scene_index']} Shot{b['shot_index']}]"
        parts.append(f"{tag}\n  묘사: {b['description']}")
        if b["scene_index"] not in seen and b["scene_text"]:
            seen.add(b["scene_index"])
            # ★씬 원문을 자르지 않는다 (저장소 절대 규칙)
            parts.append(f"  씬{b['scene_index']} 원문: {b['scene_text']}")
    return "\n\n".join(parts)


def main() -> int:
    # ★기본이 dry 다 (Codex 리뷰 #37 지적 — 돈 부류). 인자 없이 돌리면
    #  곧장 gemini+grok 유료 판정으로 들어가던 것을 뒤집었다. 실제로 사려면
    #  `--run` 을 손으로 적는다. `--dry` 는 옛 호출을 위해 남겨 둔다.
    dry = "--run" not in sys.argv
    n_each = next((int(a) for a in sys.argv[1:] if a.isdigit()), 5)
    rows = collect()
    by = collections.defaultdict(list)
    for r in rows:
        by[r["axis"]].append(r)
    rnd = random.Random(20260828)
    pick = []
    for ax in AXIS_TOKENS:
        g = by.get(ax, [])[:]
        rnd.shuffle(g)
        pick += g[:n_each]
    # ★분모를 정직하게 적는다 (Codex 지적, 2026-08-28). `len(rows)` 는
    #  **여섯 축 토큰에 걸린 것**이지 state 전체가 아니다. 현행 6 CP 의
    #  비어 있지 않은 state 는 **1,344개**다 — 52 는 그 3.9%, 표본 17 은
    #  1.3% 이고, 같은 근거가 겹쳐 **독립 근거 단위는 14개**다.
    #  「17/52 라 사실상 1/3 전수」는 틀린 해석이었다.
    print(f"\n★여섯 축 토큰에 걸린 state {len(rows):,} "
          f"(현행 CP 의 비어 있지 않은 state 전체가 **아니다**) "
          f"· 축별 표본 {n_each}")
    for ax in AXIS_TOKENS:
        print(f"   {ax:16} 후보 {len(by.get(ax, [])):5,}  뽑음 "
              f"{min(n_each, len(by.get(ax, [])))}")
    OUT.mkdir(parents=True, exist_ok=True)
    (OUT / "plan.json").write_text(
        json.dumps(pick, ensure_ascii=False, indent=1), encoding="utf-8")
    print(f"\n표본 {len(pick)}건 × 모델 2 = {len(pick) * 2} 호출 · 이미지 생성 0")
    print("   ★이 표본으로 **현행 state 전체**를 말하지 마라 — 여섯 축에"
          " 걸린 것만 보고, 같은 근거가 겹친다")
    if dry:
        print("\n★사람이 먼저 볼 것 — 축이 맞게 잡혔나, 근거가 다 실렸나\n")
        for r in pick:
            print(f"  [{r['axis']:14}] {r['epi']}:S{r['scene_index']}"
                  f"sh{r['shot_index']}  {r['element'][:24]:26} "
                  f"state={r['state'][:46]!r}")
            print(f"       batch {len(r['batch'])}샷 · 근거 "
                  f"{len(_brief(r)):,}자")
        print("\n기본이 dry 라 호출 0 — 실제로 사려면 --run")
        return 0

    from app.modules.llm.dual_vlm import ask_both

    res = []
    for i, r in enumerate(pick, 1):
        head = (
            "SHOT BRIEFS THE PLANNER RECEIVED IN THIS ONE REQUEST "
            "(verbatim, Korean):\n" + _brief(r) +
            f"\n\n── THE VALUE UNDER TEST\n"
            f"It was written for [씬{r['scene_index']} "
            f"Shot{r['shot_index']}].\n"
            f"BACKGROUND ELEMENT: {r['element']}\n"
            f"STATE THE PLANNER WROTE: {r['state']}\n\n"
            "- verdict: supported (the material states it), action_derived "
            "(it follows necessarily from an action described), or "
            "unsupported (nothing in the material supports it).\n"
            "- evidence_quote: the exact Korean substring you relied on, or "
            "an empty string when unsupported.\n"
            "- reason_en: one sentence.")
        dual = ask_both("state_support", SYSTEM,
                        [{"type": "text", "text": head}], SCHEMA,
                        schema_name="state_support")
        row = {**{k: v for k, v in r.items() if k != "batch"},
               "brief_chars": len(head), "by_model": {}}
        for c in dual.calls:
            row["by_model"][c.alias] = {"ok": c.ok, "payload": c.payload}
        res.append(row)
        vs = [((v.get("payload") or {}).get("verdict") or "?")
              for v in row["by_model"].values() if v["ok"]]
        print(f"  [{i:2d}/{len(pick)}] {r['axis']:14} "
              f"{r['element'][:18]:20} {vs}")
        (OUT / "verdicts.json").write_text(
            json.dumps(res, ensure_ascii=False, indent=1), encoding="utf-8")

    print("\n판정 (모델별로 따로 — 합의로 꾸미지 않는다)")
    agg = collections.defaultdict(collections.Counter)
    for row in res:
        for a, v in row["by_model"].items():
            if v["ok"]:
                agg[a][(v.get("payload") or {}).get("verdict")] += 1
    for a, c in agg.items():
        print(f"   {a:12} {dict(c)}")
    both = [r for r in res
            if (vs := {(v.get('payload') or {}).get('verdict')
                       for v in r['by_model'].values() if v['ok']})
            and vs == {"unsupported"}]
    print(f"\n두 모델이 **둘 다** unsupported {len(both)}/{len(res)}")
    for r in both:
        print(f"   {r['epi']}:S{r['scene_index']}sh{r['shot_index']}  "
              f"{r['element'][:24]:26} {r['state'][:44]!r}")
    return 0


if __name__ == "__main__":
    raise SystemExit(main())