#!/usr/bin/env python3
"""task-2943 실증권 골든 검증 하네스 (dev2/헤임달 QA).

★ 위치 규칙: 이 파일과 그 출력물은 `/home/jay/workspace/teams/dev2/task-2943/`
아래에만 둔다. InsuRo repo(worktree)에는 어떤 파일도 만들거나 수정하지 않는다 —
worktree 의 policy_extract/policy_normalizer 는 import 해서 읽기만 한다.

실행:
    python3 verify_golden.py            # textlayer 파트만(빠름, 기본)
    python3 verify_golden.py --vision   # + 교보생명/우체국 vision 실호출(느림, 20~120s/건)

산출: stdout 요약 + 이 디렉터리에 마스킹된 JSON 스냅샷(dump_*.json).
"""

from __future__ import annotations

import argparse
import asyncio
import json
import os
import re
import sys
from collections import Counter, defaultdict
from dataclasses import asdict, is_dataclass

WORKTREE_SERVER = "/home/jay/projects/InsuRo/.worktrees/task-2943-dev2/server"
sys.path.insert(0, WORKTREE_SERVER)

import fitz  # noqa: E402

from policy_extract.textlayer import parse_textlayer  # noqa: E402
from policy_extract.detect import detect_format  # noqa: E402
from policy_extract.vision import AnuVisionClient  # noqa: E402
from policy_normalizer import normalize, COARSE_ORDER, UNCLASSIFIED  # noqa: E402

REAL_DIR = "/home/jay/.cokacdir/workspace/autoset/강연우가족증권"
GOLDEN_PDF = "/home/jay/.cokacdir/workspace/autoset/강연우(아들) 가입현황 260115.pdf"
OUT_DIR = os.path.dirname(os.path.abspath(__file__))

FILES = {
    "hyundai": os.path.join(REAL_DIR, "강연우-현대해상 어린이보험.pdf"),
    "meritz": os.path.join(REAL_DIR, "강혁-메리츠화재 담보내용.pdf"),
    "kyobo": os.path.join(REAL_DIR, "강혁-교보생명 보험증권.pdf"),
    "post_jpg1": os.path.join(REAL_DIR, "이고은-우체국암보험_01.JPG"),
    "post_jpg2": os.path.join(REAL_DIR, "이고은-우체국암보험_02.JPG"),
}

COARSE_LABELS = set(COARSE_ORDER)


def _to_jsonable(obj):
    """dataclass -> dict, masking any leftover unmasked-looking fields defensively."""
    if is_dataclass(obj) and not isinstance(obj, type):
        return {k: _to_jsonable(v) for k, v in asdict(obj).items()}
    if isinstance(obj, list):
        return [_to_jsonable(v) for v in obj]
    if isinstance(obj, dict):
        return {k: _to_jsonable(v) for k, v in obj.items()}
    return obj


# ===========================================================================
# (1) 정답 시트 파싱 — bbox 좌표 기반 (평문 get_text() 순서 사용 금지)
# ===========================================================================


