QuantEngineByItz/tools/build_calibration_approval_list_v1.py

#!/usr/bin/env python3
"""
build_calibration_approval_list_v1.py
───────────────────────────────────────────────────────────────────────────────
calibration_review_report_v1.json을 읽어 PROVISIONAL 승격 승인 리스트를 만든다.

목적:
  - source=PROVISIONAL 인 임계값을 별도 승인 대상 리스트로 분리
  - reviewer가 바로 볼 수 있는 Markdown/JSON 산출물 생성
  - PROVISIONAL 승격과 provisional review를 분리해 운영 책임을 명확화

출력:
  Temp/calibration_approval_list_v1.json
  Temp/calibration_approval_list_v1.md

사용법:
  python tools/build_calibration_approval_list_v1.py
"""

from __future__ import annotations

import json
import sys
from datetime import datetime, timezone
from pathlib import Path
from typing import Any

ROOT = Path(__file__).resolve().parent.parent
REVIEW = ROOT / "Temp" / "calibration_review_report_v1.json"
OUT_JSON = ROOT / "Temp" / "calibration_approval_list_v1.json"
OUT_MD = ROOT / "Temp" / "calibration_approval_list_v1.md"

if sys.stdout.encoding and sys.stdout.encoding.lower() not in ("utf-8", "utf8"):
    sys.stdout = open(sys.stdout.fileno(), mode="w", encoding="utf-8", buffering=1)


def _load_json(path: Path) -> dict[str, Any]:
    if not path.exists():
        return {}
    try:
        data = json.loads(path.read_text(encoding="utf-8"))
    except Exception:
        return {}
    return data if isinstance(data, dict) else {}


def _table(rows: list[dict[str, Any]], keys: list[str], max_rows: int = 25) -> str:
    if not rows:
        return "_데이터 없음_"
    header = "| " + " | ".join(keys) + " |"
    sep = "| " + " | ".join(["---"] * len(keys)) + " |"
    body = []
    for row in rows[:max_rows]:
        body.append("| " + " | ".join(str(row.get(k, "")).replace("|", "ㅣ") for k in keys) + " |")
    suffix = f"\n\n_...총 {len(rows)}행 중 {max_rows}행 표시_" if len(rows) > max_rows else ""
    return "\n".join([header, sep, *body]) + suffix


def main() -> int:
    review = _load_json(REVIEW)
    rows = review.get("review_rows") if isinstance(review.get("review_rows"), list) else []

    approval_candidates: list[dict[str, Any]] = []
    provisional_review_candidates: list[dict[str, Any]] = []

    for row in rows:
        if not isinstance(row, dict):
            continue
        source = str(row.get("source") or "")
        readiness = str(row.get("readiness") or "")
        sample_n = int(row.get("sample_n") or 0)
        base = {
            "id": row.get("id", ""),
            "source": source,
            "sample_n": sample_n,
            "value": row.get("value"),
            "unit": row.get("unit", ""),
            "owner_formula": row.get("owner_formula", ""),
            "readiness": readiness,
            "reason": row.get("reason", ""),
        }
        if source == "PROVISIONAL":
            approval_candidates.append(base)
        elif readiness == "PROVISIONAL_CANDIDATE":
            provisional_review_candidates.append(base)

    approval_candidates.sort(key=lambda item: (-int(item.get("sample_n") or 0), str(item.get("id") or "")))
    provisional_review_candidates.sort(key=lambda item: (-int(item.get("sample_n") or 0), str(item.get("id") or "")))

    report = {
        "formula_id": "CALIBRATION_APPROVAL_LIST_V1",
        "generated_at": datetime.now(timezone.utc).isoformat(),
        "review_report_path": str(REVIEW),
        "approval_candidate_count": len(approval_candidates),
        "provisional_review_candidate_count": len(provisional_review_candidates),
        "approval_candidates": approval_candidates,
        "provisional_review_candidates": provisional_review_candidates,
    }

    OUT_JSON.write_text(json.dumps(report, ensure_ascii=False, indent=2), encoding="utf-8")

    md_lines = [
        "# Calibration Approval List",
        "",
        "## Summary",
        "",
        f"- approval candidates: {len(approval_candidates)}",
        f"- provisional review candidates: {len(provisional_review_candidates)}",
        "",
        "## Approval Candidates",
        "",
        _table(approval_candidates, ["id", "source", "sample_n", "value", "unit", "owner_formula", "readiness", "reason"]),
        "",
        "## Provisional Review Candidates",
        "",
        _table(provisional_review_candidates, ["id", "source", "sample_n", "value", "unit", "owner_formula", "readiness", "reason"]),
        "",
        "## Evidence",
        "",
        f"- review report: {REVIEW}",
    ]
    OUT_MD.write_text("\n".join(md_lines), encoding="utf-8")

    print(json.dumps({
        "formula_id": report["formula_id"],
        "gate": "PASS" if approval_candidates else "WARN",
        "approval_candidate_count": len(approval_candidates),
        "provisional_review_candidate_count": len(provisional_review_candidates),
        "json_path": str(OUT_JSON),
        "md_path": str(OUT_MD),
    }, ensure_ascii=False, indent=2))
    return 0


if __name__ == "__main__":
    raise SystemExit(main())