"""성능지표 #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 build_scorecard(settings, tp: int, fp: int, tn: int, fn: int, precision: float, recall: float, f1: float) -> dict: """성적서 "4. 모델 결과" 게재 형식으로 정리한다. 전 차수 성적서(GERI `GERIR.CE.2511-02.009`) 13페이지와 같은 키 구성이라 이 출력을 그대로 캡처해 붙일 수 있다. threshold 는 단일 수치가 아니라 세 조건의 조합이므로 객체로 둔다. 유사도만으로 판정하지 않는다는 점이 판정 기준의 핵심이다. """ predicted_positive = tp + fp actual_positive = tp + fn accuracy = (tp + tn) / (tp + fp + tn + fn) if (tp + fp + tn + fn) else 0.0 fpr = fp / (fp + tn) if (fp + tn) else 0.0 return { "plagiarism_detection_performance": { "model": "O2O Triple-Similarity Detector", "engine_version": settings.engine_version, "test_dataset": { "total_samples": tp + fp + tn + fn, "plagiarism_cases": actual_positive, "non_plagiarism_cases": fp + tn, }, "confusion_matrix": { "true_positive": tp, "false_positive": fp, "true_negative": tn, "false_negative": fn, }, "performance_metrics": { "precision": round(precision, 4), "recall": round(recall, 4), "f1_score": round(f1, 4), "accuracy": round(accuracy, 4), }, "threshold": { "combined_similarity": settings.persistent_similarity_threshold, "min_exact_span_chars": settings.persistent_min_exact_span, "min_coverage": settings.persistent_min_coverage, "require_exact_span_evidence": settings.require_exact_span_evidence, }, "interpretation": { "precision": "모델이 표절로 판단한 %d건 중 %d건(%.1f%%)이 실제 표절" % (predicted_positive, tp, precision * 100), "recall": "실제 표절 %d건 중 %d건(%.1f%%)을 정확히 탐지" % (actual_positive, tp, recall * 100), "false_positive_rate": "%.1f%%" % (fpr * 100), }, } } def main() -> int: parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--testset", type=Path, required=True) parser.add_argument("--work-dir", type=Path, help="시험 코퍼스·인덱스를 둘 경로 (기본: /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"])) scorecard = build_scorecard(settings, tp, fp, tn, fn, precision, recall, f1) rendered = json.dumps(scorecard, ensure_ascii=False, indent=2) print("\n" + "=" * 62) print("4. 모델 결과 :") print() print(rendered) print("\n최종 결과 : precision %.2f%% 달성" % (precision * 100)) (args.testset / "scorecard.json").write_text(rendered + "\n", encoding="utf-8") 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())