def parse_golden_sheet(pdf_path: str) -> list[dict]:
    """정답 시트 1p PDF를 열-구조로 파싱한다.

    열: 대구분(좌측 라벨, 첫 행에만/병합되어 나타남) | 담보명 | 권장가입금액합계 |
    현대해상종합보험 | 현대해상종합+실손(착한실손).

    ★ 방법론: get_text("dict") bbox 의 y0 를 촘촘한 허용오차(0.3pt)로 클러스터링해
    "같은 시각적 행"을 구성한다. 정답 시트 우측에는 표와 무관한 주석(예: "<80세>
    뇌혈관...", "호흡기(50), 탈장(20)...")이 x0 상으로는 값 컬럼과 겹치는 위치에
    떠 있는데, 실측 확인 결과 y0 가 데이터 행과 정확히 일치하지 않는 경우가
    대부분이라 0.3pt 허용오차로 대부분 분리된다. 단 극소수(예: "뇌/심 수술비" 행의
    긴 주석)는 y0 가 완전히 같아 분리가 안 되므로, x0 >= 310 인 셀은 항상 주석으로
    간주해 무시한다(값 컬럼 실측 최대 x0 는 304.9 로 확인됨).
    """
    doc = fitz.open(pdf_path)
    page = doc[0]
    d = page.get_text("dict")
    lines = []
    for block in d["blocks"]:
        for line in block.get("lines", []):
            spans = line.get("spans", [])
            text = "".join(s.get("text", "") for s in spans).strip()
            if not text:
                continue
            bbox = line["bbox"]
            lines.append({"y0": bbox[1], "x0": bbox[0], "text": text})
    doc.close()

    lines.sort(key=lambda l: (round(l["y0"], 1), l["x0"]))
    rows: list[list[dict]] = []
    cur: list[dict] = []
    ref_y = None
    tol = 0.3
    for ln in lines:
        if ref_y is None or abs(ln["y0"] - ref_y) <= tol:
            cur.append(ln)
            if ref_y is None:
                ref_y = ln["y0"]
        else:
            rows.append(cur)
            cur = [ln]
            ref_y = ln["y0"]
    if cur:
        rows.append(cur)
    for r in rows:
        r.sort(key=lambda l: l["x0"])

    def bucket(x0: float) -> str:
        if x0 < 165:
            return "name"
        if x0 < 205:
            return "recommended"
        if x0 < 250:
            return "hyundai_comprehensive"
        if x0 < 310:
            return "hyundai_comp_realloss"
        return "ignore"  # 실측: 우측 자유주석(annotation) 영역, 표 데이터 아님

    current_coarse = None
    parsed_rows: list[dict] = []
    started = False
    ambiguous: list[str] = []
    for row in rows:
        texts = [c["text"] for c in row]
        if not started:
            if "주요 담보" in texts:
                started = True
            continue
        if row[0]["text"].startswith("(리모델링"):
            break  # 표 종료 — 이후는 자유 서술(코멘터리)

        leftmost = row[0]
        label_prefix = None
        merged_name = None
        if leftmost["x0"] < 90:
            for lbl in COARSE_LABELS:
                if leftmost["text"] == lbl:
                    label_prefix = lbl
                    break
                if leftmost["text"].startswith(lbl) and len(leftmost["text"]) > len(lbl):
                    label_prefix = lbl
                    merged_name = leftmost["text"][len(lbl):]
                    break
        if label_prefix:
            current_coarse = label_prefix

        name_cell_text = None
        for c in row:
            if c is leftmost and label_prefix:
                continue
            if 90 <= c["x0"] < 165:
                name_cell_text = c["text"]
                break
        name_text = merged_name if merged_name is not None else name_cell_text
        if name_text is None:
            continue  # 표 데이터 행이 아님(순수 주석 행)

        cells_by_bucket: dict[str, list[str]] = defaultdict(list)
        for c in row:
            if c is leftmost and label_prefix:
                continue
            if 90 <= c["x0"] < 165:
                continue
            b = bucket(c["x0"])
            if b != "ignore":
                cells_by_bucket[b].append(c["text"])

        value_cell_count = sum(len(v) for v in cells_by_bucket.values())
        note = ""
        recommended = cells_by_bucket.get("recommended", [None])[0]
        hyundai = cells_by_bucket.get("hyundai_comprehensive", [None])[0]
        hyundai_real = cells_by_bucket.get("hyundai_comp_realloss", [None])[0]
        # ★ "기타 수술비" 류: 표 안에 값 셀이 1개뿐이라 어느 열인지 확정 불가
        # (실측: "5~2,000" 범위 표기가 권장 컬럼 x0 범위를 넘어 col2 x0 대에
        # 찍힘). 추측 배정 금지 — 두 정책열 다 None 처리하고 감사 메모만 남긴다.
        if value_cell_count == 1 and hyundai is not None and recommended is None:
            note = f"ambiguous_single_cell(raw={hyundai!r}) — 열 확정 불가, 두 정책열 모두 매칭 제외"
            ambiguous.append(name_text)
            recommended, hyundai, hyundai_real = hyundai, None, None

        parsed_rows.append(
            {
                "coarse": current_coarse,
                "name": name_text,
                "recommended": recommended,
                "hyundai_comprehensive": hyundai,
                "hyundai_comp_realloss": hyundai_real,
                "note": note,
            }
        )
    return parsed_rows


_NUM_RE = re.compile(r"^[\d,]+(?:\.\d+)?$")


