"""장소 신원 끝점 검사 — 한 run 의 production 산출을 원고의 **좌표** 선언과 대조한다.

★Codex 2026-09-02 계약(scenario-general):
  1) 서로 다른 물리적 업소·실내는 서로 다른 location ID · 인접 골목은 별도 location
  2) location_part 는 자기 업소 location 부모에만 결속
  3) 씬 primary_location 과 previous-shot/background 재사용은 **같은 location ID** 에서만
  4) 이름 목록·regex·하드코딩 금지 — 원고가 선언한 **글자 좌표**(씬 · 낱말 · 몇 번째)로만 대조
★실측 (run 69e821758f3d): 이발소·국밥집·골목이 L01 하나로 접혀 국밥집 샷이 이발소 앞 샷을
  「같은 장소의 이전 샷」으로 물려받았다.

원고 선언(fixture): `EXPECTED_SCENE_PLACES = {scene: (scene, 낱말, n)}` · `EXPECTED_PART_HOSTS =
[((부분 좌표), (자리 좌표)), …]`. 검사기는 그 좌표가 가리키는 span 을 production 행의
`occurrences[].source_span` 과 겹치는지로만 잇는다 — 낱말을 production 이름과 비교하지 않는다.

    python tools/grounding_audit/location_identity_check.py --run-id RID --fixture period_episode
"""
from __future__ import annotations

import argparse
import json
import sys
from pathlib import Path
from typing import Any, Dict, List, Optional

CHECK_CONTRACT = "3.202609022400"


def _overlaps(a: Dict[str, Any], b: Dict[str, Any]) -> bool:
    if str(a.get("segment_id")) != str(b.get("segment_id")):
        return False
    return int(a.get("start", 0)) < int(b.get("end", 0)) and int(b.get("start", 0)) < int(a.get("end", 0))


def _spans_of_row(row: Dict[str, Any]) -> List[Dict[str, Any]]:
    prov = row.get("grounding_provenance") or {}
    occ = prov.get("occurrences") or row.get("occurrences") or []
    return [o.get("source_span") or {} for o in occ if isinstance(o, dict)]


def _rows_covering(rows: List[Dict[str, Any]], span: Dict[str, Any]) -> List[Dict[str, Any]]:
    return [r for r in rows if any(_overlaps(s, span) for s in _spans_of_row(r))]


def load_outputs(root: Path) -> Dict[str, Any]:
    base = next(root.glob("projects/*/checkpoints/episodes/*"))

    def cp(step):
        p = base / step / "manifest.json"
        return json.loads(p.read_text(encoding="utf-8")) if p.is_file() else None

    return {"entity_merge": cp("entity_merge"), "scene_director": cp("scene_director"),
            "shot_dependency": cp("shot_dependency"), "reference_acquisition": cp("reference_acquisition"),
            "scene_detail": cp("scene_detail"), "scene_save": cp("scene_save")}


def spans_in_run(out: Dict[str, Any], scene: int, word: str, nth: int = 0) -> List[Dict[str, Any]]:
    """원고 좌표(씬 · 낱말 · 몇 번째)를 **production 이 실제로 읽은 세그먼트 글**(scene_save ·
    씬 제목 포함) 위의 span 으로. ★fixture 의 `span_of` 는 본문만 세어 production 의
    source_span(제목 포함 좌표)과 20자쯤 어긋난다 — 그래서 여기서는 production 글에서 센다.
    `nth=0` 이면 그 씬의 **모든** 자리 — 판독기가 제목을 인용하든 본문을 인용하든 잇는다."""
    segs = ((out.get("scene_save") or {}).get("data") or {}).get("segments") or []
    seg = next((x for x in segs if int(x.get("scene_index") or 0) == int(scene)), None)
    if seg is None:
        raise AssertionError(f"scene_save 에 씬 {scene} 이 없다")
    text = str(seg.get("text") or "")
    found: List[Dict[str, Any]] = []
    pos = text.find(word)
    while pos >= 0:
        found.append({"segment_id": f"scene-{scene}", "start": pos, "end": pos + len(word), "quote": word})
        pos = text.find(word, pos + 1)
    if not found or (nth and len(found) < nth):
        raise AssertionError(f"씬 {scene} 글에 {word!r} 가 {nth or 1}번째로 없다")
    return found if not nth else [found[nth - 1]]


