#!/usr/bin/env python3
"""kit/pipeline 검증: yaml·json 파싱 · roster(77노드·113배선·반려 15) 대조 · JSON Schema 유효성 · state.py↔schema 정합 · graph.py --dry-run
사용: python3 verify.py [--roster <PROJECT_ROOT>/설명/05_배선도/roster.json]
roster 경로를 안 주면 ../../설명/05_배선도/roster.json(프로젝트 안 키트일 때)을 찾고, 없으면 roster 대조는 pipeline.yaml counts로만 한다.
"""
from __future__ import annotations

import argparse
import json
import re
import subprocess
import sys
from pathlib import Path

HERE = Path(__file__).resolve().parent
sys.path.insert(0, str(HERE))
FAIL: list[str] = []


def ok(cond: bool, msg: str) -> None:
    print(("PASS " if cond else "FAIL ") + msg)
    if not cond:
        FAIL.append(msg)


EXAMPLE_STATE = {  # 실제 값(1편 S4 한 컷 경로) — 03_파이프라인_그래프.md ④ 시퀀스
    "brief": {"reference": {"file": "00_ref/ref_DeKsEh0sBPy.mp4", "seconds": 10}, "meme_research": "00_ref/research-meme.md",
              "flow_research": "00_ref/research-flow.md", "pinterest": {"notes": "00_ref/pinterest/NOTES.md", "count": 122}},
    "plan": {"episodes": {"V1": {"id": "V1", "title": "S-RANK PULL", "duration_s": 26.5, "bpm": 128.17, "next": "V2"}},
             "storyboard": {"V1": "STORYBOARD.md"}, "edl": {"V1": "V1-EDL.md"},
             "beat_grid": {"V1": {"bpm": 128, "drop_s": 7.5, "source": "plan"}},
             "flow_jobs": {"flow_b_v1": [{"id": "V1_S4", "first_frame": "02_frames/S4_fullbody.png", "seconds": 8}]}},
    "rules": [{"id": "R-I01", "text": "텍스트만 영상 생성 금지, 모든 컷 첫 프레임 지정", "source": "I01"}],
    "assets": {
        "images": {"S4_fullbody": {"id": "S4_fullbody", "batch": "img_frames", "size": "1024x1792", "final_file": "02_frames/S4_fullbody.png", "status": "ok"}},
        "clips": {"V1_S4": {"id": "V1_S4", "batch": "flow_b_v1", "mode": "first_frame", "first_frame": "02_frames/S4_fullbody.png", "seconds": 8,
                            "resolution": "720p", "aspect": "9:16", "result_file": "06_flow/v1/V1_S4.mp4", "credits": 12, "verdict": "채택",
                            "regen_count": 0, "blue_bg": True}},
        "mattes": {"V1_S4": {"clip_id": "V1_S4", "file": "05_matte/v1/V1_S4.webm", "qc": "05_matte/v1/V1_S4_qc.jpg", "backend": "hybrid", "status": "done"}},
    },
    "audio": {"tracks": {"BGM_MAIN": {"id": "BGM_MAIN", "episode": "V1", "file": "04_audio/bgm/BGM_MAIN_aligned.wav", "bpm": 128.17, "trim_from_raw_s": 0.92, "chosen": True, "lang": "instrumental"}},
              "beats": {"V1": {"file": "04_audio/beats.json", "bpm": 128.17, "beat_period_s": 0.46813, "drop_s": 7.484,
                               "big_hits_s": [3.773, 7.484, 11.233, 14.99, 18.734, 22.487], "source": "measured"}}},
    "blocks": {"kinetic_wall": {"id": "kinetic_wall", "file": "03_motion/compositions/kinetic_wall.html", "worker": "B", "vars": {"layer": "bg"}, "lint": "pass"}},
    "renders": {"V1": {"episode": "V1", "final": "07_render/V1_SRANK_PULL.mp4", "contact": "07_render/V1_contact.jpg", "duration_s": 26.5, "lufs": -14, "status": "approved", "reviewed": True}},
    "qa": {"verdicts": [{"node": "flow_b_v1", "target": "V1_S4", "verdict": "pass", "reason": "원테이크·블루 유지", "round": 1}],
           "retries": {"img_s6": 1}, "gates": {"flow_b_v1": {"status": "pass", "round": 1, "retry_items": []}}},
    "feedback": [{"id": "I08", "quote": "왜 아래쪽에 악역영애 말포이 좀 없애봐", "target_node": "c_v2_fix", "status": "resolved", "effect": "2편 하단 자막 제거 재합성"}],
    "budget": {"flow_credits": {"start": 1050, "spent": 386, "entries": [{"node": "flow_b_v1", "amount": 88}]},
               "eleven_credits": {"spent": 19475}, "codex": {"spent": 0}},
    "runtime": {"slots": {"heavy": 2, "flow": 5, "eleven": 2, "codex": 4, "matte_lines": 2, "subagents": 8},
                "heavy": {"limit": 2, "in_use": 0, "script": "_tools/heavy.sh"}, "toolchain": {"matte": "05_matte/matte.sh"},
                "incidents": [{"ts": "10-07 20:08", "what": "과부하로 WSL 재부팅", "fix": "heavy.sh 2슬롯"}], "resume_log": [], "trace": ["ref", "sb"]},
    "docs": {"pipe_merge": {"id": "pipe_merge", "path": "설명/04_도구별_파이프라인.md", "kind": "merged"}},
}