def _golden_cell_to_values(cell: str | None) -> list[float]:
    """정답 시트 셀 문자열을 숫자 값 리스트로. "-"/None="값없음"(빈 리스트).

    "25/5", "0/500" 같은 복합표기는 "/" 로 분리한 각 성분을 모두 반환한다
    (부분매치 허용, recall 측정 시 compound 로 별도 표기).
    """
    if cell is None or cell == "-":
        return []
    if "~" in cell:
        return []  # 범위표기 — 단일 값 매칭 대상 아님(별도 처리)
    parts = re.split(r"[/]", cell)
    values: list[float] = []
    for p in parts:
        p = p.strip()
        if _NUM_RE.match(p.replace(",", "")):
            try:
                values.append(float(p.replace(",", "")))
            except ValueError:
                pass
    return values


# ===========================================================================
# (2) 실증권 textlayer 추출 + 정규화
# ===========================================================================


def extract_and_normalize(path: str) -> list[dict]:
    result = parse_textlayer(path)
    rows = []
    for cov in result.coverages:
        norm = normalize(cov)
        rows.append(
            {
                "name": cov.name,
                "amount_manwon": cov.amount_manwon,
                "amount_raw": cov.amount_raw,
                "payout_condition": cov.payout_condition,
                "coarse": norm.coarse,
                "rule_id": norm.rule_id,
                "needs_advisor": norm.needs_advisor,
                "reason": norm.reason,
                "page": cov.provenance.page,
                "warnings": cov.warnings,
            }
        )
    return result, rows


# ===========================================================================
# (A) 대구분 집계 대조
# ===========================================================================


def section_a(golden_rows: list[dict], our_rows: list[dict]) -> dict:
    golden_counts = Counter()
    for r in golden_rows:
        if r["hyundai_comprehensive"] not in (None, "-"):
            golden_counts[r["coarse"]] += 1

    our_counts = Counter(r["coarse"] for r in our_rows)

    order_check = list(golden_counts.keys())
    # 정답시트에 등장한 대구분 순서가 실제로 나타난 순서 그대로 COARSE_ORDER 부분열인지 확인
    filtered_order = [c for c in COARSE_ORDER if c in order_check]
    golden_order_matches_coarse_order = order_check == [
        c for c in golden_counts.keys()
    ]  # 항상 True(dict는 insertion 순서 유지) — 실제 체크는 아래
    appearance_order = []
    seen = set()
    for r in golden_rows:
        if r["coarse"] and r["coarse"] not in seen:
            appearance_order.append(r["coarse"])
            seen.add(r["coarse"])
    order_is_fixed = appearance_order == list(COARSE_ORDER)

    table = []
    for c in COARSE_ORDER:
        table.append(
            {
                "coarse": c,
                "golden_count": golden_counts.get(c, 0),
                "our_count": our_counts.get(c, 0),
            }
        )
    table.append(
        {
            "coarse": UNCLASSIFIED,
            "golden_count": 0,
            "our_count": our_counts.get(UNCLASSIFIED, 0),
        }
    )
    return {
        "table": table,
        "golden_section_order": appearance_order,
        "golden_order_is_fixed_10": order_is_fixed,
    }


# ===========================================================================
# (B) 값 기준 recall (이름 무관)
# ===========================================================================


def section_b(golden_rows: list[dict], our_rows: list[dict]) -> dict:
    our_values_by_coarse: dict[str, list[float]] = defaultdict(list)
    for r in our_rows:
        if r["amount_manwon"] is not None:
            our_values_by_coarse[r["coarse"]].append(r["amount_manwon"])

    matched = []
    unmatched = []
    compound_notes = []
    for r in golden_rows:
        cell = r["hyundai_comprehensive"]
        if cell in (None, "-"):
            continue
        if "~" in (cell or ""):
            compound_notes.append(
                {"coarse": r["coarse"], "name": r["name"], "cell": cell, "kind": "range_excluded"}
            )
            continue
        values = _golden_cell_to_values(cell)
        is_compound = "/" in (cell or "")
        pool = our_values_by_coarse.get(r["coarse"], [])
        for v in values:
            hit = v in pool
            entry = {
                "coarse": r["coarse"],
                "name": r["name"],
                "golden_value": v,
                "cell_raw": cell,
                "compound": is_compound,
            }
            if hit:
                matched.append(entry)
            else:
                unmatched.append(entry)

    total = len(matched) + len(unmatched)
    return {
        "matched": matched,
        "unmatched": unmatched,
        "compound_notes": compound_notes,
        "total": total,
        "matched_count": len(matched),
        "recall_pct": (len(matched) / total * 100.0) if total else 0.0,
    }


# ===========================================================================
# (C) 감사용 수동 대조표 — 자동판정 아님, 설계사 확인 큐 성격
# ===========================================================================


