"""run 하나의 조사 단계 acceptance — chunk 지문·처리 계약·장소·아웃룩 의무·조사 완료·장소 검사기. 무료.

    python tools/grounding_audit/research_acceptance.py --run-id RID --fixture period_episode
"""
import json, os, sys, glob
from pathlib import Path
sys.path.insert(0, str(Path(__file__).resolve().parents[2])); sys.path.insert(0, str(Path(__file__).resolve().parents[2] / "tests"))
import argparse
_ap = argparse.ArgumentParser(); _ap.add_argument("--run-id", required=True); _ap.add_argument("--fixture", required=True)
_a = _ap.parse_args(); RID = _a.run_id; FIXTURE = _a.fixture
os.environ.setdefault("BACKGROUND_MODE", "off"); os.environ.setdefault("STILL_RECIPE_MODE", "off")
for k in ("OUTDOOR_LANE_PLAN_ENABLED", "OUTDOOR_LANE_PIPE_ENABLED", "OUTDOOR_DIRECT_COMPOSE_ENABLED", "OUTDOOR_MAP_CONTI_ENABLED"):
    os.environ.setdefault(k, "false")
from tools.grounding_audit import canary_bootstrap as cbs
os.environ.update(cbs.prepare_env(RID, grounding_mode="v2_chunk"))
from tools.grounding_audit import canary_run as cr, canary_isolation as ci
from app.core.database import SessionLocal
from app.services.analysis_dispatch_service import get_step_runner
root = ci.root_dir(RID)
base = next(root.glob("projects/*/checkpoints/episodes/*"))
pid, eid = base.parts[-4], base.parts[-1]
cfg = cr._owner_requirement(FIXTURE); cfg["grounding_mode"] = "v2_chunk"
db = SessionLocal()
try:
    r = get_step_runner("grounding_chunk", pid, eid, db, cfg, {})
    cp = r.load_checkpoint() or {}
    local = r._config_hash()
finally:
    db.close()
data = cp.get("data") or {}
rows = []
rows.append(("chunk CP config_hash == 현재 _config_hash", cp.get("config_hash") == local, f"{(cp.get('config_hash') or '')[:12]} vs {local[:12]}"))
rows.append(("contracts.processing == 4.202609022340", (data.get("contracts") or {}).get("processing") == "4.202609022340", str((data.get("contracts") or {}).get("processing"))))
rows.append(("grounding_salvaged 존재", "grounding_salvaged" in data, f"{len(data.get('grounding_salvaged') or [])}행 · quarantined {len(data.get('grounding_quarantined') or [])}"))
rows.append(("locations 수", len(data.get("locations") or []) >= 3, str([(l.get('short_id'), l.get('name')) for l in data.get('locations') or []])))
# ── 아웃룩 부모 결속 (Codex acceptance): outlook 전부 character 부모 · 잘못된 O→O 0 · facet debt 의 outlook 0
em = json.loads((base / "entity_merge" / "manifest.json").read_text(encoding="utf-8"))["data"]
outlooks = [o.get("short_id") for o in (em.get("outlooks") or em.get("character_outlooks") or [])]
part_of = data.get("grounding_part_of") or []
o2c = [x for x in part_of if str(x.get("part", "")).startswith("O") and str(x.get("whole", "")).startswith("C")]
o2o = [x for x in part_of if str(x.get("part", "")).startswith("O") and str(x.get("whole", "")).startswith("O")]
debt_outlooks = [x for x in (data.get("grounding_facet_debt") or []) if x.get("owner_type") == "outlook"]
# ★계약: 조사 **대상**이 되는 아웃룩은 전부 character 부모가 있고, 조각(O→O)은 대상이 아니어야 한다. 조각·미등록은 감사 사실.
o_targets = set()
ra_p0 = base / "reference_acquisition" / "manifest.json"
if ra_p0.is_file():
    for r in json.loads(ra_p0.read_text(encoding="utf-8"))["data"].get("rows") or []:
        lr = r.get("ledger_row") or {}
        if lr.get("owner_type") == "outlook" and lr.get("final_id") and r.get("disposition") not in ("auto_completed", "not_applicable"):
            o_targets.add(str(lr["final_id"]))
