o2o-plagiarism-ai/scripts/build_duplicate_report.py

"""코퍼스 안의 중복 원고를 찾아 목록과 제외 대상 세그먼트를 만든다.

중복 제거 후 재실행하면 중복에서 비롯된 침해의심 건이 결과에서 사라진다.
사라지기 전에 그 건들이 어떤 문서 사이의 중복이었는지 남겨두어야, 나중에
같은 제출자의 중복인지 다른 제출자의 표절인지 확인할 수 있다.

산출물
  - XLSX : 중복 그룹 목록과, 그 중복 때문에 침해의심이 된 건의 원문
  - TXT  : 재실행 시 제외할 segment_id 목록 (그룹마다 첫 건만 남긴다)
"""
from __future__ import annotations

import argparse
import hashlib
import json
import re
from collections import defaultdict
from pathlib import Path

from openpyxl import Workbook
from openpyxl.styles import Alignment, Font, PatternFill
from openpyxl.utils import get_column_letter

HEADER_FILL = PatternFill("solid", fgColor="D9E1F2")
NOTE_FONT = Font(italic=True, color="7F7F7F")


def normalize(text: str) -> str:
    """공백 차이만 있는 원고를 같은 것으로 본다."""
    return re.sub(r"\s+", "", text or "")


def text_key(text: str) -> str:
    return hashlib.sha256(normalize(text).encode("utf-8")).hexdigest()[:16]