def section_c(golden_rows: list[dict], our_rows: list[dict]) -> list[dict]:
    our_by_coarse: dict[str, list[dict]] = defaultdict(list)
    for r in our_rows:
        our_by_coarse[r["coarse"]].append(r)

    audit = []
    for r in golden_rows:
        cell = r["hyundai_comprehensive"]
        if cell in (None, "-"):
            continue
        candidates = our_by_coarse.get(r["coarse"], [])
        golden_values = set(_golden_cell_to_values(cell)) if "~" not in (cell or "") else set()
        cand_list = [
            {"name": c["name"], "amount_manwon": c["amount_manwon"], "amount_match": c["amount_manwon"] in golden_values}
            for c in candidates
        ]
        audit.append(
            {
                "coarse": r["coarse"],
                "golden_name": r["name"],
                "golden_value_raw": cell,
                "note": r.get("note", ""),
                "candidates": cand_list,
                "any_amount_match": any(c["amount_match"] for c in cand_list),
            }
        )
    return audit


# ===========================================================================
# UNKNOWN 정책 실측 점검
# ===========================================================================


def check_unknown_policy(all_extracted: dict[str, list[dict]]) -> dict:
    findings = []
    for label, rows in all_extracted.items():
        for r in rows:
            # amount_manwon 이 0.0 인데 amount_raw 가 명시적 "0"이 아니면 의심스러운 0-fill
            if r["amount_manwon"] == 0.0 and "0" not in (r.get("amount_raw") or ""):
                findings.append(
                    {"file": label, "name": r["name"], "issue": "amount=0.0 without explicit '0' in raw", "amount_raw": r.get("amount_raw")}
                )
    return {"violations": findings, "violation_count": len(findings)}


# ===========================================================================
# vision (실 CLI 호출) — 교보생명(4p image PDF), 우체국(2 JPG)
# ===========================================================================


async def run_vision_kyobo() -> dict:
    client = AnuVisionClient()
    result = await client.extract(FILES["kyobo"])
    return _to_jsonable(result)


async def run_vision_post() -> dict:
    client = AnuVisionClient()
    result = await client.extract_pages([FILES["post_jpg1"], FILES["post_jpg2"]])
    return _to_jsonable(result)


# ===========================================================================
# main
# ===========================================================================


