feat: add reporting xlsx builder for infringement batches

배치 산출 JSONL은 엔진 내부 값(retrieval_backend, legal_judgment_method,
case_id 등 전 건 단일값 컬럼 포함)까지 22컬럼을 담고 있어 그대로 공유하면
오독을 부른다. 원본 JSONL을 건드리지 않고 보고에 필요한 컬럼만 추려
재생성하는 별도 스크립트를 추가한다.

- 침해 판별 결과: 전수 8컬럼, 의심 건 우선 정렬
- 의심 건 상세: 침해유형 한글화 + 관련 판례 연결
- 요약 통계: 근거 스팬 보유 건수와 해석 유의사항
- 1차_vs_2차 비교: 근거 스팬/커버리지 병기, 상위 N 절단 제거

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
This commit is contained in:
hbyang 2026-09-02 13:46:31 +09:00
parent 8717fa4425
commit 87fc98119d

View File

@ -0,0 +1,256 @@
"""배치 JSONL에서 보고용 XLSX를 다시 만든다.
배치 산출 JSONL은 엔진 내부 값까지 전부 담고 있어 그대로 공유하면 오독을 부른다.
스크립트는 원본 JSONL을 건드리지 않고, 보고에 필요한 컬럼만 추려 다시 쓴다.
"""
from __future__ import annotations
import argparse
import json
from collections import Counter
from pathlib import Path
from openpyxl import Workbook
from openpyxl.styles import Alignment, Font, PatternFill
from openpyxl.utils import get_column_letter
TAG_LABELS = {
"reproduction": "복제",
"citation_missing": "출처표시 누락",
"public_transmission": "공중송신",
"adaptation": "2차적저작물작성",
"distribution": "배포",
"attribution": "성명표시",
}
RESULT_COLUMNS = [
("doc_id", "doc_id", 28),
("가명 제목", "가명 제목", 22),
("원천 구분", "원천 구분", 14),
("침해 여부", "침해 여부", 12),
("결합유사도", "결합유사도", 11),
("매칭 상대 제목", "매칭 상대 제목", 22),
("evidence 스팬 수", "근거 스팬 수", 12),
("union_coverage", "근거 커버리지", 12),
]
DETAIL_COLUMNS = [
("doc_id", "doc_id", 28),
("가명 제목", "가명 제목", 22),
("원천 구분", "원천 구분", 14),
("결합유사도", "결합유사도", 11),
("매칭 상대 제목", "매칭 상대 제목", 22),
("매칭 상대 doc", "매칭 상대 doc", 28),
("evidence 스팬 수", "근거 스팬 수", 12),
("union_coverage", "근거 커버리지", 12),
("text 점수", "표현 유사도", 11),
("lemma 점수", "어휘 유사도", 11),
("주 법령태그", "주 침해유형", 18),
("보조 법령태그", "보조 침해유형", 16),
("관련 판례 사건번호", "관련 판례 사건번호", 46),
("judgment_summary", "판단 요약", 70),
]
HEADER_FILL = PatternFill("solid", fgColor="D9E1F2")
NOTE_FONT = Font(italic=True, color="7F7F7F")
def load(path: Path) -> dict[str, dict]:
rows = {}
for line in path.read_text(encoding="utf-8").splitlines():
if line.strip():
row = json.loads(line)
rows[row["query_key"]] = row
return rows
def localize_tags(value) -> str:
tags = [tag.strip() for tag in str(value or "").split(",") if tag.strip()]
return ", ".join(TAG_LABELS.get(tag, tag) for tag in tags)
def write_header(sheet, columns) -> None:
sheet.append([label for _, label, _ in columns])
for index, (_, _, width) in enumerate(columns, start=1):
cell = sheet.cell(row=1, column=index)
cell.font = Font(bold=True)
cell.fill = HEADER_FILL
cell.alignment = Alignment(vertical="center")
sheet.column_dimensions[get_column_letter(index)].width = width
sheet.freeze_panes = "A2"
def cell_value(row: dict, key: str):
value = row.get(key)
if key in ("주 법령태그", "보조 법령태그"):
return localize_tags(value)
if key == "매칭 상대 제목" and not value:
return "-"
return value
RATIO_COLUMNS = {"결합유사도", "근거 커버리지", "표현 유사도", "어휘 유사도"}
def append_rows(sheet, rows, columns) -> None:
for row in rows:
sheet.append([cell_value(row, key) for key, _, _ in columns])
for index, (_, label, _) in enumerate(columns, start=1):
if label not in RATIO_COLUMNS:
continue
for cell in sheet.iter_cols(min_col=index, max_col=index, min_row=2):
for item in cell:
item.number_format = "0.0000"
def build_result_sheet(wb: Workbook, rows: list[dict]) -> None:
sheet = wb.create_sheet("침해 판별 결과")
write_header(sheet, RESULT_COLUMNS)
order = {"침해 의심": 0, "매칭 미탐지": 1}
ordered = sorted(rows, key=lambda row: (order.get(row["침해 여부"], 9), -(row["결합유사도"] or 0)))
append_rows(sheet, ordered, RESULT_COLUMNS)
sheet.auto_filter.ref = f"A1:{get_column_letter(len(RESULT_COLUMNS))}{sheet.max_row}"
def build_detail_sheet(wb: Workbook, rows: list[dict]) -> None:
sheet = wb.create_sheet("의심 건 상세")
write_header(sheet, DETAIL_COLUMNS)
suspected = sorted(
(row for row in rows if row["침해 의심"]),
key=lambda row: -(row["결합유사도"] or 0),
)
append_rows(sheet, suspected, DETAIL_COLUMNS)
sheet.auto_filter.ref = f"A1:{get_column_letter(len(DETAIL_COLUMNS))}{sheet.max_row}"
def build_summary_sheet(wb: Workbook, rows: list[dict], *, backend: str, notes: list[str]) -> None:
sheet = wb.create_sheet("요약 통계")
sheet.append(["구분", "총 건수", "침해 의심", "의심 비율", "근거 스팬 보유"])
for index in range(1, 6):
sheet.cell(row=1, column=index).font = Font(bold=True)
sheet.cell(row=1, column=index).fill = HEADER_FILL
groups = {
"전체": rows,
"자서전 에피소드": [row for row in rows if row["원천 구분"] == "자서전 에피소드"],
"생활수기": [row for row in rows if row["원천 구분"] == "생활수기"],
}
for label, group in groups.items():
total = len(group)
suspected = [row for row in group if row["침해 의심"]]
with_evidence = sum(1 for row in suspected if (row.get("evidence 스팬 수") or 0) >= 1)
sheet.append([label, total, len(suspected), len(suspected) / total if total else 0, with_evidence])
sheet.cell(row=sheet.max_row, column=4).number_format = "0.00%"
sheet.append([])
sheet.append(["실행 조건", ""])
sheet.cell(row=sheet.max_row, column=1).font = Font(bold=True)
sheet.cell(row=sheet.max_row, column=2).font = Font(bold=True)
for label, value in (
("판정 방식", "규칙 기반 (판례 매칭, LLM 법률판단 미사용)"),
("검색 백엔드", backend),
("원문 텍스트", "결과 파일에 저장하지 않음 (가명 식별자만 수록)"),
):
sheet.append([label, value])
sheet.append([])
sheet.append(["해석 시 유의사항"])
sheet.cell(row=sheet.max_row, column=1).font = Font(bold=True)
for note in notes:
sheet.append([note])
sheet.cell(row=sheet.max_row, column=1).font = NOTE_FONT
for index, width in enumerate((22, 20, 14, 12, 14), start=1):
sheet.column_dimensions[get_column_letter(index)].width = width
COMPARE_COLUMNS = [
"가명 제목", "원천 구분", "매칭 상대 제목",
"1차 유사도", "2차 유사도", "근거 스팬(1차)", "근거 스팬(2차)",
"근거 커버리지(1차)", "근거 커버리지(2차)",
]
def build_compare_sheet(wb: Workbook, first: dict[str, dict], second: dict[str, dict]) -> None:
sheet = wb.create_sheet("1차_vs_2차 비교")
flagged_first = {key for key, row in first.items() if row["침해 의심"]}
flagged_second = {key for key, row in second.items() if row["침해 의심"]}
shared = sorted(set(first) & set(second))
deltas = [(second[key]["결합유사도"] or 0) - (first[key]["결합유사도"] or 0) for key in shared]
mean_delta = sum(deltas) / len(deltas) if deltas else 0
sheet.append(["구분", "건수"])
for index in (1, 2):
sheet.cell(row=1, column=index).font = Font(bold=True)
sheet.cell(row=1, column=index).fill = HEADER_FILL
for label, value in (
("1차 의심", len(flagged_first)),
("2차 의심", len(flagged_second)),
("양쪽 공통", len(flagged_first & flagged_second)),
("2차 신규", len(flagged_second - flagged_first)),
("1차에서만", len(flagged_first - flagged_second)),
):
sheet.append([label, value])
sheet.append([])
sheet.append(["결합유사도 평균 변화", round(mean_delta, 4)])
sheet.append([
"주의: 2차는 임베딩 교체로 전 건 유사도가 일괄 상승했습니다. "
"임계값이 동일하므로 의심 건수 증가분을 곧바로 탐지력 향상으로 해석하면 안 됩니다."
])
sheet.cell(row=sheet.max_row, column=1).font = NOTE_FONT
def section(title: str, keys: list[str], source: dict[str, dict]) -> None:
sheet.append([])
sheet.append([title])
sheet.cell(row=sheet.max_row, column=1).font = Font(bold=True)
sheet.append(COMPARE_COLUMNS)
for index in range(1, len(COMPARE_COLUMNS) + 1):
sheet.cell(row=sheet.max_row, column=index).fill = HEADER_FILL
for key in sorted(keys, key=lambda k: -(source[k]["결합유사도"] or 0)):
a, b = first[key], second[key]
sheet.append([
b["가명 제목"], b["원천 구분"], source[key].get("매칭 상대 제목") or "-",
a["결합유사도"], b["결합유사도"],
a.get("evidence 스팬 수"), b.get("evidence 스팬 수"),
a.get("union_coverage"), b.get("union_coverage"),
])
for column in (4, 5, 8, 9):
sheet.cell(row=sheet.max_row, column=column).number_format = "0.0000"
new_keys = sorted(flagged_second - flagged_first)
with_evidence = [key for key in new_keys if (second[key].get("evidence 스팬 수") or 0) >= 1]
without_evidence = [key for key in new_keys if key not in with_evidence]
section(f"2차 신규 — 근거 스팬 있음 ({len(with_evidence)}건, 우선 검토 대상)", with_evidence, second)
section(f"2차 신규 — 근거 스팬 없음 ({len(without_evidence)}건, 임계값 근접 건)", without_evidence, second)
section(f"1차에서만 검출 ({len(flagged_first - flagged_second)}건)", sorted(flagged_first - flagged_second), first)
for index, width in enumerate((24, 14, 24, 11, 11, 13, 13, 16, 16), start=1):
sheet.column_dimensions[get_column_letter(index)].width = width
def main() -> int:
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument("--jsonl", type=Path, required=True, help="보고 대상 배치 JSONL")
parser.add_argument("--out-xlsx", type=Path, required=True)
parser.add_argument("--compare-jsonl", type=Path, help="비교 시트를 붙일 이전 회차 JSONL")
parser.add_argument("--backend", default="", help="검색 백엔드 표기 (미지정 시 JSONL 값 사용)")
parser.add_argument("--note", action="append", default=[], help="요약 시트에 남길 유의사항")
args = parser.parse_args()
records = load(args.jsonl)
rows = list(records.values())
backend = args.backend or str(rows[0].get("retrieval_backend") or "미기재")
wb = Workbook()
wb.remove(wb.active)
build_result_sheet(wb, rows)
build_detail_sheet(wb, rows)
build_summary_sheet(wb, rows, backend=backend, notes=args.note)
if args.compare_jsonl:
build_compare_sheet(wb, load(args.compare_jsonl), records)
args.out_xlsx.parent.mkdir(parents=True, exist_ok=True)
wb.save(args.out_xlsx)
print(f"wrote {args.out_xlsx} sheets={wb.sheetnames}")
return 0
if __name__ == "__main__":
raise SystemExit(main())