110 lines
4.0 KiB
Python
110 lines
4.0 KiB
Python
#!/usr/bin/env python3
|
|
"""리스트업 등급을 엔진 적재본에 반영한다.
|
|
|
|
왜 필요한가:
|
|
사람이 판시 본문을 읽고 매긴 A/B/C 등급이 `precedent_listup.csv` 에만 있고
|
|
엔진이 로드하는 `precedents.jsonl` 에는 없었다. 그래서 판정 때 인용되는 상위
|
|
5건이 수집 단계 라벨(`work_types`, `legal_tags`)만으로 정해졌고, 리스트업에서
|
|
"어문 표절 기준을 주지 않는다"며 제외한 사건이 그대로 근거로 인용됐다.
|
|
|
|
실제로 운영 스모크 테스트에서 `95가합11403`(폰트·프로그램)이 인용됐다.
|
|
|
|
무엇을 하는가:
|
|
CSV 의 `grade` 를 JSONL 각 레코드에 `grade` 필드로 넣는다. 등급이 없는 건은
|
|
필드를 넣지 않는다. 엔진은 545건 전체를 검색하되 A/B/C를 작은 품질 가중치와
|
|
검토 상태 표시에 사용한다. 미등급 판례도 내용 관련성이 높으면 인용될 수 있다.
|
|
|
|
원천 판례를 지우지 않는다. 545건 모두 적재본과 런타임 검색 후보로 유지한다.
|
|
|
|
사용:
|
|
python scripts/apply_precedent_grades.py # 미리보기
|
|
python scripts/apply_precedent_grades.py --write # 실제 반영
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import csv
|
|
import io
|
|
import json
|
|
import sys
|
|
from collections import Counter
|
|
from pathlib import Path
|
|
|
|
ROOT = Path(__file__).resolve().parent.parent
|
|
|
|
|
|
def load_grades(csv_path: Path) -> dict[str, str]:
|
|
if not csv_path.exists():
|
|
raise SystemExit(f"{csv_path} 가 없습니다.")
|
|
with io.open(csv_path, encoding="utf-8-sig", newline="") as fh:
|
|
rows = list(csv.DictReader(fh))
|
|
grades = {}
|
|
for row in rows:
|
|
case_id = (row.get("case_id") or "").strip()
|
|
grade = (row.get("grade") or "").strip().upper()
|
|
if case_id and grade in {"A", "B", "C"}:
|
|
grades[case_id] = grade
|
|
if not grades:
|
|
raise SystemExit(f"{csv_path} 에서 등급을 읽지 못했습니다.")
|
|
return grades
|
|
|
|
|
|
def main() -> int:
|
|
ap = argparse.ArgumentParser(
|
|
description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter
|
|
)
|
|
ap.add_argument("--csv", type=Path,
|
|
default=ROOT / "data/precedents/precedent_listup.csv")
|
|
ap.add_argument("--jsonl", type=Path,
|
|
default=ROOT / "data/precedents/precedents.jsonl")
|
|
ap.add_argument("--write", action="store_true",
|
|
help="실제로 파일을 고친다. 없으면 미리보기만 한다.")
|
|
args = ap.parse_args()
|
|
|
|
grades = load_grades(args.csv)
|
|
lines = args.jsonl.read_text(encoding="utf-8").splitlines()
|
|
|
|
out: list[str] = []
|
|
applied = Counter()
|
|
cleared = 0
|
|
unmatched = set(grades)
|
|
|
|
for line in lines:
|
|
if not line.strip():
|
|
continue
|
|
rec = json.loads(line)
|
|
case_id = str(rec.get("case_id", ""))
|
|
unmatched.discard(case_id)
|
|
grade = grades.get(case_id)
|
|
if grade:
|
|
rec["grade"] = grade
|
|
applied[grade] += 1
|
|
elif "grade" in rec:
|
|
# 리스트업에서 빠진 건에 예전 등급이 남아 있으면 지운다.
|
|
del rec["grade"]
|
|
cleared += 1
|
|
out.append(json.dumps(rec, ensure_ascii=False, separators=(",", ":")))
|
|
|
|
print(f"적재본 {len(out)}건")
|
|
print(f" 등급 부여: A {applied['A']} · B {applied['B']} · C {applied['C']} "
|
|
f"(합 {sum(applied.values())})")
|
|
print(f" 미검토 : {len(out) - sum(applied.values())}건 (grade 필드 없음)")
|
|
if cleared:
|
|
print(f" 등급 제거: {cleared}건 (리스트업에서 빠짐)")
|
|
if unmatched:
|
|
print(f" ⚠️ CSV 에 있으나 적재본에 없는 사건: {len(unmatched)}건 → "
|
|
f"{', '.join(sorted(unmatched)[:5])}")
|
|
|
|
if not args.write:
|
|
print("\n미리보기입니다. 실제 반영은 --write 를 붙이세요.")
|
|
return 0
|
|
|
|
args.jsonl.write_text("\n".join(out) + "\n", encoding="utf-8")
|
|
print(f"\n반영 완료: {args.jsonl}")
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main())
|