737 lines
36 KiB
Python
737 lines
36 KiB
Python
"""배치 JSONL에서 보고용 XLSX를 다시 만든다.
|
||
|
||
배치 산출 JSONL은 엔진 내부 값까지 전부 담고 있어 그대로 공유하면 오독을 부른다.
|
||
이 스크립트는 원본 JSONL을 건드리지 않고, 보고에 필요한 컬럼만 추려 다시 쓴다.
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import argparse
|
||
import hashlib
|
||
import json
|
||
import re
|
||
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),
|
||
("id", "id (제출자 이메일)", 26),
|
||
("원문 자서전 제목", "원문 자서전 제목", 22),
|
||
("가명 제목", "가명 제목", 22),
|
||
("원천 구분", "원천 구분", 14),
|
||
("침해 여부", "침해 여부", 12),
|
||
("결합유사도", "결합유사도", 11),
|
||
("매칭 상대 제목", "매칭 상대 제목", 22),
|
||
("매칭 상대 id", "매칭 상대 id (이메일)", 26),
|
||
("동일 이메일", "동일 이메일", 12),
|
||
("evidence 스팬 수", "근거 스팬 수", 12),
|
||
("union_coverage", "근거 커버리지", 12),
|
||
]
|
||
|
||
DETAIL_COLUMNS = [
|
||
("doc_id", "doc_id", 28),
|
||
("id", "id (제출자 이메일)", 26),
|
||
("원문 자서전 제목", "원문 자서전 제목", 22),
|
||
("원문 에피소드 제목", "원문 에피소드 제목", 26),
|
||
("원본 엑셀 행", "원본 엑셀 행", 12),
|
||
("가명 제목", "가명 제목", 22),
|
||
("원천 구분", "원천 구분", 14),
|
||
("결합유사도", "결합유사도", 11),
|
||
("매칭 상대 제목", "매칭 상대 제목", 22),
|
||
("매칭 상대 doc", "매칭 상대 doc", 28),
|
||
("매칭 상대 id", "매칭 상대 id (이메일)", 26),
|
||
("매칭 상대 원문 제목", "매칭 상대 원문 제목", 22),
|
||
("동일 이메일", "동일 이메일", 12),
|
||
("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}"
|
||
sheet.append([])
|
||
for line in (
|
||
"관련 판례는 침해 판정의 근거가 아닙니다. 엔진이 법령태그와 어휘 유사도로 뽑은 상위 5건이며, "
|
||
"점수 하한이 없어 관련성이 낮은 판례도 항상 채워집니다.",
|
||
"실제로 매칭 미탐지 건에도 같은 방식으로 판례가 붙습니다. 판례가 있다는 사실만으로 "
|
||
"침해 가능성을 판단하시면 안 되며, 사건번호를 직접 확인하셔야 합니다.",
|
||
):
|
||
sheet.append([line])
|
||
sheet.cell(row=sheet.max_row, column=1).font = NOTE_FONT
|
||
|
||
|
||
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 (
|
||
("침해 판별", "규칙 기반 (유사도·연속일치·커버리지 3개 조건)"),
|
||
("판례 검색", "규칙 기반. 법령태그·어휘 유사도로 상위 5건을 뽑으며 점수 하한이 없음"),
|
||
("법적 판단", "미수행 (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
|
||
|
||
|
||
|
||
CRITERIA_ROWS = [
|
||
(1, "결합유사도", "≥ 0.65", "persistent_similarity_threshold",
|
||
"네 가지 유사도를 가중 합산한 값이 임계값 이상"),
|
||
(2, "최장 연속 일치 길이", "≥ 35자 (9어절)", "persistent_min_exact_span",
|
||
"두 글이 끊기지 않고 그대로 이어지는 최장 구간. 근거는 '어절 기준 비교' 시트"),
|
||
(3, "문서 단위 일치 커버리지", "≥ 0.30 (30%)", "persistent_min_coverage",
|
||
"질의 문서 전체에서 일치 구간이 차지하는 비율(중복 제외)"),
|
||
]
|
||
|
||
WEIGHT_ROWS = [
|
||
("어휘 유사도 (lemma)", 0.45, "형태소 원형 기준. 어미·조사 변형을 흡수"),
|
||
("표현 유사도 (text)", 0.30, "표층 문자열 기준"),
|
||
("문자 유사도 (character)", 0.15, "문자 3~4-gram 기준"),
|
||
("모티프 유사도 (motif)", 0.10, "사건·소재 요소 일치"),
|
||
]
|
||
|
||
PARAMETER_ROWS = [
|
||
("후보 재정렬 대상", "상위 20건", "persistent_rerank_top_k",
|
||
"정밀 비교에 참여하는 후보 수"),
|
||
("증거 스팬 최소 길이", "12자", "_evidence_spans(min_match)",
|
||
"이보다 짧은 일치는 증거로 세지 않음"),
|
||
("증거 스팬 표시 개수", "최대 10개", "_evidence_spans(limit)",
|
||
"커버리지 계산은 전량, 표시만 제한"),
|
||
("자기 문서 제외", "적용", "source_group 필터",
|
||
"동일 문서 및 동일 원천 그룹은 후보에서 제외"),
|
||
("판례 검색", "상위 5건 고정", "legal_risk.assess()",
|
||
"법령태그 0.55 + 어휘 0.20 + 판시기준 0.10 + 유형 0.08 + 충실도 0.02 로 정렬"),
|
||
("판례 점수 하한", "없음", "legal_risk.assess()",
|
||
"하한이 없어 관련성이 낮아도 항상 5건이 붙는다. 매칭 미탐지 건에도 판례가 채워지는 이유"),
|
||
("법적 판단", "미수행", "USE_LLM_LEGAL_JUDGE=false",
|
||
"판례를 근거로 침해 여부를 판단하는 단계는 실행하지 않음. 외부 LLM에 원문 미전송"),
|
||
]
|
||
|
||
|
||
def build_criteria_sheet(wb: Workbook, rows: list[dict], excerpts: list[dict] | None = None) -> None:
|
||
sheet = wb.create_sheet("판정 기준")
|
||
|
||
def header(*labels: str) -> None:
|
||
sheet.append(list(labels))
|
||
for index in range(1, len(labels) + 1):
|
||
cell = sheet.cell(row=sheet.max_row, column=index)
|
||
cell.font = Font(bold=True)
|
||
cell.fill = HEADER_FILL
|
||
|
||
def title(text: str) -> None:
|
||
sheet.append([text])
|
||
sheet.cell(row=sheet.max_row, column=1).font = Font(bold=True, size=12)
|
||
|
||
def note(text: str) -> None:
|
||
sheet.append([text])
|
||
sheet.cell(row=sheet.max_row, column=1).font = NOTE_FONT
|
||
|
||
title("1. 침해 의심 판정 조건")
|
||
note("세 조건 중 하나 이상을 충족하면 침해 의심으로 분류합니다 (OR 조건). 세 조건 모두를 요구하지 않습니다.")
|
||
header("번호", "조건", "기준값", "설정 키", "설명")
|
||
for row in CRITERIA_ROWS:
|
||
sheet.append(list(row))
|
||
|
||
sheet.append([])
|
||
title("2. 결합유사도 산식")
|
||
note("결합유사도 = 어휘×0.45 + 표현×0.30 + 문자×0.15 + 모티프×0.10 (실측 기반 가중치)")
|
||
header("구성 요소", "가중치", "설명")
|
||
for label, weight, description in WEIGHT_ROWS:
|
||
sheet.append([label, weight, description])
|
||
sheet.cell(row=sheet.max_row, column=2).number_format = "0.00"
|
||
|
||
sheet.append([])
|
||
title("3. 조건별 충족 현황 (침해 의심 건 기준)")
|
||
suspected = [row for row in rows if row["침해 의심"]]
|
||
longest_by_key = {}
|
||
if excerpts:
|
||
for item in excerpts:
|
||
spans = item.get("일치 구간") or []
|
||
longest_by_key[item["query_key"]] = max((len(span) for span in spans), default=0)
|
||
# 조건별 충족 수와 조합별 건수는 서로 다른 집계다. 한 카운터에 섞으면
|
||
# 단독 사유 건이 양쪽에 잡혀 합계가 전체 건수를 넘는다.
|
||
condition_counts = Counter()
|
||
combo_counts = Counter()
|
||
for row in suspected:
|
||
reasons = judgment_reasons(row, longest_by_key.get(row["query_key"], 0))
|
||
combo_counts[" + ".join(reasons) if reasons else "(사유 미상)"] += 1
|
||
for reason in reasons:
|
||
condition_counts[reason] += 1
|
||
header("구분", "건수", "비고")
|
||
sheet.append(["침해 의심 전체", len(suspected), ""])
|
||
note("아래 세 줄은 중복 집계입니다. 한 건이 여러 조건을 동시에 충족할 수 있어 합계가 103을 넘습니다.")
|
||
sheet.append(["조건 ① 유사도 (0.65 이상) 충족", condition_counts["①유사도"], ""])
|
||
if excerpts:
|
||
sheet.append(["조건 ② 연속일치 (35자·9어절 이상) 충족", condition_counts["②연속일치"], "원문 대조로 실측"])
|
||
sheet.append(["조건 ③ 커버리지 (30% 이상) 충족", condition_counts["③커버리지"], ""])
|
||
sheet.append([])
|
||
sheet.append(["판정 사유 조합별 건수", "", "'원문 대조' 시트의 판정 사유 컬럼과 같은 값. 합계 = 103"])
|
||
sheet.cell(row=sheet.max_row, column=1).font = Font(bold=True)
|
||
for key, value in sorted(combo_counts.items(), key=lambda item: (-item[1], item[0])):
|
||
single = " + " not in key and not key.startswith("(")
|
||
sheet.append([
|
||
key + (" 단독" if single else ""), value,
|
||
"이 조건 하나만으로 걸린 건" if single else "",
|
||
])
|
||
sheet.append(["합계", sum(combo_counts.values()), ""])
|
||
sheet.cell(row=sheet.max_row, column=1).font = Font(bold=True)
|
||
if not excerpts:
|
||
note("조건 ②(최장 연속 일치 80자)는 원문 발췌 파일이 있어야 집계할 수 있습니다. --excerpts-jsonl 을 주면 실측값이 채워집니다.")
|
||
|
||
sheet.append([])
|
||
title("4. 그 밖의 동작 기준")
|
||
header("항목", "값", "설정 키", "설명")
|
||
for row in PARAMETER_ROWS:
|
||
sheet.append(list(row))
|
||
|
||
sheet.append([])
|
||
title("5. 기준값 신뢰도에 관한 한계")
|
||
for line in (
|
||
"결합유사도 임계값 0.65는 실데이터의 오탐 분포를 측정해 도출한 값이 아닌 잠정값입니다 "
|
||
"(similarity_threshold_calibrated = false). API 응답에도 provisional = true 로 표기됩니다.",
|
||
"따라서 현재 결과로 정확도(정밀도·재현율)를 제시할 수 없습니다. "
|
||
"원문 대조로 정답 라벨을 만든 뒤 임계값을 재조정하는 것이 다음 단계입니다.",
|
||
"'매칭 미탐지'는 등록 코퍼스 안에서 일치 증거를 찾지 못했다는 뜻이며 비침해 확정이 아닙니다.",
|
||
):
|
||
note(line)
|
||
|
||
for index, width in enumerate((26, 24, 18, 30, 62), start=1):
|
||
sheet.column_dimensions[get_column_letter(index)].width = width
|
||
|
||
|
||
|
||
def _norm(text: str) -> str:
|
||
"""공백 차이만 있는 원고를 같은 것으로 본다."""
|
||
return re.sub(r"\s+", "", text or "")
|
||
|
||
|
||
def _text_key(text: str) -> str:
|
||
return hashlib.sha256(_norm(text).encode("utf-8")).hexdigest()[:16]
|
||
|
||
|
||
def duplicate_groups(excerpts: list[dict]) -> dict[str, list[dict]]:
|
||
groups: dict[str, list[dict]] = {}
|
||
for row in excerpts:
|
||
if _norm(row.get("검사 대상 원문", "")):
|
||
groups.setdefault(_text_key(row["검사 대상 원문"]), []).append(row)
|
||
return {key: rows for key, rows in groups.items() if len(rows) > 1}
|
||
|
||
|
||
DUP_COLUMNS = [
|
||
("가명 제목", 20), ("문서 ID", 28), ("매칭 상대 제목", 20), ("매칭 상대 문서 ID", 28),
|
||
("결합유사도", 11), ("본문 완전 동일", 13), ("본문 길이", 10),
|
||
("같은 본문을 가진 문서", 46), ("검사 대상 원문", 70), ("매칭 상대 원문", 70),
|
||
]
|
||
|
||
|
||
def build_duplicate_sheet(wb: Workbook, excerpts: list[dict], records: dict[str, dict]) -> None:
|
||
"""중복 제거 후 재실행하면 사라질 건들을 원문째로 남긴다."""
|
||
sheet = wb.create_sheet("중복 원고")
|
||
groups = duplicate_groups(excerpts)
|
||
member_of = {r["query_key"]: rows for rows in groups.values() for r in rows}
|
||
affected = [
|
||
r for r in excerpts
|
||
if r.get("침해 여부") == "침해 의심" and r["query_key"] in member_of
|
||
]
|
||
cross = sum(1 for rows in groups.values()
|
||
if len({records[r["query_key"]]["doc_id"] for r in rows}) > 1)
|
||
suspected = sum(1 for r in excerpts if r.get("침해 여부") == "침해 의심")
|
||
|
||
sheet.append(["중복 원고 현황"])
|
||
sheet.cell(row=1, column=1).font = Font(bold=True, size=12)
|
||
for label, value, note in (
|
||
("검사 대상 전체", len(excerpts), ""),
|
||
("고유 본문", len(excerpts) - sum(len(v) - 1 for v in groups.values()), ""),
|
||
("중복 그룹", len(groups), ""),
|
||
(" 서로 다른 문서 사이", cross, "같은 제출자의 재제출인지 다른 제출자의 표절인지 확인 필요"),
|
||
(" 같은 문서 안", len(groups) - cross, "적재 오류로 보임"),
|
||
("중복에 속한 건", sum(len(v) for v in groups.values()), ""),
|
||
("침해의심 전체", suspected, ""),
|
||
(" 중복에서 비롯된 건", len(affected), "중복 제거 후 재실행하면 사라짐"),
|
||
(" 중복과 무관한 건", suspected - len(affected), "중복을 제거해도 남음"),
|
||
):
|
||
sheet.append([label, value, note])
|
||
|
||
sheet.append([])
|
||
sheet.append([f"중복에서 비롯된 침해의심 {len(affected)}건 — 사라지기 전 원문 보존"])
|
||
sheet.cell(row=sheet.max_row, column=1).font = Font(bold=True)
|
||
header_row = sheet.max_row + 1
|
||
sheet.append([label for label, _ in DUP_COLUMNS])
|
||
for index, (_, width) in enumerate(DUP_COLUMNS, start=1):
|
||
cell = sheet.cell(row=header_row, column=index)
|
||
cell.font = Font(bold=True)
|
||
cell.fill = HEADER_FILL
|
||
sheet.column_dimensions[get_column_letter(index)].width = width
|
||
for row in sorted(affected, key=lambda r: -(r.get("결합유사도") or 0)):
|
||
meta = records[row["query_key"]]
|
||
peers = sorted({records[r["query_key"]]["doc_id"] for r in member_of[row["query_key"]]})
|
||
sheet.append([
|
||
row["가명 제목"], meta["doc_id"], row.get("매칭 상대 제목") or "-",
|
||
meta.get("매칭 상대 doc") or "-", row.get("결합유사도"),
|
||
"예" if _norm(row["검사 대상 원문"]) == _norm(row.get("매칭 상대 원문", "")) else "아니오",
|
||
len(_norm(row["검사 대상 원문"])), ", ".join(peers),
|
||
row["검사 대상 원문"], row.get("매칭 상대 원문") or "-",
|
||
])
|
||
for column in (8, 9, 10):
|
||
sheet.cell(row=sheet.max_row, column=column).alignment = Alignment(
|
||
wrap_text=True, vertical="top")
|
||
sheet.cell(row=sheet.max_row, column=5).number_format = "0.0000"
|
||
|
||
sheet.append([])
|
||
for line in (
|
||
"중복 제거는 원본 코퍼스를 지우는 것이 아니라, 재실행 시 해당 세그먼트를 검사 대상에서 빼는 작업입니다.",
|
||
"'서로 다른 문서 사이'의 중복은 같은 제출자가 두 번 낸 것일 수도, 다른 제출자의 표절일 수도 있습니다.",
|
||
"후자라면 그 자체가 침해 사례이므로, 위 목록의 문서 출처를 확인한 뒤 판단해야 합니다.",
|
||
):
|
||
sheet.append([line])
|
||
sheet.cell(row=sheet.max_row, column=1).font = NOTE_FONT
|
||
|
||
|
||
|
||
TOTAL_DOC_PAIRS = 148240 # 545문서 전수 조합
|
||
|
||
|
||
def build_sweep_sheet(wb: Workbook, sweep: list[dict], rows: list[dict],
|
||
exhaustive: list[dict] | None = None) -> None:
|
||
"""어절 기준을 3~9로 바꿔가며 재판정한 결과.
|
||
|
||
기준값은 검색 후보를 바꾸지 않고 채택 여부만 정하므로, 한 번 실행해 남긴
|
||
후보 판정재료로 모든 기준을 다시 계산할 수 있다.
|
||
"""
|
||
sheet = wb.create_sheet("어절 기준 비교")
|
||
sheet.append(["어절 기준별 탐지 건수와 겹침률"])
|
||
sheet.cell(row=1, column=1).font = Font(bold=True, size=12)
|
||
for line in (
|
||
"연속 일치 기준을 3~9어절로 바꿔가며 다시 판정한 결과입니다.",
|
||
"겹침률은 서로 다른 문서의 원고를 임의로 15만 쌍 짝지어, 그 기준만큼 연속으로 겹치는 비율을 잰 값입니다.",
|
||
"임의 조합이라 표절 관계일 가능성이 낮으므로 오탐 지표로 봅니다. 다만 각 쌍이 실제로 무관한지 검증한 값은 아닙니다.",
|
||
):
|
||
sheet.append([line])
|
||
sheet.cell(row=sheet.max_row, column=1).font = NOTE_FONT
|
||
|
||
sheet.append([])
|
||
header = ["어절", "글자 수", "탐지 건수", "연속일치로만 걸린 건",
|
||
"임의 조합 겹침률", "6087건 대조 시 겹칠 확률", "판단"]
|
||
sheet.append(header)
|
||
for index in range(1, len(header) + 1):
|
||
cell = sheet.cell(row=sheet.max_row, column=index)
|
||
cell.font = Font(bold=True)
|
||
cell.fill = HEADER_FILL
|
||
for item in sweep:
|
||
cumulative = item["누적"]
|
||
verdict = ("사용 불가" if cumulative >= 90 else
|
||
"위험" if cumulative >= 30 else
|
||
"판단 필요" if cumulative >= 10 else "권장 구간")
|
||
sheet.append([
|
||
f"{item['어절']}어절", item["자"], item["탐지"], item["연속일치만"],
|
||
item["쌍당"] / 100, cumulative / 100, verdict,
|
||
])
|
||
sheet.cell(row=sheet.max_row, column=5).number_format = "0.00000%"
|
||
sheet.cell(row=sheet.max_row, column=6).number_format = "0.0%"
|
||
|
||
sheet.append([])
|
||
# 어절 기준을 아무리 바꿔도 유사도·커버리지만으로 걸리는 건은 그대로다.
|
||
base = sum(
|
||
1 for row in rows
|
||
if any(c["combined"] >= SCORE_MIN or c["coverage"] >= COVERAGE_MIN
|
||
for c in (row.get("후보 판정재료") or []))
|
||
)
|
||
sheet.append([f"기준과 무관하게 걸리는 건 (유사도 0.65 또는 커버리지 30% 충족): {base}건"])
|
||
sheet.cell(row=sheet.max_row, column=1).font = Font(bold=True)
|
||
if exhaustive:
|
||
sheet.append([])
|
||
sheet.append(["기준을 계속 올리면 겹침이 사라지는가 — 전수 확인"])
|
||
sheet.cell(row=sheet.max_row, column=1).font = Font(bold=True, size=12)
|
||
sheet.append([f"표본이 아니라 545개 문서의 전체 조합 {TOTAL_DOC_PAIRS:,}쌍을 모두 대조한 결과입니다."])
|
||
sheet.cell(row=sheet.max_row, column=1).font = NOTE_FONT
|
||
header2 = ["어절", "중복 제거 전 겹치는 문서쌍", "중복 제거 후", "전체 문서쌍 대비"]
|
||
sheet.append(header2)
|
||
for index in range(1, len(header2) + 1):
|
||
cell = sheet.cell(row=sheet.max_row, column=index)
|
||
cell.font = Font(bold=True)
|
||
cell.fill = HEADER_FILL
|
||
for item in exhaustive:
|
||
sheet.append([
|
||
f"{item['어절']}어절", item["제거전"], item["제거후"],
|
||
item["제거후"] / TOTAL_DOC_PAIRS,
|
||
])
|
||
sheet.cell(row=sheet.max_row, column=4).number_format = "0.00000%"
|
||
for line in (
|
||
"100어절(약 390자)까지 올려도 18개 문서쌍은 계속 겹칩니다. 겹침이 0이 되는 어절 수는 존재하지 않습니다.",
|
||
"10~15어절 구간에서 27쌍으로 고정되어 더 줄지 않습니다. 오탐이라면 기준을 올릴수록 계속 줄어야 합니다.",
|
||
"서로 다른 문서에 390자가 연속으로 같다는 것은 우연이 아니라 실제로 같은 내용이 들어 있다는 뜻입니다.",
|
||
"따라서 기준은 '겹침이 0이 되는 값'이 아니라, 상투어 노이즈가 걷히는 지점으로 정해야 합니다.",
|
||
"상투어 노이즈는 4→5어절에서 대부분 걷히고, 9어절부터는 남는 것이 전부 실제 유사 원고입니다.",
|
||
"이 근거로 9어절(35자)을 적용했습니다.",
|
||
):
|
||
sheet.append([line])
|
||
sheet.cell(row=sheet.max_row, column=1).font = NOTE_FONT
|
||
sheet.append([])
|
||
|
||
for line in (
|
||
"어절 기준을 아무리 낮춰도 이 건수는 줄지 않습니다. 어절 기준이 더하는 몫이 '연속일치로만 걸린 건'입니다.",
|
||
"3어절과 4어절은 임의 조합의 99% 이상이 겹쳐, 침해 신호로 쓸 수 없습니다. 겹친 표현은 '할 수 있는', '할 수 있을 것' 같은 상투어였습니다.",
|
||
"8어절과 9어절은 탐지 건수가 같습니다. 더 올려도 얻는 것이 없다는 뜻입니다.",
|
||
"겹침률이 0이 되는 지점은 9어절까지 없습니다. 따라서 '오탐이 사라지는 값'이 아니라 탐지력과 오탐의 교환으로 정해야 합니다.",
|
||
"정확한 오탐률은 사람이 표본을 판별해 정답을 만든 뒤에만 산출할 수 있습니다.",
|
||
):
|
||
sheet.append([line])
|
||
sheet.cell(row=sheet.max_row, column=1).font = NOTE_FONT
|
||
for index, width in enumerate((12, 10, 12, 20, 18, 24, 12), start=1):
|
||
sheet.column_dimensions[get_column_letter(index)].width = width
|
||
|
||
|
||
def load_hash_map(path: Path) -> dict:
|
||
"""segment_id → 원문 정보. 가명 해시를 제출 이메일·원문 제목으로 되돌린다."""
|
||
return json.loads(path.read_text(encoding="utf-8"))
|
||
|
||
|
||
def enrich_with_identity(rows: list[dict], hash_map: dict) -> None:
|
||
"""각 행에 제출자 이메일과 원문 제목을 붙인다.
|
||
|
||
매칭 상대는 세그먼트 ID가 비어 있는 경우가 있어 문서 단위로도 찾도록
|
||
doc → 원문 색인을 함께 만든다. 생활수기는 익명화된 상태로 수령해 이메일이
|
||
없으므로 '-' 로 남긴다.
|
||
"""
|
||
by_doc: dict[str, dict] = {}
|
||
for info in hash_map.values():
|
||
by_doc.setdefault(info["doc"], info)
|
||
for row in rows:
|
||
mine = hash_map.get(row.get("query_segment_id") or "") or by_doc.get(row.get("doc_id"))
|
||
peer = hash_map.get(row.get("매칭 상대 segment") or "") or by_doc.get(row.get("매칭 상대 doc"))
|
||
row["id"] = (mine or {}).get("email") or "-"
|
||
row["원문 자서전 제목"] = (mine or {}).get("book") or "-"
|
||
row["원문 에피소드 제목"] = (mine or {}).get("episode") or "-"
|
||
row["원본 엑셀 행"] = (mine or {}).get("row")
|
||
row["매칭 상대 id"] = (peer or {}).get("email") or "-"
|
||
row["매칭 상대 원문 제목"] = (peer or {}).get("book") or "-"
|
||
row["동일 이메일"] = (
|
||
"예" if row["id"] != "-" and row["id"] == row["매칭 상대 id"] else "아니오"
|
||
)
|
||
|
||
|
||
SCORE_MIN = 0.65
|
||
SPAN_MIN = 35
|
||
COVERAGE_MIN = 0.30
|
||
|
||
|
||
def judgment_reasons(row: dict, longest_span: int) -> list[str]:
|
||
"""어느 조건으로 의심 판정이 났는지. detector.py 의 OR 조건과 같은 기준.
|
||
|
||
엔진은 후보 하나하나에 조건을 적용한다. 행 단위 대표값(결합유사도·
|
||
union_coverage)으로 재현하면 서로 다른 후보가 각기 다른 조건을 충족한
|
||
경우를 놓치므로, 후보 판정재료가 있으면 그것을 쓴다.
|
||
"""
|
||
candidates = row.get("후보 판정재료")
|
||
if candidates:
|
||
reasons = []
|
||
if any(c["combined"] >= SCORE_MIN for c in candidates):
|
||
reasons.append("①유사도")
|
||
if any(c["longest"] >= SPAN_MIN for c in candidates):
|
||
reasons.append("②연속일치")
|
||
if any(c["coverage"] >= COVERAGE_MIN for c in candidates):
|
||
reasons.append("③커버리지")
|
||
return reasons
|
||
reasons = []
|
||
if (row.get("결합유사도") or 0) >= SCORE_MIN:
|
||
reasons.append("①유사도")
|
||
if longest_span >= SPAN_MIN:
|
||
reasons.append("②연속일치")
|
||
if (row.get("union_coverage") or 0) >= COVERAGE_MIN:
|
||
reasons.append("③커버리지")
|
||
return reasons
|
||
|
||
|
||
EXCERPT_COLUMNS = [
|
||
("id (제출자 이메일)", 26), ("매칭 상대 id (이메일)", 26), ("동일 이메일", 12),
|
||
("가명 제목", 20), ("침해 여부", 12), ("판정 사유", 26), ("매칭 상대 제목", 20),
|
||
("결합유사도", 11), ("일치 구간 길이", 13),
|
||
("일치 구간 (똑같은 문장)", 70), ("검사 대상 원문", 70), ("매칭 상대 원문", 70),
|
||
]
|
||
|
||
|
||
def build_excerpt_sheet(wb: Workbook, excerpts: list[dict], records: dict[str, dict]) -> None:
|
||
sheet = wb.create_sheet("원문 대조")
|
||
sheet.append([label for label, _ in EXCERPT_COLUMNS])
|
||
for index, (_, width) in enumerate(EXCERPT_COLUMNS, start=1):
|
||
cell = sheet.cell(row=1, column=index)
|
||
cell.font = Font(bold=True)
|
||
cell.fill = HEADER_FILL
|
||
sheet.column_dimensions[get_column_letter(index)].width = width
|
||
sheet.freeze_panes = "D2"
|
||
legend = (
|
||
"판정 사유: ①유사도 = 결합유사도 0.65 이상 / ②연속일치 = 똑같은 글자가 35자(9어절) 이상 이어짐 "
|
||
"/ ③커버리지 = 문서의 30% 이상이 겹침. 하나만 충족해도 침해 의심입니다."
|
||
)
|
||
order = {"침해 의심": 0, "매칭 미탐지": 1}
|
||
ordered = sorted(
|
||
excerpts,
|
||
key=lambda r: (order.get(r.get("침해 여부"), 9), -(r.get("결합유사도") or 0)),
|
||
)
|
||
for row in ordered:
|
||
spans = row.get("일치 구간") or []
|
||
longest = max((len(span) for span in spans), default=0)
|
||
record = records.get(row["query_key"], {})
|
||
if row.get("침해 여부") == "침해 의심":
|
||
reasons = " + ".join(judgment_reasons(record, longest)) or "(사유 미상)"
|
||
else:
|
||
reasons = "-"
|
||
sheet.append([
|
||
record.get("id") or "-", record.get("매칭 상대 id") or "-",
|
||
record.get("동일 이메일") or "-",
|
||
row.get("가명 제목"), row.get("침해 여부"), reasons,
|
||
row.get("매칭 상대 제목") or "-", row.get("결합유사도"), longest,
|
||
"\n---\n".join(spans) if spans else "(일치 구간 없음)", row.get("검사 대상 원문"),
|
||
row.get("매칭 상대 원문") or "(매칭된 상대 글 없음)",
|
||
])
|
||
for column in (10, 11, 12):
|
||
sheet.cell(row=sheet.max_row, column=column).alignment = Alignment(
|
||
wrap_text=True, vertical="top"
|
||
)
|
||
sheet.cell(row=sheet.max_row, column=8).number_format = "0.0000"
|
||
sheet.auto_filter.ref = f"A1:{get_column_letter(len(EXCERPT_COLUMNS))}{sheet.max_row}"
|
||
sheet.append([])
|
||
sheet.append([legend])
|
||
sheet.cell(row=sheet.max_row, column=1).font = NOTE_FONT
|
||
|
||
|
||
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차 비교")
|
||
# 중복 제거 회차는 검사 대상이 줄어 키 집합이 다르다. 양쪽에 다 있는
|
||
# 건만 비교해야 없는 키를 참조하지 않는다.
|
||
shared_keys = set(first) & set(second)
|
||
flagged_first = {k for k in shared_keys if first[k]["침해 의심"]}
|
||
flagged_second = {k for k in shared_keys if second[k]["침해 의심"]}
|
||
shared = sorted(shared_keys)
|
||
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="요약 시트에 남길 유의사항")
|
||
parser.add_argument("--hash-map", type=Path,
|
||
help="가명 해시를 제출 이메일·원문 제목으로 되돌릴 매핑 JSON")
|
||
parser.add_argument("--sweep-json", type=Path, help="어절 기준 비교 시트용 데이터")
|
||
parser.add_argument("--exhaustive-json", type=Path,
|
||
help="문서쌍 전수 대조 결과 (겹침이 0이 되는지 확인한 표)")
|
||
parser.add_argument("--dup-batch-jsonl", type=Path,
|
||
help="중복 원고 시트를 만들 때 참조할 이전 회차 배치 JSONL")
|
||
parser.add_argument("--no-excerpt-sheet", action="store_true",
|
||
help="원문 대조 시트는 만들지 않고 중복 원고 시트만 붙인다")
|
||
parser.add_argument("--excerpts-jsonl", type=Path,
|
||
help="원문 대조 시트를 붙일 발췌 JSONL (extract_suspected_excerpts.py 산출물)")
|
||
args = parser.parse_args()
|
||
|
||
records = load(args.jsonl)
|
||
rows = list(records.values())
|
||
backend = args.backend or str(rows[0].get("retrieval_backend") or "미기재")
|
||
if args.hash_map:
|
||
enrich_with_identity(rows, load_hash_map(args.hash_map))
|
||
|
||
wb = Workbook()
|
||
wb.remove(wb.active)
|
||
build_result_sheet(wb, rows)
|
||
build_detail_sheet(wb, rows)
|
||
excerpts = None
|
||
if args.excerpts_jsonl:
|
||
excerpts = [
|
||
json.loads(line)
|
||
for line in args.excerpts_jsonl.read_text(encoding="utf-8").splitlines()
|
||
if line.strip()
|
||
]
|
||
if not args.no_excerpt_sheet:
|
||
build_excerpt_sheet(wb, excerpts, records)
|
||
dup_records = load(args.dup_batch_jsonl) if args.dup_batch_jsonl else records
|
||
build_duplicate_sheet(wb, excerpts, dup_records)
|
||
if args.sweep_json:
|
||
build_sweep_sheet(
|
||
wb, json.loads(args.sweep_json.read_text(encoding="utf-8")), rows,
|
||
json.loads(args.exhaustive_json.read_text(encoding="utf-8"))
|
||
if args.exhaustive_json else None,
|
||
)
|
||
build_criteria_sheet(wb, rows, excerpts if args.excerpts_jsonl else None)
|
||
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())
|