312 lines
14 KiB
Python
312 lines
14 KiB
Python
#!/usr/bin/env python3
|
|
"""익명 런타임 코퍼스 전체의 규칙 기반 침해 검토 배치를 실행한다.
|
|
|
|
출력에는 원문·증거 문구를 저장하지 않고 가명 ID, 점수, 태그, 판례 ID, 건수만
|
|
기록한다. 각 쿼리는 같은 document_id 및 같은 익명 source_group 후보를 인덱스
|
|
순위 단계에서 제외한다. OpenAI 법률판단은 허용하지 않는다.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import json
|
|
import logging
|
|
import os
|
|
import sys
|
|
import time
|
|
from collections import Counter, defaultdict
|
|
from dataclasses import dataclass
|
|
from pathlib import Path
|
|
|
|
if __package__ in (None, ""):
|
|
sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
|
|
|
|
from openpyxl import Workbook
|
|
|
|
from app.api.schemas import DetectOptions
|
|
from app.core.config import get_settings
|
|
from app.engine.detector import PlagiarismDetector
|
|
from app.engine.provenance import CorpusStore, SegmentRecord
|
|
|
|
|
|
LOGGER = logging.getLogger("infringement_batch")
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class QueryItem:
|
|
query_key: str
|
|
document_id: str
|
|
title: str
|
|
source_group: str
|
|
source_kind: str
|
|
text: str
|
|
source_segment_id: str | None = None
|
|
|
|
|
|
def _env_is_false(value: str | None) -> bool:
|
|
return str(value or "").strip().lower() in {"", "0", "false", "no", "off"}
|
|
|
|
|
|
def _reconstruct_document(segments: list[SegmentRecord]) -> str:
|
|
"""중첩 청크(document offsets)를 원래의 익명화 후 문서로 되조립한다."""
|
|
output = ""
|
|
for segment in sorted(segments, key=lambda item: (item.char_start or 0, item.ordinal)):
|
|
start = int(segment.char_start or 0)
|
|
if start > len(output):
|
|
output += "\n" * (start - len(output))
|
|
overlap = max(0, len(output) - start)
|
|
output += segment.text[overlap:]
|
|
return output
|
|
|
|
|
|
def build_query_plan(store: CorpusStore) -> list[QueryItem]:
|
|
documents = {record.document_id: record for record in store.documents_with_metadata()}
|
|
grouped: dict[str, list[SegmentRecord]] = defaultdict(list)
|
|
for segment in store.iter_segments():
|
|
grouped[segment.document_id].append(segment)
|
|
|
|
items: list[QueryItem] = []
|
|
for document_id, document in documents.items():
|
|
metadata = document.metadata
|
|
source_group = str(
|
|
metadata.get("source_group") or metadata.get("author_group") or ""
|
|
).strip()
|
|
source_kind = str(metadata.get("corpus_kind") or metadata.get("provenance") or "unknown")
|
|
segments = grouped.get(document_id, [])
|
|
if source_kind == "life_writing":
|
|
text = _reconstruct_document(segments).strip()
|
|
if text:
|
|
items.append(QueryItem(
|
|
query_key=document_id,
|
|
document_id=document_id,
|
|
title=document.title,
|
|
source_group=source_group,
|
|
source_kind="life_writing",
|
|
text=text,
|
|
))
|
|
continue
|
|
|
|
# 수령 XLSX는 한 문서 안에 여러 독립 에피소드가 있으므로 각 에피소드가
|
|
# 보고서의 1건이다. document_id는 같은 책/작성자 자기매칭 제외에 쓴다.
|
|
for segment in segments:
|
|
if segment.text.strip():
|
|
items.append(QueryItem(
|
|
query_key=segment.segment_id,
|
|
document_id=document_id,
|
|
title=document.title,
|
|
source_group=source_group,
|
|
source_kind="autobiography_episode",
|
|
text=segment.text,
|
|
source_segment_id=segment.segment_id,
|
|
))
|
|
return sorted(items, key=lambda item: item.query_key)
|
|
|
|
|
|
def _join(values: list[str]) -> str:
|
|
return ", ".join(dict.fromkeys(value for value in values if value))
|
|
|
|
|
|
def result_row(item: QueryItem, response, *, retrieval_backend: str) -> dict:
|
|
top = response.matches[0] if response.matches else None
|
|
breakdown = top.score_breakdown if top and top.score_breakdown else None
|
|
tags = list(top.tags) if top else []
|
|
primary = _join([tag.tag for tag in tags if tag.role == "primary"])
|
|
secondary = _join([tag.tag for tag in tags if tag.role == "secondary"])
|
|
legal = response.legal_risk
|
|
return {
|
|
"query_key": item.query_key,
|
|
"doc_id": item.document_id,
|
|
"query_segment_id": item.source_segment_id or "",
|
|
"가명 제목": item.title,
|
|
"원천 구분": "생활수기" if item.source_kind == "life_writing" else "자서전 에피소드",
|
|
"침해 여부": "침해 의심" if response.is_infringement else "매칭 미탐지",
|
|
"침해 의심": bool(response.is_infringement),
|
|
"결합유사도": response.confidence,
|
|
"text 점수": breakdown.text_sim if breakdown else None,
|
|
"lemma 점수": breakdown.lemma_sim if breakdown else None,
|
|
"character 점수": breakdown.character_sim if breakdown else None,
|
|
"motif 점수": breakdown.motif_sim if breakdown else None,
|
|
"주 법령태그": primary,
|
|
"보조 법령태그": secondary,
|
|
"case_id": top.case_id if top else None,
|
|
"관련 판례 사건번호": _join(list(legal.precedent_ids) if legal else []),
|
|
"judgment_summary": legal.judgment_summary if legal else "",
|
|
"legal_judgment_method": legal.judgment_method if legal else "",
|
|
"evidence 스팬 수": sum(len(match.evidence_spans) for match in response.matches),
|
|
"매칭 상대 doc": (top.source_document_id or top.source_doc) if top else None,
|
|
"매칭 상대 segment": top.source_segment_id if top else None,
|
|
"매칭 상대 제목": top.source_title if top else None,
|
|
"union_coverage": response.score_semantics.union_coverage if response.score_semantics else None,
|
|
"retrieval_backend": retrieval_backend,
|
|
}
|
|
|
|
|
|
def load_completed(path: Path) -> set[str]:
|
|
if not path.exists():
|
|
return set()
|
|
completed: set[str] = set()
|
|
for line in path.read_text(encoding="utf-8").splitlines():
|
|
if line.strip():
|
|
completed.add(str(json.loads(line)["query_key"]))
|
|
return completed
|
|
|
|
|
|
def build_xlsx(rows: list[dict], path: Path, *, backend: str) -> None:
|
|
wb = Workbook(write_only=True)
|
|
detail = wb.create_sheet("건별 결과")
|
|
columns = [
|
|
"doc_id", "가명 제목", "원천 구분", "침해 여부", "결합유사도",
|
|
"text 점수", "lemma 점수", "character 점수", "motif 점수",
|
|
"주 법령태그", "보조 법령태그", "case_id", "관련 판례 사건번호",
|
|
"judgment_summary", "evidence 스팬 수", "매칭 상대 doc", "매칭 상대 제목",
|
|
"query_segment_id", "매칭 상대 segment", "union_coverage", "legal_judgment_method",
|
|
"retrieval_backend",
|
|
]
|
|
detail.append(columns)
|
|
for row in rows:
|
|
detail.append([row.get(column) for column in columns])
|
|
|
|
distributions = wb.create_sheet("태그_케이스 분포")
|
|
distributions.append(["구분", "항목", "전체 건수", "침해 의심 건수", "침해 의심 비율"])
|
|
counters: dict[str, Counter[str]] = {
|
|
"주 법령태그": Counter(), "보조 법령태그": Counter(), "case_id": Counter(),
|
|
}
|
|
flagged: dict[str, Counter[str]] = {
|
|
"주 법령태그": Counter(), "보조 법령태그": Counter(), "case_id": Counter(),
|
|
}
|
|
for row in rows:
|
|
for category in counters:
|
|
for value in str(row.get(category) or "").split(", "):
|
|
if not value:
|
|
continue
|
|
counters[category][value] += 1
|
|
if row["침해 의심"]:
|
|
flagged[category][value] += 1
|
|
for category in ("주 법령태그", "보조 법령태그", "case_id"):
|
|
for value, count in sorted(counters[category].items(), key=lambda item: (-item[1], item[0])):
|
|
suspected = flagged[category][value]
|
|
distributions.append([category, value, count, suspected, suspected / count])
|
|
|
|
summary = wb.create_sheet("요약 통계")
|
|
summary.append(["구분", "총 건수", "침해 의심 건수", "침해 의심 비율"])
|
|
groups = {
|
|
"전체": rows,
|
|
"자서전 에피소드": [row for row in rows if row["원천 구분"] == "자서전 에피소드"],
|
|
"생활수기": [row for row in rows if row["원천 구분"] == "생활수기"],
|
|
}
|
|
for label, group_rows in groups.items():
|
|
total = len(group_rows)
|
|
suspected = sum(bool(row["침해 의심"]) for row in group_rows)
|
|
summary.append([label, total, suspected, suspected / total if total else 0])
|
|
summary.append([])
|
|
summary.append(["실행 조건", "값"])
|
|
summary.append(["법률 판단", "rule_based (USE_LLM_LEGAL_JUDGE=false)"])
|
|
summary.append(["검색 백엔드", backend])
|
|
summary.append(["GPU 사용", "아니오 — persistent hashing char 3-4 (CPU TF-IDF 계열) 폴백"])
|
|
summary.append(["원문 저장", "결과 파일에는 저장하지 않음"])
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
wb.save(path)
|
|
|
|
|
|
def main() -> int:
|
|
parser = argparse.ArgumentParser(description=__doc__)
|
|
parser.add_argument("--database", type=Path, required=True)
|
|
parser.add_argument("--index-dir", type=Path, required=True)
|
|
parser.add_argument("--out-jsonl", type=Path, required=True)
|
|
parser.add_argument("--out-xlsx", type=Path, required=True)
|
|
parser.add_argument("--progress-every", type=int, default=50)
|
|
parser.add_argument("--resume", action="store_true")
|
|
parser.add_argument("--shard-count", type=int, default=1)
|
|
parser.add_argument("--shard-index", type=int, default=0)
|
|
parser.add_argument(
|
|
"--no-xlsx", action="store_true",
|
|
help="분할 작업용: JSONL만 쓰고 XLSX 요약은 병합 단계에서 생성",
|
|
)
|
|
args = parser.parse_args()
|
|
logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s")
|
|
if not _env_is_false(os.environ.get("USE_LLM_LEGAL_JUDGE")):
|
|
raise RuntimeError("USE_LLM_LEGAL_JUDGE=false 이어야 원고 외부 전송 없이 실행할 수 있습니다")
|
|
if not args.database.exists():
|
|
parser.error(f"database does not exist: {args.database}")
|
|
if args.shard_count < 1 or not 0 <= args.shard_index < args.shard_count:
|
|
parser.error("shard-index는 0 이상이고 shard-count보다 작아야 합니다")
|
|
|
|
get_settings.cache_clear()
|
|
settings = get_settings().model_copy(update={
|
|
"corpus_db_path": str(args.database),
|
|
"persistent_index_dir": str(args.index_dir),
|
|
})
|
|
if settings.use_llm_legal_judge or settings.has_llm_legal_judge:
|
|
raise RuntimeError("LLM 법률판단 설정이 켜져 있어 배치를 중단합니다")
|
|
detector = PlagiarismDetector(settings)
|
|
if not detector.uses_persistent_index:
|
|
raise RuntimeError("영속 인덱스가 준비되지 않아 배치를 중단합니다")
|
|
store = CorpusStore(args.database)
|
|
all_items = build_query_plan(store)
|
|
items = [
|
|
item for position, item in enumerate(all_items)
|
|
if position % args.shard_count == args.shard_index
|
|
]
|
|
completed = load_completed(args.out_jsonl) if args.resume else set()
|
|
if args.out_jsonl.exists() and not args.resume:
|
|
args.out_jsonl.unlink()
|
|
args.out_jsonl.parent.mkdir(parents=True, exist_ok=True)
|
|
pending = [item for item in items if item.query_key not in completed]
|
|
backend = "persistent-hashing-char-3-4 (CPU; GPU embedding backend not enabled)"
|
|
LOGGER.info(
|
|
"batch_start total=%d pending=%d resumed=%d backend=%s legal_judge=rule_based",
|
|
len(items), len(pending), len(completed), backend,
|
|
)
|
|
started = time.monotonic()
|
|
source_groups = store.document_source_groups()
|
|
options = DetectOptions(top_k=5, return_evidence=True)
|
|
with args.out_jsonl.open("a", encoding="utf-8") as stream:
|
|
for position, item in enumerate(pending, 1):
|
|
excluded_groups = {item.source_group} if item.source_group else set()
|
|
response = detector.detect(
|
|
item.query_key,
|
|
item.text,
|
|
options=options,
|
|
include_ai_segments=False,
|
|
include_ai_detection=False,
|
|
exclude_document_ids={item.document_id},
|
|
exclude_source_groups=excluded_groups,
|
|
)
|
|
if response.legal_risk and response.legal_risk.judgment_method != "rule_based":
|
|
raise RuntimeError("규칙 기반 이외의 법률판단 결과가 감지되어 중단합니다")
|
|
if response.matches:
|
|
matched_doc = response.matches[0].source_document_id
|
|
if matched_doc == item.document_id or (
|
|
item.source_group and source_groups.get(matched_doc or "") == item.source_group
|
|
):
|
|
raise RuntimeError("자기매칭 제외 검증 실패")
|
|
row = result_row(item, response, retrieval_backend=backend)
|
|
stream.write(json.dumps(row, ensure_ascii=False) + "\n")
|
|
if position % 10 == 0:
|
|
stream.flush()
|
|
if position % args.progress_every == 0 or position == len(pending):
|
|
elapsed = max(time.monotonic() - started, 0.001)
|
|
done = len(completed) + position
|
|
rate = position / elapsed
|
|
remaining = (len(pending) - position) / rate if rate else 0
|
|
LOGGER.info(
|
|
"progress processed=%d/%d elapsed_sec=%.1f eta_sec=%.1f",
|
|
done, len(items), elapsed, remaining,
|
|
)
|
|
stream.flush()
|
|
|
|
rows = [json.loads(line) for line in args.out_jsonl.read_text(encoding="utf-8").splitlines() if line.strip()]
|
|
if len(rows) != len(items):
|
|
raise RuntimeError(f"결과 행 수 불일치: {len(rows)} != {len(items)}")
|
|
if args.no_xlsx:
|
|
LOGGER.info("batch_shard_completed shard=%d/%d total=%d jsonl=%s", args.shard_index, args.shard_count, len(rows), args.out_jsonl)
|
|
return 0
|
|
build_xlsx(rows, args.out_xlsx, backend=backend)
|
|
suspected = sum(bool(row["침해 의심"]) for row in rows)
|
|
LOGGER.info("batch_completed total=%d suspected=%d jsonl=%s xlsx=%s", len(rows), suspected, args.out_jsonl, args.out_xlsx)
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|