scripts/materialize_math_ink_06_p_formula.py

"""Curated AIFlow Ink v1과 사람 annotation을 P Formula v1 JSONL로 엄격 변환한다."""

from __future__ import annotations

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

import torch

PROJECT_ROOT = Path(__file__).parents[1]
SOURCE_ROOT = PROJECT_ROOT / "src"
if str(SOURCE_ROOT) not in sys.path:
    sys.path.insert(0, str(SOURCE_ROOT))

from math_grid_drawer.research.p_curation import read_p_records
from math_grid_drawer.research.p_formula_gate06 import audit_p_formula_records06
from math_grid_drawer.research.p_formula_intake06 import (
    materialize_p_formula_records06,
    read_p_formula_annotations06,
    write_p_formula_jsonl06,
)


def _checkpoint_labels06(path: Path) -> tuple[str, ...]:
    """필요 변수: 신뢰된 로컬 0.6 checkpoint. 작동 원리: 중복 없는 고정 378 exact vocabulary만 반환한다."""

    payload: dict[str, Any] = torch.load(
        path,
        map_location="cpu",
        weights_only=False,
    )
    raw_labels = payload.get("exact_labels")
    if not isinstance(raw_labels, (list, tuple)):
        raise ValueError("checkpoint에 exact_labels 계약이 없습니다.")
    labels = tuple(str(label) for label in raw_labels)
    if len(labels) != 378 or len(set(labels)) != 378 or any(not label for label in labels):
        raise ValueError("P Formula materialize에는 중복 없는 378 exact label checkpoint가 필요합니다.")
    return labels


def main() -> None:
    """필요 변수: curated root·annotation JSONL·출력·감사 report. 작동 원리: 전체 product gate 통과 후에만 원자적으로 materialize한다."""

    parser = argparse.ArgumentParser(description="Materialize AIFlow P Formula v1")
    parser.add_argument("--input", type=Path, required=True)
    parser.add_argument("--annotations", type=Path, required=True)
    parser.add_argument("--checkpoint", type=Path, required=True)
    parser.add_argument("--output", type=Path, required=True)
    parser.add_argument("--report", type=Path, required=True)
    parser.add_argument("--minimum-independent-sources", type=int, default=2)
    parser.add_argument("--overwrite", action="store_true")
    args = parser.parse_args()

    intake_records = read_p_records(args.input)
    annotations = read_p_formula_annotations06(args.annotations)
    labels = _checkpoint_labels06(args.checkpoint)
    records = materialize_p_formula_records06(
        intake_records,
        annotations,
        allowed_labels=labels,
    )
    report = audit_p_formula_records06(
        records,
        minimum_independent_sources=args.minimum_independent_sources,
    )
    report.update({
        "generated_at": datetime.now(timezone.utc).isoformat(),
        "input": str(args.input),
        "annotations": str(args.annotations),
        "checkpoint": str(args.checkpoint),
        "vocabulary_labels": len(labels),
        "output": str(args.output),
        "materialized": False,
    })
    args.report.parent.mkdir(parents=True, exist_ok=True)
    args.report.write_text(
        json.dumps(report, ensure_ascii=False, indent=2) + "\n",
        encoding="utf-8",
    )
    if not report["eligible_for_product_evaluation"]:
        print(json.dumps(report, ensure_ascii=False, indent=2))
        raise SystemExit(2)
    write_p_formula_jsonl06(args.output, records, overwrite=args.overwrite)
    report["materialized"] = True
    args.report.write_text(
        json.dumps(report, ensure_ascii=False, indent=2) + "\n",
        encoding="utf-8",
    )
    print(json.dumps(report, ensure_ascii=False, indent=2))


if __name__ == "__main__":
    main()