From c35c5aabfc716781dbd507a7fdaf6035cef13040 Mon Sep 17 00:00:00 2001 From: hbyang Date: Wed, 2 Sep 2026 09:23:45 +0900 Subject: [PATCH] feat: add anonymized infringement batch reporting --- app/engine/detector.py | 59 +++--- app/engine/persistent_index.py | 60 +++++- app/engine/provenance.py | 40 ++++ scripts/ingest_life_writing.py | 123 +++++++++++++ scripts/run_infringement_batch.py | 296 ++++++++++++++++++++++++++++++ tests/test_provenance_index.py | 25 +++ 6 files changed, 573 insertions(+), 30 deletions(-) create mode 100644 scripts/ingest_life_writing.py create mode 100644 scripts/run_infringement_batch.py diff --git a/app/engine/detector.py b/app/engine/detector.py index a18b364..678f87d 100644 --- a/app/engine/detector.py +++ b/app/engine/detector.py @@ -204,6 +204,10 @@ class PlagiarismDetector: options: DetectOptions | None = None, include_ai_segments: bool = True, legal_context: LegalContext | None = None, + *, + exclude_document_ids: set[str] | None = None, + exclude_source_groups: set[str] | None = None, + include_ai_detection: bool = True, ) -> DetectResponse: opts = options or DetectOptions() default_threshold = ( @@ -242,6 +246,8 @@ class PlagiarismDetector: text, top_k=max(opts.top_k, self.settings.persistent_rerank_top_k), evidence_limit=self.settings.persistent_rerank_top_k, + exclude_document_ids=exclude_document_ids, + exclude_source_groups=exclude_source_groups, ) persistent_hits = result.hits union_coverage = result.union_coverage @@ -341,29 +347,36 @@ class PlagiarismDetector: ) # AI 탐지는 전처리 전 raw text에서만 실행한다. - ai_result = self._ai_detector.detect(text, with_segments=include_ai_segments) - ai_signal = AiGenerationSignal( - suspicion_level=ai_result.suspicion_level or "unknown", - score=ai_result.score, - available=ai_result.available, - provenance=ai_result.provenance, - is_stub=ai_result.is_stub, - model_version=ai_result.model_version, - feature_set_version=ai_result.feature_set_version, - pos_available=ai_result.pos_available, - warnings=ai_result.warnings, - segments=[{ - "index": s.index, "start": s.start, "end": s.end, - "char_count": s.char_count, "score": s.score, - "suspicion_level": s.suspicion_level, "scored": s.scored, - "note": s.note, - } for s in ai_result.segments], - top_contributions=[ - {"feature": name, "contribution": round(value, 4)} - for name, value in ai_result.top_contributions - ], - note=ai_result.note, - ) + if include_ai_detection: + ai_result = self._ai_detector.detect(text, with_segments=include_ai_segments) + ai_signal = AiGenerationSignal( + suspicion_level=ai_result.suspicion_level or "unknown", + score=ai_result.score, + available=ai_result.available, + provenance=ai_result.provenance, + is_stub=ai_result.is_stub, + model_version=ai_result.model_version, + feature_set_version=ai_result.feature_set_version, + pos_available=ai_result.pos_available, + warnings=ai_result.warnings, + segments=[{ + "index": s.index, "start": s.start, "end": s.end, + "char_count": s.char_count, "score": s.score, + "suspicion_level": s.suspicion_level, "scored": s.scored, + "note": s.note, + } for s in ai_result.segments], + top_contributions=[ + {"feature": name, "contribution": round(value, 4)} + for name, value in ai_result.top_contributions + ], + note=ai_result.note, + ) + else: + ai_signal = AiGenerationSignal( + suspicion_level="unknown", + available=False, + note="침해 판별 배치에서는 AI 생성 의심도 채점을 생략했습니다.", + ) sim_pct = _calibrate_similarity(confidence, threshold) review = ReviewSummary( originality_percent=100 - sim_pct, diff --git a/app/engine/persistent_index.py b/app/engine/persistent_index.py index 3f0fbe3..f1ce381 100644 --- a/app/engine/persistent_index.py +++ b/app/engine/persistent_index.py @@ -18,7 +18,9 @@ import numpy as np from app.engine.provenance import CorpusStore, SegmentRecord -INDEX_VERSION = 1 +# v2: segment → document 및 문서 → 익명 작성자 그룹 메타를 함께 보관한다. +# 배치에서 자기 문서/같은 작성자의 후보를 후보 순위 단계부터 제외하기 위함이다. +INDEX_VERSION = 2 VECTORIZER_CONFIG = { "analyzer": "char_wb", "ngram_range": [3, 4], @@ -158,7 +160,10 @@ class PersistentCorpusIndex: return False try: meta = json.loads(meta_path.read_text(encoding="utf-8")) - return self._matrix_path(meta).exists() + return ( + meta.get("version") == INDEX_VERSION + and self._matrix_path(meta).exists() + ) except (OSError, ValueError, json.JSONDecodeError): return False @@ -214,6 +219,7 @@ class PersistentCorpusIndex: mode = "append" by_id = {s.segment_id: s for s in segments} + document_source_groups = self.store.document_source_groups() existing_set = set(existing_ids) new_ids = [s.segment_id for s in segments if s.segment_id not in existing_set] vectorizer = _vectorizer(n_features) @@ -238,6 +244,8 @@ class PersistentCorpusIndex: "n_features": n_features, "vectorizer_config": VECTORIZER_CONFIG, "segment_ids": ids, + "segment_document_ids": [by_id[i].document_id for i in ids], + "document_source_groups": document_source_groups, "text_hashes": {i: current[i] for i in ids}, "document_count": self.store.document_count(), "matrix_file": matrix_file, @@ -251,10 +259,22 @@ class PersistentCorpusIndex: self._matrix, self._meta = matrix, meta return {"mode": mode, "total": len(ids), "added": len(new_ids)} - def query(self, text: str, top_k: int = 50, min_score: float = 0.0, - evidence_limit: int | None = None) -> list[PersistentHit]: + def query( + self, + text: str, + top_k: int = 50, + min_score: float = 0.0, + evidence_limit: int | None = None, + *, + exclude_document_ids: set[str] | None = None, + exclude_source_groups: set[str] | None = None, + ) -> list[PersistentHit]: """후보만 필요할 때 쓰는 얇은 래퍼. union coverage 가 필요하면 search().""" - return self.search(text, top_k, min_score, evidence_limit).hits + return self.search( + text, top_k, min_score, evidence_limit, + exclude_document_ids=exclude_document_ids, + exclude_source_groups=exclude_source_groups, + ).hits def search( self, @@ -262,6 +282,9 @@ class PersistentCorpusIndex: top_k: int = 50, min_score: float = 0.0, evidence_limit: int | None = None, + *, + exclude_document_ids: set[str] | None = None, + exclude_source_groups: set[str] | None = None, ) -> PersistentQueryResult: """후보 검색 + 상위 evidence_limit 개에 대해서만 정밀 비교. @@ -296,8 +319,31 @@ class PersistentCorpusIndex: improved = local_scores > scores scores[improved] = local_scores[improved] best_chunk[improved] = batch_start + local_argmax[improved] - k = min(max(1, top_k), len(scores)) - indexes = np.argpartition(scores, -k)[-k:] + # 자기 문서와 같은 익명 작성자 그룹을 여기서 제외한다. 상위 K를 먼저 + # 뽑아 뒤늦게 제거하면 자기매칭이 실제 후보를 밀어내므로, 전량 점수에서 + # 허용 후보만 골라 순위를 만든다. + excluded_documents = set(exclude_document_ids or ()) + excluded_groups = set(exclude_source_groups or ()) + document_ids = self._meta.get("segment_document_ids", []) + if len(document_ids) != len(scores): + raise ValueError("인덱스 문서 메타가 없어 재빌드가 필요합니다") + source_groups = self._meta.get("document_source_groups", {}) + eligible = np.fromiter( + ( + document_id not in excluded_documents + and source_groups.get(document_id, "") not in excluded_groups + for document_id in document_ids + ), + dtype=bool, + count=len(document_ids), + ) + eligible_indexes = np.flatnonzero(eligible) + if not len(eligible_indexes): + return PersistentQueryResult([], 0.0, 0, len(text), evidence_limit or 0, False, {}) + k = min(max(1, top_k), len(eligible_indexes)) + eligible_scores = scores[eligible_indexes] + selected = np.argpartition(eligible_scores, -k)[-k:] + indexes = eligible_indexes[selected] indexes = indexes[np.argsort(scores[indexes])[::-1]] ids = [self._meta["segment_ids"][int(i)] for i in indexes if scores[int(i)] >= min_score] records = self.store.get_segments(ids) diff --git a/app/engine/provenance.py b/app/engine/provenance.py index 089ec29..27cf82a 100644 --- a/app/engine/provenance.py +++ b/app/engine/provenance.py @@ -315,6 +315,46 @@ class CorpusStore: ).fetchall() return [dict(row) for row in rows] + def document_source_groups(self) -> dict[str, str]: + """문서별 익명 작성자 그룹을 반환한다. + + ``author_group`` 은 수령 XLSX 적재기의 기존 필드이고, ``source_group`` 은 + 새 수집 원문·학습 데이터에서 쓰는 공통 필드다. 둘 다 가명 값만 허용한다. + 영속 인덱스가 자기 작성자의 후보를 검색 단계에서 제외할 때 사용한다. + """ + if not self.path.exists(): + return {} + with self._connect() as con: + rows = con.execute( + "SELECT document_id, metadata_json FROM documents" + ).fetchall() + groups: dict[str, str] = {} + for row in rows: + metadata = json.loads(row["metadata_json"] or "{}") + group = str( + metadata.get("source_group") or metadata.get("author_group") or "" + ).strip() + if group: + groups[str(row["document_id"])] = group + return groups + + def documents_with_metadata(self) -> list[DocumentRecord]: + """배치·감사용 문서 목록. 원문 세그먼트는 포함하지 않는다.""" + if not self.path.exists(): + return [] + with self._connect() as con: + rows = con.execute( + """SELECT document_id, title, source_path, source_sha256, metadata_json + FROM documents ORDER BY document_id""" + ).fetchall() + return [DocumentRecord( + document_id=str(row["document_id"]), + title=str(row["title"]), + source_path=row["source_path"], + source_sha256=row["source_sha256"], + metadata=json.loads(row["metadata_json"] or "{}"), + ) for row in rows] + def delete_document(self, document_id: str) -> bool: if not self.path.exists(): return False diff --git a/scripts/ingest_life_writing.py b/scripts/ingest_life_writing.py new file mode 100644 index 0000000..9a63131 --- /dev/null +++ b/scripts/ingest_life_writing.py @@ -0,0 +1,123 @@ +#!/usr/bin/env python3 +"""로컬 생활수기 TXT를 익명화해 런타임 코퍼스에 적재한다. + +원본 파일명·작성자 이름은 SQLite에 넣지 않는다. 파일별 문서/작성자 그룹은 +운영 salt 기반 가명으로만 보존하고, 본문의 이메일·전화·주민번호 형식은 제거한다. +""" + +from __future__ import annotations + +import argparse +import hashlib +import json +import os +import sys +from pathlib import Path + +if __package__ in (None, ""): + sys.path.insert(0, str(Path(__file__).resolve().parents[1])) + +from app.engine.provenance import CorpusStore, DocumentRecord, SegmentRecord, stable_id +from app.engine.source_extraction import chunk_with_offsets +from app.engine.training_data import pseudonymous_id, redact_direct_identifiers + + +def file_sha256(path: Path) -> str: + digest = hashlib.sha256() + with path.open("rb") as stream: + for block in iter(lambda: stream.read(1024 * 1024), b""): + digest.update(block) + return digest.hexdigest() + + +def main() -> int: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("source_dir", type=Path) + parser.add_argument("--database", type=Path, required=True) + parser.add_argument("--anonymize", action="store_true") + parser.add_argument("--anonymization-salt-env", default="DATA_ANONYMIZATION_SALT") + parser.add_argument("--provenance", default="life_writing") + parser.add_argument("--chunk-size", type=int, default=1000) + parser.add_argument("--stride", type=int, default=500) + args = parser.parse_args() + if not args.source_dir.is_dir(): + parser.error(f"source directory does not exist: {args.source_dir}") + if not args.anonymize: + parser.error("생활수기 적재에는 --anonymize 가 필요합니다") + salt = os.environ.get(args.anonymization_salt_env, "") + if not salt: + parser.error(f"{args.anonymization_salt_env} 환경변수가 필요합니다") + + files = sorted(path for path in args.source_dir.glob("*.txt") if path.is_file()) + if not files: + parser.error(".txt 원문이 없습니다") + store = CorpusStore(args.database) + documents: list[DocumentRecord] = [] + segments: list[SegmentRecord] = [] + skipped_empty = 0 + redaction_changes = 0 + + for path in files: + raw = path.read_text(encoding="utf-8", errors="replace") + clean = redact_direct_identifiers(raw).strip() + if not clean: + skipped_empty += 1 + continue + raw_hash = file_sha256(path) + document_id = pseudonymous_id(f"life-writing:{raw_hash}", salt, prefix="doc") + source_group = pseudonymous_id(f"life-writing:{raw_hash}", salt, prefix="source") + display_title = f"익명 생활수기 {document_id.split(':', 1)[-1][:8]}" + documents.append(DocumentRecord( + document_id=document_id, + title=display_title, + metadata={ + "provenance": args.provenance, + "corpus_kind": "life_writing", + "source_group": source_group, + "human_verified": True, + "ai_assistance": False, + "coordinate_scope": "document", + "anonymized": True, + }, + )) + if clean != raw.strip(): + redaction_changes += 1 + for ordinal, (start, end, chunk) in enumerate( + chunk_with_offsets(clean, size=args.chunk_size, stride=args.stride), 1 + ): + segments.append(SegmentRecord( + segment_id=stable_id("seg", document_id, str(start), chunk), + document_id=document_id, + text=chunk, + ordinal=str(ordinal), + coordinate_scope="document", + char_start=start, + char_end=end, + source_locator=f"life_writing://{document_id}#chars={start}-{end}", + metadata={ + "provenance": args.provenance, + "corpus_kind": "life_writing", + "human_verified": True, + "ai_assistance": False, + "provenance_quality": "text_extraction_document_offsets", + }, + )) + + store.upsert_documents(documents) + inserted, duplicates = store.add_segments(segments) + print(json.dumps({ + "files": len(files), + "documents": len(documents), + "inserted_segments": inserted, + "duplicate_segments": duplicates, + "skipped_empty_files": skipped_empty, + "files_with_direct_identifier_redaction": redaction_changes, + "anonymized": True, + "provenance": args.provenance, + "store": store.stats(), + }, ensure_ascii=False, indent=2)) + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/scripts/run_infringement_batch.py b/scripts/run_infringement_batch.py new file mode 100644 index 0000000..d88726b --- /dev/null +++ b/scripts/run_infringement_batch.py @@ -0,0 +1,296 @@ +#!/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") + 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}") + + 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) + items = build_query_plan(store) + 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)}") + 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()) diff --git a/tests/test_provenance_index.py b/tests/test_provenance_index.py index 2dbd579..306d3bb 100644 --- a/tests/test_provenance_index.py +++ b/tests/test_provenance_index.py @@ -67,3 +67,28 @@ def test_long_query_finds_middle_copy_without_dilution(tmp_path): assert hit.segment_id == "seg-1" assert hit.longest_span >= 100 assert hit.evidence[0]["start"] > 1000 + + +def test_index_excludes_same_document_and_source_group_before_ranking(tmp_path): + db = tmp_path / "corpus.sqlite3" + store = CorpusStore(db) + copied = "바닷가 마을에서 파도 소리를 들으며 자란 기억이 아직도 선명하다." + store.upsert_documents([ + DocumentRecord(document_id="doc-self", title="나", metadata={"author_group": "author:a"}), + DocumentRecord(document_id="doc-same-author", title="같은 작성자", metadata={"source_group": "author:a"}), + DocumentRecord(document_id="doc-other", title="다른 작성자", metadata={"source_group": "author:b"}), + ]) + store.add_segments([ + SegmentRecord("seg-self", "doc-self", copied, "1"), + SegmentRecord("seg-same", "doc-same-author", copied, "1"), + SegmentRecord("seg-other", "doc-other", copied + " 다른 결말입니다.", "1"), + ]) + index = PersistentCorpusIndex(db, tmp_path / "index") + index.sync() + + hits = index.query( + copied, top_k=5, + exclude_document_ids={"doc-self"}, + exclude_source_groups={"author:a"}, + ) + assert [hit.document_id for hit in hits] == ["doc-other"]