def main() -> None:
    parser = argparse.ArgumentParser()
    parser.add_argument("--vision", action="store_true", help="교보생명/우체국 실 CLI vision 호출도 실행")
    args = parser.parse_args()

    print("=" * 70)
    print("[1] 정답 시트 파싱 (bbox 좌표 기반)")
    print("=" * 70)
    golden_rows = parse_golden_sheet(GOLDEN_PDF)
    print(f"정답 시트 데이터 행 수: {len(golden_rows)}")
    appearance_order = []
    seen = set()
    for r in golden_rows:
        if r["coarse"] and r["coarse"] not in seen:
            appearance_order.append(r["coarse"])
            seen.add(r["coarse"])
    print(f"대구분 등장 순서: {appearance_order}")
    print(f"COARSE_ORDER 와 일치: {appearance_order == list(COARSE_ORDER)}")
    with open(os.path.join(OUT_DIR, "dump_golden_rows.json"), "w", encoding="utf-8") as f:
        json.dump(golden_rows, f, ensure_ascii=False, indent=2)

    print()
    print("=" * 70)
    print("[2] 현대해상 어린이보험 textlayer 추출 + 정규화")
    print("=" * 70)
    hyundai_result, hyundai_rows = extract_and_normalize(FILES["hyundai"])
    print(f"method={hyundai_result.method} page_count={hyundai_result.page_count} n_coverages={len(hyundai_result.coverages)}")
    print(f"top-level warnings: {hyundai_result.warnings}")
    print(f"header: {hyundai_result.header}")
    with open(os.path.join(OUT_DIR, "dump_hyundai_extracted.json"), "w", encoding="utf-8") as f:
        json.dump({"header": _to_jsonable(hyundai_result.header), "rows": hyundai_rows, "top_level_warnings": hyundai_result.warnings}, f, ensure_ascii=False, indent=2)

    print()
    print("=" * 70)
    print("[3] 메리츠화재 담보내용 textlayer 추출 (정답시트 없음 — 형식검증용)")
    print("=" * 70)
    meritz_result, meritz_rows = extract_and_normalize(FILES["meritz"])
    print(f"method={meritz_result.method} page_count={meritz_result.page_count} n_coverages={len(meritz_result.coverages)}")
    print(f"header: {meritz_result.header}")
    print(f"warnings: {meritz_result.warnings}")
    with open(os.path.join(OUT_DIR, "dump_meritz_extracted.json"), "w", encoding="utf-8") as f:
        json.dump({"header": _to_jsonable(meritz_result.header), "rows": meritz_rows, "top_level_warnings": meritz_result.warnings}, f, ensure_ascii=False, indent=2)

    print()
    print("=" * 70)
    print("(A) 대구분 집계 대조")
    print("=" * 70)
    a_result = section_a(golden_rows, hyundai_rows)
    print(f"{'coarse':10s} {'golden':>8s} {'ours':>8s}")
    for row in a_result["table"]:
        print(f"{row['coarse']:10s} {row['golden_count']:>8d} {row['our_count']:>8d}")
    print(f"golden 대구분 등장순서 == COARSE_ORDER 고정순서: {a_result['golden_order_is_fixed_10']}")

    print()
    print("=" * 70)
    print("(B) 값 기준 recall (이름 무관)")
    print("=" * 70)
    b_result = section_b(golden_rows, hyundai_rows)
    print(f"일치 {b_result['matched_count']} / 정답 {b_result['total']} = {b_result['recall_pct']:.1f}%")
    print("미매칭 목록:")
    for u in b_result["unmatched"]:
        print(f"  - [{u['coarse']}] {u['name']} golden={u['golden_value']} (raw={u['cell_raw']!r}, compound={u['compound']})")
    print(f"범위표기(제외): {len(b_result['compound_notes'])}건")
    for c in b_result["compound_notes"]:
        print(f"  - [{c['coarse']}] {c['name']} raw={c['cell']!r}")

    print()
    print("=" * 70)
    print("(C) 감사용 수동 대조표 (자동판정 아님 — 설계사 확인 큐)")
    print("=" * 70)
    c_result = section_c(golden_rows, hyundai_rows)
    for row in c_result:
        mark = "OK" if row["any_amount_match"] else "??"
        print(f"[{mark}] [{row['coarse']}] 정답='{row['golden_name']}'({row['golden_value_raw']}) note={row['note']}")
        for cand in row["candidates"]:
            flag = "MATCH" if cand["amount_match"] else "     "
            print(f"       {flag} 후보='{cand['name']}' amount={cand['amount_manwon']}")
    with open(os.path.join(OUT_DIR, "dump_audit_table.json"), "w", encoding="utf-8") as f:
        json.dump(c_result, f, ensure_ascii=False, indent=2)

    print()
    print("=" * 70)
    print("UNKNOWN 정책 실측 점검 (0-fill 위반 탐지)")
    print("=" * 70)
    unknown_check = check_unknown_policy({"hyundai": hyundai_rows, "meritz": meritz_rows})
    print(f"위반 건수: {unknown_check['violation_count']}")
    for v in unknown_check["violations"]:
        print(f"  - {v}")

    if args.vision:
        print()
        print("=" * 70)
        print("[4] vision 실 CLI 호출 — 교보생명(4p image PDF)")
        print("=" * 70)
        kyobo_json = asyncio.run(run_vision_kyobo())
        print(json.dumps({k: v for k, v in kyobo_json.items() if k != "coverages"}, ensure_ascii=False, indent=2))
        print(f"n_coverages={len(kyobo_json.get('coverages', []))} warnings={kyobo_json.get('warnings')}")
        with open(os.path.join(OUT_DIR, "dump_kyobo_vision.json"), "w", encoding="utf-8") as f:
            json.dump(kyobo_json, f, ensure_ascii=False, indent=2)

        print()
        print("=" * 70)
        print("[5] vision 실 CLI 호출 — 우체국암보험 (JPG 2장, 한 증권)")
        print("=" * 70)
        post_json = asyncio.run(run_vision_post())
        print(json.dumps({k: v for k, v in post_json.items() if k != "coverages"}, ensure_ascii=False, indent=2))
        print(f"n_coverages={len(post_json.get('coverages', []))} warnings={post_json.get('warnings')}")
        with open(os.path.join(OUT_DIR, "dump_post_vision.json"), "w", encoding="utf-8") as f:
            json.dump(post_json, f, ensure_ascii=False, indent=2)

    print()
    print("done.")


if __name__ == "__main__":
    main()