def load_jsonl(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 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
        sheet.column_dimensions[get_column_letter(index)].width = width
    sheet.freeze_panes = "A2"


GROUP_COLUMNS = [
    ("중복 그룹", 10), ("중복 건수", 10), ("문서 수", 9), ("성격", 22),
    ("가명 제목들", 40), ("문서 ID들", 60), ("본문 길이", 10), ("본문", 80),
]

SUSPECT_COLUMNS = [
    ("가명 제목", 20), ("문서 ID", 28), ("매칭 상대 제목", 20), ("매칭 상대 문서 ID", 28),
    ("결합유사도", 11), ("본문 완전 동일", 13), ("본문 길이", 10),
    ("검사 대상 원문", 70), ("매칭 상대 원문", 70),
]


def main() -> int:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--excerpts-jsonl", type=Path, required=True)
    parser.add_argument("--batch-jsonl", type=Path, required=True)
    parser.add_argument("--out-xlsx", type=Path, required=True)
    parser.add_argument("--out-exclude", type=Path, required=True,
                        help="재실행 시 제외할 segment_id 목록 (한 줄에 하나)")
    args = parser.parse_args()

    excerpts = load_jsonl(args.excerpts_jsonl)
    batch = load_jsonl(args.batch_jsonl)

    groups: dict[str, list[str]] = defaultdict(list)
    for key, row in excerpts.items():
        if normalize(row["검사 대상 원문"]):
            groups[text_key(row["검사 대상 원문"])].append(key)
    duplicates = {key: keys for key, keys in groups.items() if len(keys) > 1}

    # 그룹마다 첫 건만 남기고 나머지를 제외 대상으로 삼는다.
    excluded = [key for keys in duplicates.values() for key in sorted(keys)[1:]]
    segment_ids = [batch[key].get("query_segment_id") or batch[key]["doc_id"] for key in excluded]
    args.out_exclude.parent.mkdir(parents=True, exist_ok=True)
    args.out_exclude.write_text("\n".join(segment_ids) + "\n", encoding="utf-8")

    wb = Workbook()
    wb.remove(wb.active)

    # 1) 중복 때문에 침해의심이 된 건 — 사라지기 전에 원문째로 남긴다.
    sheet = wb.create_sheet("중복 기인 의심 건")
    write_header(sheet, SUSPECT_COLUMNS)
    affected = [
        key for key, row in excerpts.items()
        if row["침해 여부"] == "침해 의심" and len(groups[text_key(row["검사 대상 원문"])]) > 1
    ]
    for key in sorted(affected, key=lambda k: -(excerpts[k]["결합유사도"] or 0)):
        row, meta = excerpts[key], batch[key]
        identical = normalize(row["검사 대상 원문"]) == normalize(row["매칭 상대 원문"])
        sheet.append([
            row["가명 제목"], meta["doc_id"], row.get("매칭 상대 제목") or "-",
            meta.get("매칭 상대 doc") or "-", row["결합유사도"],
            "예" if identical else "아니오", len(normalize(row["검사 대상 원문"])),
            row["검사 대상 원문"], row["매칭 상대 원문"],
        ])
        for column in (8, 9):
            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.auto_filter.ref = f"A1:{get_column_letter(len(SUSPECT_COLUMNS))}{sheet.max_row}"

    # 2) 중복 그룹 전체 목록
    sheet = wb.create_sheet("중복 그룹 전체")
    write_header(sheet, GROUP_COLUMNS)
    for index, (_, keys) in enumerate(
        sorted(duplicates.items(), key=lambda item: (-len(item[1]), item[0])), start=1
    ):
        docs = sorted({batch[key]["doc_id"] for key in keys})
        titles = sorted({excerpts[key]["가명 제목"] for key in keys})
        body = excerpts[keys[0]]["검사 대상 원문"]
        sheet.append([
            index, len(keys), len(docs),
            "서로 다른 문서 사이" if len(docs) > 1 else "같은 문서 안 (적재 오류)",
            ", ".join(titles), ", ".join(docs), len(normalize(body)), body,
        ])
        sheet.cell(row=sheet.max_row, column=8).alignment = Alignment(
            wrap_text=True, vertical="top")
    sheet.auto_filter.ref = f"A1:{get_column_letter(len(GROUP_COLUMNS))}{sheet.max_row}"

    # 3) 요약
    sheet = wb.create_sheet("요약")
    cross = sum(1 for keys in duplicates.values()
                if len({batch[key]["doc_id"] for key in keys}) > 1)
    for label, value in (
        ("검사 대상 전체", len(excerpts)),
        ("고유 본문", len(groups)),
        ("중복 그룹", len(duplicates)),
        ("  서로 다른 문서 사이", cross),
        ("  같은 문서 안 (적재 오류)", len(duplicates) - cross),
        ("중복에 속한 건", sum(len(keys) for keys in duplicates.values())),
        ("재실행 시 제외할 건", len(excluded)),
        ("제외 후 검사 대상", len(excerpts) - len(excluded)),
        ("", ""),
        ("침해의심 전체", sum(1 for r in excerpts.values() if r["침해 여부"] == "침해 의심")),
        ("  중복에서 비롯된 건", len(affected)),
        ("  중복과 무관한 건",
         sum(1 for r in excerpts.values() if r["침해 여부"] == "침해 의심") - len(affected)),
    ):
        sheet.append([label, value])
    sheet.append([])
    for line in (
        "중복 제거는 원본 코퍼스를 지우는 것이 아니라, 재실행 시 해당 세그먼트를 검사 대상에서 빼는 것입니다.",
        "'서로 다른 문서 사이'의 중복은 같은 제출자가 두 번 낸 것일 수도, 다른 제출자의 표절일 수도 있습니다.",
        "후자라면 그것이 곧 침해 사례이므로, 이 목록의 문서 출처를 확인한 뒤 판단해야 합니다.",
    ):
        sheet.append([line])
        sheet.cell(row=sheet.max_row, column=1).font = NOTE_FONT
    sheet.column_dimensions["A"].width = 30
    sheet.column_dimensions["B"].width = 14

    args.out_xlsx.parent.mkdir(parents=True, exist_ok=True)
    wb.save(args.out_xlsx)
    print(f"wrote {args.out_xlsx}")
    print(f"wrote {args.out_exclude} ({len(segment_ids)}건 제외 대상)")
    print(f"중복 그룹 {len(duplicates)} / 중복 기인 의심 건 {len(affected)}")
    return 0


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