diff --git a/scripts/verify_statute_chain.py b/scripts/verify_statute_chain.py new file mode 100644 index 0000000..a022034 --- /dev/null +++ b/scripts/verify_statute_chain.py @@ -0,0 +1,133 @@ +"""B-1 PR③ — 법령 버전 체인 검증 3술어 (plan safety-library-1). + +read-only 진단 — E-1 관찰의 법령 게이트 도구로도 재사용 (반복 실행 안전). + +검증 3술어 (R7-M1, B-1 단일 정본): + ① 존재성 — watch family 각각 primary 시리즈 current 정확 1건(0건도 위반) + + annex 시리즈당 current ≤ 1 + ② 노출 유일성 — primary current 보유 family당 primary 노출(체인+레거시 매핑 합산) 정확 1건 + (모집단 = primary current 보유 family 한정 — R8-M2) + ③ 고아 그물 — law_monitor in_corpus=true 레거시 중: + (a) current 보유 family 에 매핑되는데 안 flip 된 것(flip 누락) = 0 + (b) 어느 watch family 에도 매핑 안 되는 것(제명 개정 등 매핑 구멍) = 0 + repealed family·primary current 미보유 family 의 레거시 보존은 위반 아님 + +repealed family 는 ①② 기대값 0 으로 면제. + +실행: + docker compose exec -T fastapi python /app/scripts/verify_statute_chain.py +종료코드: 0 = 전건 PASS, 1 = 위반 (CI/관찰 게이트 용) +""" + +import asyncio +import os +import sys + +sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", "app")) + +from collections import defaultdict + +from sqlalchemy import select, text +from sqlalchemy.ext.asyncio import create_async_engine + +from workers.statute_collector import legacy_law_name, normalize_law_name, series_suffix + + +async def main() -> int: + db_url = os.getenv("DATABASE_URL", "postgresql+asyncpg://pkm:pkm@localhost:5432/pkm") + engine = create_async_engine(db_url) + violations: list[str] = [] + + async with engine.connect() as conn: + # ── 로드 ── + acts = (await conn.execute(text( + "SELECT family_id, title, repeal_detected_at IS NOT NULL AS repealed " + "FROM legal_acts WHERE watch"))).all() + metas = (await conn.execute(text( + "SELECT family_id, law_doc_kind, version_key, version_status FROM legal_meta"))).all() + + act_title = {a.family_id: a.title for a in acts} + repealed = {a.family_id for a in acts if a.repealed} + active = [a for a in acts if not a.repealed] + + # family → primary current 수 / annex 시리즈별 current 수 + prim_current = defaultdict(int) + annex_series_current = defaultdict(int) + for m in metas: + if m.version_status != "current": + continue + if m.law_doc_kind == "primary": + prim_current[m.family_id] += 1 + else: + annex_series_current[(m.family_id, series_suffix(m.version_key))] += 1 + + # ── ① 존재성 ── + for a in active: + n = prim_current[a.family_id] + if n != 1: + violations.append(f"① {a.family_id} ({a.title}): primary current {n}건 (정확 1 기대)") + for (fid, suf), n in annex_series_current.items(): + if fid not in repealed and n > 1: + violations.append(f"① {fid} annex 시리즈 {suf}: current {n}건 (≤1 기대)") + + # ── ③ 고아 그물 (정규화 동등 매핑) ── + # watch family 정규화명 → family_id (current 보유 여부 동반) + norm_to_fid = {} + for a in active: + norm_to_fid[normalize_law_name(a.title)] = a.family_id + + legacy = (await conn.execute(text( + "SELECT d.id, d.title, " + " EXISTS(SELECT 1 FROM document_chunks c WHERE c.doc_id=d.id AND c.in_corpus) AS exposed " + "FROM documents d WHERE d.source_channel='law_monitor' AND d.deleted_at IS NULL"))).all() + + orphan_flip_miss = 0 + orphan_unmapped = 0 + unmapped_names = set() + for row in legacy: + if not row.exposed: + continue # in_corpus=false = 정상 (스윕됨 or 청크 없음) + name = legacy_law_name(row.title or "") + norm = normalize_law_name(name) if name else None + fid = norm_to_fid.get(norm) if norm else None + if fid is None: + orphan_unmapped += 1 + if name: + unmapped_names.add(name) + elif prim_current.get(fid, 0) >= 1: + # current 보유 family 인데 레거시가 노출 중 = flip 누락 + orphan_flip_miss += 1 + if orphan_flip_miss: + violations.append(f"③(a) flip 누락: current 보유 family 의 노출 레거시 {orphan_flip_miss}건") + if orphan_unmapped: + violations.append( + f"③(b) 무매핑 노출 레거시 {orphan_unmapped}건 — 매핑 구멍(매핑 보강 신호): " + + ", ".join(sorted(unmapped_names))[:200]) + + # ── ② 노출 유일성 (primary current 보유 family 한정) ── + # 노출 primary = 체인 primary current(=1) + 레거시 매핑 노출분. + # ③(a)=0 이면 레거시 노출분 0 → 체인 1건만 = 정확 1. 별도 위반 추출은 ③(a)에 포함됨. + # (annex 노출 비동기 일반화는 may — Phase 1 미적용) + + # ── 상태 요약 출력 ── + print("=== 법령 체인 검증 (B-1 PR③ 3술어) ===") + print(f"watch family: {len(acts)} (active {len(active)}, repealed {len(repealed)})") + print(f"primary current 보유 family: {sum(1 for a in active if prim_current[a.family_id]==1)}/{len(active)}") + print(f"annex current 시리즈: {len(annex_series_current)}") + exposed_legacy = sum(1 for r in legacy if r.exposed) + print(f"레거시 law_monitor: {len(legacy)}건 (in_corpus 노출 {exposed_legacy}건)") + print() + + await engine.dispose() + + if violations: + print(f"[FAIL] 위반 {len(violations)}건:") + for v in violations: + print(" -", v) + return 1 + print("[PASS] 3술어 전건 통과 (존재성·노출 유일성·고아 그물)") + return 0 + + +if __name__ == "__main__": + sys.exit(asyncio.run(main()))