o_pieces = {str(x.get("part")) for x in o2o}
parents = {str(x.get("part")): str(x.get("whole")) for x in o2c}
whole_of = {str(x.get("part")): str(x.get("whole")) for x in part_of}
def _root(node, seen=()):
    """O→O 사슬을 따라 O→C 뿌리까지 — cycle 이면 None."""
    if node in seen: return None
    nxt = whole_of.get(node)
    if nxt is None: return None
    if nxt.startswith("C"): return nxt
    return _root(nxt, seen + (node,))
chains = {pc: _root(pc) for pc in sorted(o_pieces)}
ok_bind = (bool(o_targets) and all(t in parents for t in o_targets)
           and not (o_targets & o_pieces) and all(v is not None for v in chains.values()))
rows.append(("outlook 조사 대상 전부 character 부모 · 조각(O→O)은 대상 아니고 사슬이 C 뿌리에 닿음(cycle 0)", ok_bind,
             f"대상 {sorted(o_targets)} → {[parents.get(t) for t in sorted(o_targets)]} · 조각 사슬 {chains} · 미등록 조각 debt {len(debt_outlooks)} · O→C {len(o2c)}"))
# ── 조사 결과 표
ra_p = base / "reference_acquisition" / "manifest.json"
if ra_p.is_file():
    ra = json.loads(ra_p.read_text(encoding="utf-8"))
    rrows = ra["data"].get("rows") or []
    import collections
    outc = collections.Counter(r.get("outcome") for r in rrows)
    retry = [((r.get("ledger_row") or {}).get("final_id")) for r in rrows if ((r.get("acquisition") or {}).get("status")) == "retryable" and (r.get("acquisition") or {}).get("rounds")]
    rows.append(("조사 스텝 completed · retryable 잔존 0", ra.get("status") == "completed" and not retry, f"status={ra.get('status')} · outcome {dict(outc)} · retryable 잔존 {retry} · rejudge_pending {ra['data'].get('rejudge_pending')}"))
    out_rows = [r for r in rrows if (r.get("ledger_row") or {}).get("owner_type") == "outlook" and r.get("disposition") not in ("auto_completed", "not_applicable")]
    n_auto = sum(1 for r in rrows if r.get("disposition") == "auto_completed")
    n_na = sum(1 for r in rrows if r.get("disposition") == "not_applicable")
    needed = [r for r in rrows if r.get("disposition") not in ("not_applicable",) and r.get("outcome")]
    sel = sum(1 for r in needed if r.get("outcome") == "selected"); un = sum(1 for r in needed if r.get("outcome") == "reference_unavailable")
    disp = [r for r in needed if r.get("disposition") != "auto_completed"]
    rows.append(("분모 갈라 적기: 참조 필요 = selected + unavailable · 구매 대상 = 그중 auto_completed 뺀 것", len(needed) == sel + un and len(disp) == ra["data"].get("purchases", {}).get("targets_total"),
                 f"참조 필요 {len(needed)} = selected {sel} + unavailable {un} · 구매 대상 {len(disp)} = selected {sum(1 for r in disp if r.get('outcome')=='selected')} + unavailable {sum(1 for r in disp if r.get('outcome')=='reference_unavailable')} · auto_completed(미등록) {n_auto} · not_applicable {n_na} · rows {len(rrows)}"))
    # ★언어 잠금 감사(비차단): 나간 질의가 잠금 언어의 글자를 하나도 안 담으면 어김 — 글자 종류(스크립트)만 본다
    def _has_hangul(t):
        """잠금 언어의 글자가 **주된** 글자인가 — 한글 음절 수가 라틴 글자 수보다 많아야 한다(좌표 낱말만 한글인 영어 질의는 어김).
        ★검색 연산자(`site:` · `filetype:` …)와 URL 은 언어가 아니다 — 세기 전에 뺀다 (2026-09-03 실측: `site:emuseum.go.kr 됫박 …` 이 어김으로 잡혔다)."""
        import re as _re
        t = _re.sub(r"\b[a-z]+:\S+", " ", str(t))          # site:host · filetype:pdf 같은 연산자 토큰
        t = _re.sub(r"https?://\S+", " ", t)
        h = sum(1 for ch in t if '\uac00' <= ch <= '\ud7a3'); l = sum(1 for ch in t if ch.isascii() and ch.isalpha())
        return h > l
    def _korean_lock(lock):
        lock = str(lock or "")
        return bool(lock) and ("한국" in lock or lock.lower().startswith("ko"))
    viol = []; expanded = 0
    for r in rrows:
        lr = r.get("ledger_row") or {}; acq = r.get("acquisition") or {}
        lock = str(((lr.get("grounding_producer_payload") or {}).get("language_lock_native")) or "")
        if not _korean_lock(lock):
            continue
        for rd in acq.get("rounds") or []:
            # ★우리가 만든 질의(directive · terms)만 잠금의 대상이다 — provider 가 스스로 더한 질의(`queries`)는
            #  관측값이지 우리 것이 아니다(라운드 계약 `query_provenance`). 그것은 따로 센다.
            ours = [rd.get("directive_native") or ""] + [str(x) for x in (rd.get("terms_native") or ())]
            for text in ours:
                if text and not _has_hangul(text):
                    viol.append(f"{lr.get('final_id')} r{rd.get('round_no')}: {str(text)[:40]}")
            for q in rd.get("queries") or []:
                for text in (q if isinstance(q, list) else [q]):
                    text = text.get("text") if isinstance(text, dict) else text
                    if text and not _has_hangul(text):
                        expanded += 1
    rows.append(("(감사·비차단) 언어 잠금 어긴 질의(우리 것)", True,
                 f"{len(viol)}건 {viol[:3]} · provider 확장 질의 중 비한국어 {expanded}건(관측)"))
    import collections as _c
    per_fid = _c.Counter((r.get("ledger_row") or {}).get("final_id") for r in out_rows if (r.get("ledger_row") or {}).get("final_id"))
    rows.append(("outlook final_id 당 조사 행 1 (obligation_kind=outfit)", bool(per_fid) and all(v == 1 for v in per_fid.values()) and all((r.get("ledger_row") or {}).get("obligation_kind") == "outfit" for r in out_rows if (r.get("ledger_row") or {}).get("final_id")), f"{dict(per_fid)}"))
    rows.append(("outlook 조사 대상 (final_id 있음)", all((r.get("ledger_row") or {}).get("final_id") for r in out_rows) and bool(out_rows), str([((r.get("ledger_row") or {}).get("final_id"), r.get("outcome")) for r in out_rows])))
else:
    rows.append(("조사 스텝 CP", False, "없음"))
from tools.grounding_audit import location_identity_check as lic
got = lic.check(RID, FIXTURE)
rows.append(("location_identity_check ok", bool(got.get("ok")), json.dumps({k: got.get(k) for k in ("scene_location", "primary_location", "part_parent", "cross_location_refs", "same_place_split")}, ensure_ascii=False)))
for f in got.get("findings") or []:
    rows.append(("  ✗ finding", False, f))
overall_ok = all(b for _, b, _ in rows)
(root / "research_acceptance.json").write_text(json.dumps({"overall_ok": overall_ok, "rows": [{"check": a, "ok": b, "detail": c} for a, b, c in rows], "checker": got}, ensure_ascii=False, indent=2), encoding="utf-8")
for a, b, c in rows:
    print(("✅" if b else "❌"), a, "—", c[:300])
print("적었다:", root / "research_acceptance.json")
print("overall_ok:", overall_ok)
sys.exit(0 if overall_ok else 1)