SECRET_PAT = re.compile(r"(sk-[A-Za-z0-9]{16,}|xi-api-key\s*[:=]\s*\S{8,}|[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[a-z]{2,}|AIza[0-9A-Za-z_-]{20,}|Bearer\s+[A-Za-z0-9._-]{20,})")


def main() -> int:
    ap = argparse.ArgumentParser()
    ap.add_argument("--roster", default=str(HERE.parent.parent / "설명/05_배선도/roster.json"))
    a = ap.parse_args()

    import yaml
    P = yaml.safe_load((HERE / "pipeline.yaml").read_text(encoding="utf-8"))
    ok(isinstance(P, dict), "pipeline.yaml 파싱")
    schema = json.loads((HERE / "state_schema.json").read_text(encoding="utf-8"))
    ok(isinstance(schema, dict), "state_schema.json 파싱")

    # --- roster 대조
    ids = [n["id"] for n in P["nodes"]]
    ok(len(ids) == len(set(ids)), f"노드 id 중복 없음 ({len(ids)})")
    rp = Path(a.roster)
    if rp.exists():
        R = json.loads(rp.read_text(encoding="utf-8"))
        rnodes = {n["id"]: st["id"] for st in R["stages"] for n in st["nodes"]}
        ok(len(rnodes) == 77, f"roster 노드 77 (실제 {len(rnodes)})")
        ok(set(ids) == set(rnodes), f"roster 노드 전부 pipeline.yaml에 있음 (누락 {sorted(set(rnodes) - set(ids))}, 초과 {sorted(set(ids) - set(rnodes))})")
        ok(all(n["stage"] == rnodes[n["id"]] for n in P["nodes"]), "노드 스테이지 일치")
        ok(len(R["edges"]) == 113 and len(P["edges"]) == 113, f"배선 113 (roster {len(R['edges'])}, yaml {len(P['edges'])})")
        ok([tuple(e) for e in P["edges"]] == [tuple(e) for e in R["edges"]], "배선 113 순서·내용 일치")
        ok(len(R["interrupts"]) == 15 and len(P["interrupts"]) == 15, f"반려 15 (roster {len(R['interrupts'])}, yaml {len(P['interrupts'])})")
        ok([(i["quote"], i["to"]) for i in P["interrupts"]] == [(i["quote"], i["to"]) for i in R["interrupts"]], "반려 quote·to 일치")
    else:
        print(f"SKIP roster 없음({rp}) → counts로 대조")
        ok(P["counts"] == {"nodes": 77, "edges": 113, "interrupts": 15}, "counts 77·113·15")
    idset = set(ids)
    ok(all(a_ in idset and b in idset for a_, b in P["edges"]), "배선 양끝이 노드에 있음")
    ok(all(i["to"] in idset for i in P["interrupts"]), "반려 되돌림 대상이 노드에 있음")

    # --- JSON Schema
    try:
        from jsonschema import Draft202012Validator
        Draft202012Validator.check_schema(schema)
        ok(True, "JSON Schema draft 2020-12 메타 검사")
        v = Draft202012Validator(schema)
        errs = list(v.iter_errors(EXAMPLE_STATE))
        ok(not errs, "예시 State(1편 S4 실제 값) 유효" + ("" if not errs else f": {errs[0].message} @ {list(errs[0].path)}"))
        from state import initial_state
        errs = list(v.iter_errors(initial_state()))
        ok(not errs, "state.initial_state() 유효" + ("" if not errs else f": {errs[0].message}"))
        bad = dict(EXAMPLE_STATE, assets={"clips": {"x": {"id": "x", "mode": "text_only"}}})
        ok(any(True for _ in v.iter_errors(bad)), "음성 시험: mode=text_only(텍스트만 Flow) 거부")
    except ImportError:
        print("SKIP jsonschema 미설치 → 메타 검사 생략")

    # --- reads/writes 키 경로가 schema에 있나
    props = schema["properties"]
    def key_ok(k: str) -> bool:
        top, _, sub = k.partition(".")
        if top not in props:
            return False
        return not sub or sub in (props[top].get("properties") or {}) or "additionalProperties" in props[top] and props[top]["additionalProperties"] is not False
    badk = sorted({k for n in P["nodes"] for k in n["reads"] + n["writes"] if not key_ok(k)})
    ok(not badk, f"노드 reads/writes 키가 schema에 있음 (불일치 {badk})")
    no_w = [n["id"] for n in P["nodes"] if not n["writes"]]
    ok(not no_w, f"모든 노드가 쓰는 키 1개 이상 (없음 {no_w})")

    # --- state.py ↔ schema
    from state import MalfoyState, REDUCER_OF
    ok(set(MalfoyState.__annotations__) == set(props), f"state.py 키 = schema 키 ({len(props)})")
    mism = [k for k in props if props[k]["x-reducer"]["name"] != REDUCER_OF[k].__name__]
    ok(not mism, f"reducer 이름 일치 (불일치 {mism})")
    # reducer 동작 표본
    from state import merge_budget, merge_qa, upsert_by_id
    b = merge_budget({"flow_credits": {"spent": 10}}, {"flow_credits": {"spent": 12}})
    ok(b["flow_credits"]["spent"] == 22, "merge_budget: 병렬 spent 합산")
    q = merge_qa({"retries": {"c_v4": 1}, "verdicts": [1]}, {"retries": {"c_v4": 1}, "verdicts": [2]})
    ok(q["retries"]["c_v4"] == 2 and q["verdicts"] == [1, 2], "merge_qa: retries 합산·verdicts 누적")
    f = upsert_by_id([{"id": "I08", "status": "open"}], [{"id": "I08", "status": "resolved"}])
    ok(len(f) == 1 and f[0]["status"] == "resolved", "upsert_by_id: feedback status 갱신")

    # --- 그래프 구조
    from graph import build_structure
    S = build_structure(P)
    ok(all(i in S["gnodes"] for i in ids), "graph 구조에 roster 77노드 전부")
    ok(len(S["realized"]) == 113, f"graph 배선 실현 {len(S['realized'])}/113")
    gates = [n["id"] for n in P["nodes"] if "gate" in n]
    fan = [n["id"] for n in P["nodes"] if "fanout" in n]
    ok("flow_test" in gates and any(c["router"] == "gate" and c["src"] == "flow_test" for c in S["cond"]), "테스트 게이트(flow_test) 조건 엣지")
    ok(len(fan) == 9 and len(S["groups"]) == 5, f"Send fan-out: 배치 {len(fan)}(이미지 4·Flow 5) + 섹션 그룹 {len(S['groups'])}")
    r = subprocess.run([sys.executable, str(HERE / "graph.py"), "--dry-run"], capture_output=True, text=True, timeout=120)
    ok(r.returncode == 0 and "누락 0" in r.stdout and "113/113" in r.stdout, "graph.py --dry-run 종료 0·누락 0·113/113")

    # --- 비밀값
    hits = []
    for p in HERE.iterdir():
        if p.suffix in (".py", ".yaml", ".json", ".md") and p.name != "verify.py":
            hits += [f"{p.name}: {m.group(0)[:12]}…" for m in SECRET_PAT.finditer(p.read_text(encoding="utf-8"))]
    ok(not hits, f"비밀값·이메일 패턴 0 ({hits[:3]})")

    print(f"\n결과: {'전부 PASS' if not FAIL else f'FAIL {len(FAIL)}건'}")
    return 1 if FAIL else 0


if __name__ == "__main__":
    sys.exit(main())