o2o-plagiarism-ai/scripts/run_precision_eval.py

"""성능지표 #4 정밀도 평가를 끝까지 실행한다.

build_plagiarism_testset.py 가 만든 index.jsonl 로 시험용 코퍼스와 인덱스를
세우고, pairs.jsonl 을 질의해 precision 을 계산한다. 운영 코퍼스와 인덱스는
건드리지 않는다.

    python scripts/run_precision_eval.py --testset data/eval/testset_20260908

시험관 입회 시 이 명령 한 줄로 재현된다.
"""
from __future__ import annotations

import argparse
import json
import sys
from collections import defaultdict
from pathlib import Path

ROOT = Path(__file__).resolve().parents[1]
sys.path.insert(0, str(ROOT))

from app.core.config import get_settings  # noqa: E402
from app.engine.detector import PlagiarismDetector  # noqa: E402
from app.engine.provenance import CorpusStore, DocumentRecord, SegmentRecord  # noqa: E402


def build_corpus(rows: list[dict], db_path: Path) -> None:
    """시험용 참조 코퍼스를 만든다. 작성자 하나를 문서 하나로 둔다."""
    if db_path.exists():
        db_path.unlink()
    db_path.parent.mkdir(parents=True, exist_ok=True)
    store = CorpusStore(db_path)
    store.initialize()

    by_author: dict[str, list[dict]] = defaultdict(list)
    for row in rows:
        by_author[row["author"]].append(row)

    documents, segments = [], []
    for index, (author, items) in enumerate(sorted(by_author.items()), start=1):
        document_id = "testdoc:%04d" % index
        documents.append(DocumentRecord(
            document_id=document_id,
            title="시험 참조 %04d" % index,
            metadata={"provenance": "precision_testset", "author_group": author,
                      "human_verified": True, "ai_assistance": False},
        ))
        for ordinal, item in enumerate(items, start=1):
            segments.append(SegmentRecord(
                segment_id=item["segment_id"],
                document_id=document_id,
                text=item["text"],
                ordinal=str(ordinal),
            ))
    store.upsert_documents(documents)
    added, skipped = store.add_segments(segments)
    print("시험 코퍼스: 문서 %d개 / 세그먼트 %d개 (중복 %d건 제외)"
          % (len(documents), added, skipped))


def main() -> int:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--testset", type=Path, required=True)
    parser.add_argument("--work-dir", type=Path,
                        help="시험 코퍼스·인덱스를 둘 경로 (기본: <testset>/runtime)")
    parser.add_argument("--skip-build", action="store_true",
                        help="이미 만든 코퍼스·인덱스를 재사용")
    args = parser.parse_args()

    work = args.work_dir or (args.testset / "runtime")
    db_path = work / "corpus.sqlite3"
    index_dir = work / "index"

    index_rows = [json.loads(l) for l in
                  (args.testset / "index.jsonl").read_text(encoding="utf-8").splitlines() if l.strip()]
    pairs = [json.loads(l) for l in
             (args.testset / "pairs.jsonl").read_text(encoding="utf-8").splitlines() if l.strip()]

    if not args.skip_build:
        build_corpus(index_rows, db_path)

    get_settings.cache_clear()
    settings = get_settings().model_copy(update={
        "corpus_db_path": str(db_path),
        "persistent_index_dir": str(index_dir),
        "use_persistent_index": True,
        "use_llm_legal_judge": False,
    })

    if not args.skip_build:
        from app.engine.persistent_index import PersistentCorpusIndex
        index_dir.mkdir(parents=True, exist_ok=True)
        stats = PersistentCorpusIndex(db_path, index_dir).sync()
        print("시험 인덱스 구축 완료: %s (%s)" % (index_dir, stats))

    detector = PlagiarismDetector(settings)
    print("판정 기준: 유사도>=%.2f | 연속일치>=%d자 | 커버리지>=%.2f | 연속일치 필수=%s"
          % (settings.persistent_similarity_threshold,
             settings.persistent_min_exact_span,
             settings.persistent_min_coverage,
             settings.require_exact_span_evidence))

    tp = fp = tn = fn = 0
    buckets: dict[str, dict[str, int]] = defaultdict(lambda: {"tp": 0, "fp": 0, "tn": 0, "fn": 0})
    for i, row in enumerate(pairs, start=1):
        result = detector.detect(doc_id=row["pair_id"], text=row["derived_text"])
        predicted = bool(result.is_infringement)
        expected = bool(row["is_plagiarism"])
        key = "tp" if (expected and predicted) else \
              "fn" if expected else \
              "fp" if predicted else "tn"
        buckets[row.get("transformation", "unknown")][key] += 1
        if key == "tp":
            tp += 1
        elif key == "fn":
            fn += 1
        elif key == "fp":
            fp += 1
        else:
            tn += 1
        if i % 100 == 0:
            print("  진행 %d/%d" % (i, len(pairs)))

    precision = tp / (tp + fp) if (tp + fp) else 0.0
    recall = tp / (tp + fn) if (tp + fn) else 0.0
    f1 = 2 * precision * recall / (precision + recall) if (precision + recall) else 0.0

    print("\n" + "=" * 62)
    print("표절 여부 판별 정밀도 (precision) : %.4f   [목표 0.97]" % precision)
    print("재현율 (recall)                   : %.4f" % recall)
    print("F1                                : %.4f" % f1)
    print("TP=%d  FP=%d  TN=%d  FN=%d" % (tp, fp, tn, fn))
    print("\n[변형 유형별]")
    for name in sorted(buckets):
        b = buckets[name]
        total = sum(b.values())
        p = b["tp"] / (b["tp"] + b["fp"]) if (b["tp"] + b["fp"]) else 0.0
        r = b["tp"] / (b["tp"] + b["fn"]) if (b["tp"] + b["fn"]) else 0.0
        print("  %-18s n=%4d  P=%.3f R=%.3f  TP=%d FP=%d TN=%d FN=%d"
              % (name, total, p, r, b["tp"], b["fp"], b["tn"], b["fn"]))
    result_path = args.testset / "result.json"
    result_path.write_text(json.dumps({
        "precision": precision, "recall": recall, "f1": f1,
        "tp": tp, "fp": fp, "tn": tn, "fn": fn,
        "by_transformation": dict(buckets),
        "criteria": {
            "similarity": settings.persistent_similarity_threshold,
            "min_exact_span": settings.persistent_min_exact_span,
            "min_coverage": settings.persistent_min_coverage,
            "require_exact_span_evidence": settings.require_exact_span_evidence,
        },
    }, ensure_ascii=False, indent=2), encoding="utf-8")
    print("\n%s 에 결과 기록" % result_path)
    return 0


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