def _rows_covering_any(rows: List[Dict[str, Any]], spans: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
    return [r for r in rows if any(_overlaps(s, sp) for s in _spans_of_row(r) for sp in spans)]


def check(run_id: str, fixture: str) -> Dict[str, Any]:
    from tools.grounding_audit import canary_bootstrap as cbs, canary_isolation as ci

    fx = cbs.load_fixture(fixture)
    places = dict(getattr(fx, "EXPECTED_SCENE_PLACES", {}) or {})
    hosts = list(getattr(fx, "EXPECTED_PART_HOSTS", []) or [])
    if not places:
        return {"contract": CHECK_CONTRACT, "ok": False, "why": f"원고 {fixture!r} 에 EXPECTED_SCENE_PLACES 선언이 없다"}
    out = load_outputs(ci.root_dir(run_id))
    got = check_outputs(out, fx)
    got.update({"run_id": run_id, "fixture": fixture})
    return got


def check_outputs(out: Dict[str, Any], fx: Any) -> Dict[str, Any]:
    """★순수 — run 디렉토리 없이 production 산출 dict 와 원고 모듈만으로 판정한다."""
    places = dict(getattr(fx, "EXPECTED_SCENE_PLACES", {}) or {})
    hosts = list(getattr(fx, "EXPECTED_PART_HOSTS", []) or [])
    em = (out.get("entity_merge") or {}).get("data") or {}
    locs = list(em.get("locations") or [])
    parts = list(em.get("location_parts") or [])
    findings: List[str] = []

    # 1·3-a) 씬마다 벌어지는 자리 → 정확히 한 location 행 · 씬 primary_location 과 같아야
    #  ★그 씬에서는 안 불렸지만(판독기가 씬 제목의 자리를 안 적는 「덜 적음」 — 실측 골목 씬 4)
    #   원고가 **같은 낱말**로 선언한 다른 씬의 자리를 덮는 행이 하나면 그 행으로 잇고 적어 둔다.
    scene_loc: Dict[int, Optional[str]] = {}
    covered_via: Dict[str, str] = {}
    for scene, at in sorted(places.items()):
        cov = _rows_covering_any(locs, spans_in_run(out, *at))
        ids = sorted({str(r.get("short_id")) for r in cov})
        if not ids:
            # ★원고가 선언한 **그 낱말**이 production 글의 다른 씬에 있는 자리들(좌표) — 그것을
            #  덮는 행이 정확히 하나면 그 행이다. 이름 비교가 아니라 같은 글자열의 span 이다.
            segs = ((out.get("scene_save") or {}).get("data") or {}).get("segments") or []
            other = []
            for sg in segs:
                s2 = int(sg.get("scene_index") or 0)
                if s2 == int(scene) or str(at[1]) not in str(sg.get("text") or ""):
                    continue
                other += spans_in_run(out, s2, str(at[1]), 0)
            alt = sorted({str(r.get("short_id")) for r in _rows_covering_any(locs, other)}) if other else []
            if len(alt) == 1:
                ids = alt
                covered_via[str(scene)] = f"같은 낱말({at[1]})의 다른 씬 자리로 이었다 → {alt[0]}"
        if len(ids) != 1:
            findings.append(f"씬{scene} 자리 좌표 {at} 를 덮는 location 행이 {ids or '없음'} — 정확히 하나여야 한다")
            scene_loc[int(scene)] = None
        else:
            scene_loc[int(scene)] = ids[0]
    sd = ((out.get("scene_director") or {}).get("data") or {}).get("scenes") or []
    primary = {int(s.get("scene_index")): str(s.get("primary_location") or "") for s in sd if s.get("scene_index") is not None}
    for scene, want in scene_loc.items():
        got = primary.get(scene)
        if want and got != want:
            findings.append(f"씬{scene} primary_location={got!r} 인데 자리 좌표는 {want} 를 가리킨다")

    # 1-b) 서로 다른 자리(다른 낱말 좌표)는 서로 다른 ID. ★같은 자리(같은 낱말)가 씬에
    #  따라 다른 ID 로 갈리는 것(한 업소의 앞과 안)은 **어긋남이 아니다** — 참조 사진이
    #  다른 두 세트다. 적어 두기만 한다(`same_place_split`).
    by_word: Dict[str, set] = {}
    for scene, at in places.items():
        if scene_loc.get(int(scene)):
            by_word.setdefault(str(at[1]), set()).add(scene_loc[int(scene)])
    same_place_split = {w: sorted(ids) for w, ids in by_word.items() if len(ids) > 1}
    words = list(by_word)
    for a in range(len(words)):
        for b in range(a + 1, len(words)):
            shared = by_word[words[a]] & by_word[words[b]]
            if shared:
                findings.append(f"서로 다른 자리({words[a]} · {words[b]})가 같은 location ID 로 접혔다 {sorted(shared)}")

    # 2) 부분 → 자기 자리의 행에 결속 (parent_final_id)
    ra_rows = (((out.get("reference_acquisition") or {}).get("data") or {}).get("rows")) or []
    parent_of: Dict[str, Optional[str]] = {}
    for r in ra_rows:
        lr = r.get("ledger_row") or {}
        if lr.get("owner_type") == "location_part" and lr.get("final_id"):
            parent_of.setdefault(str(lr["final_id"]), lr.get("parent_final_id"))
    for part_at, host_at in hosts:
        prow = _rows_covering_any(parts, spans_in_run(out, *part_at))
        hrow = _rows_covering_any(locs, spans_in_run(out, *host_at))
        pid = sorted({str(r.get("short_id")) for r in prow})
        hid = sorted({str(r.get("short_id")) for r in hrow})
        if len(pid) != 1 or len(hid) != 1:
            findings.append(f"부분 {part_at}→{pid} / 자리 {host_at}→{hid} — 각각 정확히 하나여야 한다")
            continue
        got_parent = parent_of.get(pid[0])
        if got_parent != hid[0]:
            findings.append(f"부분 {pid[0]} 의 parent_final_id={got_parent!r} 인데 자리 좌표는 {hid[0]} 다")

    # 3-b) cross-location background refs 0
    deps = (((out.get("shot_dependency") or {}).get("data") or {}).get("dependencies")) or []
    cross = []
    for d in deps:
        s = int(d.get("scene_index") or 0)
        for ref in d.get("location_refs") or []:
            s2 = int(ref.get("scene_index") or 0)
            if primary.get(s) and primary.get(s2) and primary[s] != primary[s2]:
                cross.append(f"{s}:{d.get('shot_index')}→{s2}:{ref.get('shot_index')} ({primary[s]} ≠ {primary[s2]})")
    if cross:
        findings.append(f"장소가 다른 샷을 배경으로 물려받은 의존 {len(cross)}: {cross[:6]}")

    return {"contract": CHECK_CONTRACT, "same_place_split": same_place_split, "covered_via": covered_via,
            "scene_location": {str(k): v for k, v in scene_loc.items()}, "primary_location": {str(k): v for k, v in primary.items()},
            "part_parent": parent_of, "cross_location_refs": cross, "findings": findings,
            "ok": not findings}


def main() -> int:
    ap = argparse.ArgumentParser()
    ap.add_argument("--run-id", required=True)
    ap.add_argument("--fixture", required=True)
    a = ap.parse_args()
    sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "tests"))
    got = check(a.run_id, a.fixture)
    from tools.grounding_audit import canary_isolation as ci
    p = ci.root_dir(a.run_id) / "location_identity_check.json"
    p.write_text(json.dumps(got, ensure_ascii=False, indent=2), encoding="utf-8")
    print(json.dumps({k: got[k] for k in ("ok", "scene_location", "primary_location", "part_parent", "cross_location_refs")}, ensure_ascii=False))
    for f in got.get("findings") or []:
        print("  ✗", f)
    print("  적었다:", p)
    return 0 if got.get("ok") else 1


if __name__ == "__main__":
    sys.path.insert(0, str(Path(__file__).resolve().parents[2]))
    raise SystemExit(main())