From 725e44e9992b7c73c1e1437918cd7da4efed9b7d Mon Sep 17 00:00:00 2001 From: hbyang Date: Mon, 10 Aug 2026 16:42:17 +0900 Subject: [PATCH] =?UTF-8?q?Claude=20=EB=A6=AC=EB=B7=B0=20=EA=B8=B0?= =?UTF-8?q?=EB=B0=98=20=EC=9A=B4=EC=98=81=20=EC=95=88=EC=A0=95=EC=84=B1=20?= =?UTF-8?q?=EB=B3=B4=EA=B0=95?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .env.example | 6 + Dockerfile | 1 + app/api/schemas.py | 70 +++++- app/core/config.py | 15 +- app/engine/detector.py | 126 +++++++++-- app/engine/persistent_index.py | 135 +++++++++-- app/engine/provenance.py | 112 ++++++--- app/main.py | 57 ++++- docs/IMPLEMENTATION_RUNBOOK.md | 72 ++++++ scripts/build_persistent_index.py | 42 ++++ tests/test_ai_training_split.py | 167 ++++++++++++++ tests/test_api.py | 66 ++++++ tests/test_auth_middleware.py | 144 ++++++++++++ tests/test_coverage_and_features.py | 337 ++++++++++++++++++++++++++++ tests/test_review_regressions.py | 144 ++++++++++++ 15 files changed, 1427 insertions(+), 67 deletions(-) create mode 100644 tests/test_ai_training_split.py create mode 100644 tests/test_auth_middleware.py create mode 100644 tests/test_coverage_and_features.py create mode 100644 tests/test_review_regressions.py diff --git a/.env.example b/.env.example index 8e4563f..cd522ff 100644 --- a/.env.example +++ b/.env.example @@ -7,6 +7,12 @@ RELOAD=false ROOT_PATH= # 설정하면 /v1/health를 제외한 API 요청에 X-API-Key가 필요합니다. API_KEY= +# 운영 fail-closed. true인데 API_KEY가 비면 앱이 기동에 실패합니다. +# ⚠️ King 실제 활성화는 바이칼 측 키 전달 후 진행 (docs/AI_DETECTION.md 참조). +REQUIRE_API_KEY=false +# 인증 없이 열어둘 경로 제어 +PUBLIC_HEALTH=true +PUBLIC_DOCS=true ENGINE_VERSION=o2o-plagiarism-2.1.0-kosimcse REFERENCE_CORPUS_DIR=./data/reference diff --git a/Dockerfile b/Dockerfile index 8e94b13..28529dc 100644 --- a/Dockerfile +++ b/Dockerfile @@ -5,6 +5,7 @@ WORKDIR /app ENV PYTHONDONTWRITEBYTECODE=1 \ PYTHONUNBUFFERED=1 \ PIP_NO_CACHE_DIR=1 \ + PYTHONPATH=/app \ HOST=0.0.0.0 \ PORT=8000 diff --git a/app/api/schemas.py b/app/api/schemas.py index 576a487..fbc7057 100644 --- a/app/api/schemas.py +++ b/app/api/schemas.py @@ -55,11 +55,33 @@ class DetectOptions(BaseModel): ) +class LegalContext(BaseModel): + """사람이 확인한 법적 사실. 제공하지 않으면 missing_factors 로 남는다 (#10). + + 엔진은 이 값들을 **추론하지 않는다.** 텍스트만으로는 알 수 없는 사실이므로, + 검토자가 확인한 경우에만 전달받아 판례 매칭과 위험도 판단에 반영한다. + """ + work_type: str = Field( + default="literary", description="저작물 유형 (literary/musical/visual 등)" + ) + access_evidence: bool | None = Field( + default=None, + description="원저작물 접근·의거 가능성이 확인되었는지. None이면 미제공.", + ) + protected_expression_reviewed: bool = Field( + default=False, description="보호되는 창작적 표현인지 사람이 검토했는지", + ) + rights_verified: bool = Field( + default=False, description="저작권 귀속·이용허락·인용 요건이 확인되었는지", + ) + + class DetectRequest(BaseModel): doc_id: str text: str = Field(..., min_length=1) metadata: DocumentMetadata | None = None options: DetectOptions = Field(default_factory=DetectOptions) + legal_context: LegalContext | None = None class EvidenceSpan(BaseModel): @@ -117,8 +139,18 @@ class MatchResult(BaseModel): paragraph_number: int | None = None source_char_start: int | None = None source_char_end: int | None = None - matched_coverage: float = Field(default=0.0, ge=0.0, le=1.0) + matched_coverage: float = Field( + default=0.0, ge=0.0, le=1.0, + description="이 세그먼트 일치 구간이 질의 전체 길이에서 차지하는 비율", + ) longest_span: int = Field(default=0, ge=0) + match_reasons: list[str] = Field( + default_factory=list, + description=( + "이 후보가 채택된 이유 (#9). score_threshold=결합점수가 임계 초과, " + "exact_span=연속 일치 길이 조건 충족, coverage=커버리지 조건 충족." + ), + ) class ExtractedElements(BaseModel): @@ -142,6 +174,41 @@ class LegalRiskSignal(BaseModel): disclaimer: str +class ScoreSemantics(BaseModel): + """점수와 임계값의 의미를 명시 (#9). + + ``confidence``/``similarity`` 는 hashing 어휘 점수와 lemma 겹침을 설정 + 가중치로 섞은 **검색 랭킹 점수**이며, 침해 확률도 법적 판정도 아니다. + 임계값 역시 실데이터 캘리브레이션 전이라 provisional 이다. + """ + combined_score: float = Field(..., ge=0.0, le=1.0) + score_kind: str = Field( + default="lexical_lemma_blend", + description="점수 구성. 확률값이 아니며 서로 다른 코퍼스 간 비교 불가.", + ) + threshold_used: float = Field(..., ge=0.0, le=1.0) + threshold_source: Literal["request_override", "server_default"] + threshold_calibrated: bool = Field( + default=False, description="실데이터 FP 분포로 캘리브레이션되었는지", + ) + provisional: bool = Field( + default=True, description="True면 임계값이 잠정값이라 판정 근거로 쓸 수 없음", + ) + union_coverage: float = Field( + default=0.0, ge=0.0, le=1.0, + description="정밀 비교 후보 전체의 비중복 일치 구간 / 질의 길이 (#3)", + ) + covered_chars: int = Field(default=0, ge=0) + query_chars: int = Field(default=0, ge=0) + evidence_truncated: bool = Field( + default=False, + description="True면 CPU 상한으로 일부 후보는 정밀 비교하지 않음", + ) + note: str = ( + "검색 랭킹 점수이며 침해 확률이 아닙니다. 임계값은 캘리브레이션 전 잠정값입니다." + ) + + class AiSegmentSignal(BaseModel): index: int start: int @@ -203,6 +270,7 @@ class DetectResponse(BaseModel): has_similarity_match: bool | None = None corpus_scope_note: str | None = None legal_risk: LegalRiskSignal | None = None + score_semantics: ScoreSemantics | None = None autobiography_mode: bool = False candidates_before_filter: int | None = None engine_version: str diff --git a/app/core/config.py b/app/core/config.py index c77c3c0..faf4466 100644 --- a/app/core/config.py +++ b/app/core/config.py @@ -13,7 +13,15 @@ class Settings(BaseSettings): log_level: str = "info" # debug / info / warning / error reload: bool = False # 개발용 자동 재시작 root_path: str = "" # 리버스 프록시 sub-path (예: /plagiarism) - api_key: str = "" # 설정 시 /v1/health 외 X-API-Key 필수 + + # --- 인증 (#1) --- + # api_key 만으로는 "빈 값 = 무인증"이 조용히 성립한다. 운영에서는 + # require_api_key=true 로 명시적 fail-closed 를 걸어, 키가 없으면 앱이 + # 아예 뜨지 않게 한다. + api_key: str = "" # 설정 시 X-API-Key 필수 + require_api_key: bool = False # true 인데 api_key 가 비면 기동 실패 + public_health: bool = True # /v1/health 를 인증 없이 공개할지 + public_docs: bool = True # /docs, /openapi.json, /redoc 공개 여부 engine_version: str = "o2o-plagiarism-2.0.0-pdf-v1.2" reference_corpus_dir: str = "./data/reference" @@ -25,6 +33,8 @@ class Settings(BaseSettings): persistent_similarity_threshold: float = 0.65 persistent_min_exact_span: int = 80 persistent_min_coverage: float = 0.30 + # 상위 N개 후보만 SequenceMatcher 정밀 비교 + union coverage 에 참여시킨다. + # 이 값이 곧 요청당 O(질의길이 × 세그먼트길이) 연산의 상한이다 (#3/#5). persistent_rerank_top_k: int = 20 precedents_path: str = "./data/precedents/precedents.jsonl" ai_detector_model_path: str = "./data/models/ai_detector.joblib" @@ -33,6 +43,9 @@ class Settings(BaseSettings): # PDF VII-4 권장: 정밀도 우선 보수적 임계값 similarity_threshold: float = 0.85 + # 임계값이 실데이터로 캘리브레이션되었는지. false 면 API 응답에 + # provisional=true 로 노출된다 (#9). 79권 FP 분포 측정 후 true 로 전환. + similarity_threshold_calibrated: bool = False # KoSimCSE / KoSBERT (PDF VII-3 권장) - 한국어 오픈소스 임베딩 use_kosimcse: bool = False diff --git a/app/engine/detector.py b/app/engine/detector.py index 4b34c31..1459708 100644 --- a/app/engine/detector.py +++ b/app/engine/detector.py @@ -23,13 +23,16 @@ from app.api.schemas import ( DetectRequest, DetectResponse, DocumentMetadata, + ExtractedElements, InfringementTag, InfringementType, + LegalContext, LegalRiskSignal, MatchResult, PartialPlagiarismSignal, ReviewSummary, ScoreBreakdown, + ScoreSemantics, ) from app.core.config import Settings, get_settings from app.engine.autobiography_filter import preprocess_for_autobiography @@ -54,8 +57,12 @@ logger = logging.getLogger(__name__) class PlagiarismDetector: + #: 참조 특징 프로세스 캐시 상한. 초과하면 통째로 비운다(단순 LRU 대용). + _FEATURE_CACHE_MAX = 20_000 + def __init__(self, settings: Settings | None = None, extractor: Extractor | None = None): self.settings = settings or get_settings() + self._feature_cache: dict[str, tuple[list[str], "ExtractedElements"]] = {} self._extractor: Extractor = extractor or get_extractor(self.settings) self.taxonomy: Taxonomy | None = load_taxonomy(self.settings.taxonomy_path) self._legal_engine = LegalRiskEngine(load_precedents(self.settings.precedent_path)) @@ -174,6 +181,7 @@ class PlagiarismDetector: metadata: DocumentMetadata | None = None, options: DetectOptions | None = None, include_ai_segments: bool = True, + legal_context: LegalContext | None = None, ) -> DetectResponse: opts = options or DetectOptions() default_threshold = ( @@ -202,18 +210,30 @@ class PlagiarismDetector: # 영속 CPU 인덱스: 전량 행렬곱으로 후보를 구하고 상위 후보만 증거 비교한다. persistent_hits: list[PersistentHit] = [] + union_coverage = 0.0 + covered_chars = 0 + evidence_truncated = False + document_coverage: dict[str, float] = {} if self._persistent: persistent_query_lemmas = extract_lemmas(text) - persistent_hits = self._persistent.query( + result = self._persistent.search( text, top_k=max(opts.top_k, self.settings.persistent_rerank_top_k), + evidence_limit=self.settings.persistent_rerank_top_k, ) + persistent_hits = result.hits + union_coverage = result.union_coverage + covered_chars = result.covered_chars + evidence_truncated = result.evidence_truncated + document_coverage = result.document_coverage + # 정밀 특징 비교도 rerank 대상(reranked=True)에만 수행한다 (#5). hits = [ self._persistent_to_similarity_hit(h, persistent_query_lemmas, elements) - for h in persistent_hits + for h in persistent_hits if h.reranked ] + self._backfill_segment_features(persistent_hits) hits.sort(key=lambda h: h.score, reverse=True) - candidates_count = len(hits) + candidates_count = len(persistent_hits) lsh_jaccards: dict[str, float] = {} else: hits = [] @@ -240,14 +260,22 @@ class PlagiarismDetector: persistent_by_id = {h.segment_id: h for h in persistent_hits} matches = [] for h in hits: + if len(matches) >= opts.top_k: + break provenance_hit = persistent_by_id.get(h.doc_id) - exact_partial_match = bool( - provenance_hit and ( - provenance_hit.longest_span >= self.settings.persistent_min_exact_span - or provenance_hit.coverage >= self.settings.persistent_min_coverage - ) - ) - if (h.score < threshold and not exact_partial_match) or len(matches) >= opts.top_k: + # 채택 이유를 명시적으로 남긴다 (#9). 임계 초과가 아니라 연속 일치나 + # 커버리지 때문에 올라온 후보를 검토자가 구분할 수 있어야 한다. + reasons: list[str] = [] + if h.score >= threshold: + reasons.append("score_threshold") + if provenance_hit: + if provenance_hit.longest_span >= self.settings.persistent_min_exact_span: + reasons.append("exact_span") + # 커버리지 조건은 문서 단위 union 으로 판단한다. 세그먼트 단독 + # 비율은 긴 원고에서 구조적으로 작아 조건이 성립하지 않는다 (#3). + if document_coverage.get(provenance_hit.document_id, 0.0) >= self.settings.persistent_min_coverage: + reasons.append("coverage") + if not reasons: continue match = self._to_match( h, opts.return_evidence, lsh_jaccards.get(h.doc_id), @@ -255,20 +283,25 @@ class PlagiarismDetector: ) if provenance_hit: match = self._add_provenance(match, provenance_hit) - matches.append(match) + matches.append(match.model_copy(update={"match_reasons": reasons})) confidence = matches[0].similarity if matches else (hits[0].score if hits else 0.0) is_infringement = bool(matches) # 후방호환 필드. 법적 확정이 아니라 임계 초과 매칭. ccl_basis = self._build_ccl_basis(matches) if is_infringement else None - top_coverage = max((m.matched_coverage for m in matches), default=0.0) top_longest = max((m.longest_span for m in matches), default=0) legal_tags = [t.tag for m in matches for t in m.tags] + ctx = legal_context or LegalContext() legal = self._legal_engine.assess( max_similarity=matches[0].similarity if matches else 0.0, - coverage=top_coverage, + # 문서 단위 union coverage 를 쓴다. 세그먼트 최대값을 쓰면 긴 원고에서 + # 항상 0에 가까워 strong_copy 판정이 성립하지 않았다 (#3). + coverage=union_coverage, longest_span=top_longest, legal_tags=legal_tags, - work_type="literary", + work_type=ctx.work_type, + access_evidence=ctx.access_evidence, + protected_expression_reviewed=ctx.protected_expression_reviewed, + rights_verified=ctx.rights_verified, ) # AI 탐지는 전처리 전 raw text에서만 실행한다. @@ -329,6 +362,19 @@ class PlagiarismDetector: precedent_ids=list(legal.precedent_ids), disclaimer=legal.disclaimer, ), + score_semantics=ScoreSemantics( + combined_score=round(confidence, 4), + threshold_used=threshold, + threshold_source=( + "request_override" if opts.threshold is not None else "server_default" + ), + threshold_calibrated=self.settings.similarity_threshold_calibrated, + provisional=not self.settings.similarity_threshold_calibrated, + union_coverage=round(union_coverage, 4), + covered_chars=covered_chars, + query_chars=len(text), + evidence_truncated=evidence_truncated, + ), autobiography_mode=autobio_mode, candidates_before_filter=candidates_count, engine_version=self.settings.engine_version, @@ -336,7 +382,53 @@ class PlagiarismDetector: ) def detect_request(self, req: DetectRequest) -> DetectResponse: - return self.detect(req.doc_id, req.text, req.metadata, req.options) + return self.detect( + req.doc_id, req.text, req.metadata, req.options, + legal_context=req.legal_context, + ) + + def reference_features(self, hit: PersistentHit) -> tuple[list[str], ExtractedElements]: + """참조 세그먼트의 lemma/요소. 인덱스 캐시 → 프로세스 캐시 → 계산 순 (#5). + + 인덱싱 때 채워둔 값이 있으면 형태소 분석을 아예 하지 않는다. v1 DB 나 + API 업로드분처럼 캐시가 없으면 계산하되 프로세스 캐시에 담고, 호출자가 + DB 로 백필한다. + """ + from app.api.schemas import ExtractedElements as _EE + + if hit.reference_lemmas is not None and hit.reference_elements is not None: + return hit.reference_lemmas, _EE(**hit.reference_elements) + + cached = self._feature_cache.get(hit.segment_id) + if cached is not None: + return cached + + lemmas = extract_lemmas(hit.reference_text) + elements = self._extractor.extract(hit.reference_text) + if len(self._feature_cache) >= self._FEATURE_CACHE_MAX: + self._feature_cache.clear() + self._feature_cache[hit.segment_id] = (lemmas, elements) + return lemmas, elements + + def _backfill_segment_features(self, hits: list[PersistentHit]) -> None: + """질의 중 계산한 참조 특징을 DB 에 되돌려 다음 요청부터 재사용.""" + if not self._persistent: + return + pending = [ + (h.segment_id, *self._feature_cache[h.segment_id]) + for h in hits + if h.reranked + and h.reference_lemmas is None + and h.segment_id in self._feature_cache + ] + if not pending: + return + try: + self._persistent.store.update_segment_features( + (sid, lemmas, elements.model_dump()) for sid, lemmas, elements in pending + ) + except Exception as exc: # 백필 실패가 탐지 응답을 막아서는 안 된다 + logger.warning("Segment feature backfill failed: %s", exc) def _persistent_to_similarity_hit( self, hit: PersistentHit, query_lemmas: list[str], query_elements, @@ -345,9 +437,9 @@ class PlagiarismDetector: from app.engine.similarity import _element_similarities from app.engine.structural import lemma_overlap_ratio - reference_elements = self._extractor.extract(hit.reference_text) + reference_lemmas, reference_elements = self.reference_features(hit) element_sim = _element_similarities(query_elements, reference_elements) - lemma_sim = lemma_overlap_ratio(query_lemmas, extract_lemmas(hit.reference_text)) + lemma_sim = lemma_overlap_ratio(query_lemmas, reference_lemmas) s = self.settings combined = ( s.weight_text_sim * hit.score diff --git a/app/engine/persistent_index.py b/app/engine/persistent_index.py index b8fd12f..3f0fbe3 100644 --- a/app/engine/persistent_index.py +++ b/app/engine/persistent_index.py @@ -28,6 +28,29 @@ VECTORIZER_CONFIG = { } +@dataclass(frozen=True) +class PersistentQueryResult: + """질의 1건의 후보와 **문서 단위 union coverage**. + + coverage 정의 (#3): + · ``PersistentHit.coverage`` — 이 세그먼트 하나가 질의 전체 길이에서 + 차지하는 비율. 분모가 질의 전체라 긴 원고에서는 필연적으로 작다. + · ``PersistentQueryResult.union_coverage`` — 정밀 비교 대상 후보 + 전체(evidence_limit 개)의 일치 구간을 **질의 좌표에서 합집합**으로 + 묶어 계산한 비율. 중복 구간을 두 번 세지 않는다. + "이 원고의 몇 %가 등록 코퍼스와 겹치는가"에 답하는 값은 이쪽이며, + 법적 위험도 판단에는 반드시 이 값을 쓴다. + """ + + hits: list["PersistentHit"] + union_coverage: float + covered_chars: int + query_chars: int + evidence_limit: int + evidence_truncated: bool + document_coverage: dict[str, float] + + @dataclass(frozen=True) class PersistentHit: segment_id: str @@ -44,6 +67,12 @@ class PersistentHit: source_char_start: int | None source_char_end: int | None reference_text: str + #: 인덱싱 때 캐시된 참조 특징 (#5). None 이면 호출자가 계산·백필해야 한다. + reference_lemmas: list[str] | None = None + reference_elements: dict | None = None + #: SequenceMatcher 정밀 비교를 실제로 수행했는지. False 면 evidence/coverage/ + #: longest_span 은 미계산(0) 이며 score 만 의미가 있다. + reranked: bool = True def _vectorizer(n_features: int, config: dict | None = None): @@ -61,10 +90,17 @@ def _vectorizer(n_features: int, config: dict | None = None): ) -def _evidence_spans(query: str, reference: str, min_match: int = 12, limit: int = 10) -> tuple[list[dict], float, int]: - """원문 query 좌표의 공통 연속 구간과 coverage를 반환.""" +def _evidence_spans( + query: str, reference: str, min_match: int = 12, limit: int = 10 +) -> tuple[list[dict], list[tuple[int, int]], int]: + """원문 query 좌표의 공통 연속 구간을 반환. + + 반환: (표시용 상위 span, coverage 계산용 전체 구간 [start,end), 최장 일치 길이) + 두 번째 값은 union coverage 를 위해 **잘리지 않은 전체 구간**이다. 표시용은 + limit 개로 줄이지만 coverage 는 전량으로 계산해야 과소평가되지 않는다. + """ if not query or not reference: - return [], 0.0, 0 + return [], [], 0 blocks = SequenceMatcher(None, query, reference, autojunk=False).get_matching_blocks() useful = [b for b in blocks if b.size >= min_match] useful.sort(key=lambda b: (-b.size, b.a)) @@ -79,15 +115,35 @@ def _evidence_spans(query: str, reference: str, min_match: int = 12, limit: int } for b in selected ] - covered: set[int] = set() - for b in useful: - covered.update(range(b.a, b.a + b.size)) - return spans, len(covered) / max(1, len(query)), max((b.size for b in useful), default=0) + intervals = [(b.a, b.a + b.size) for b in useful] + return spans, intervals, max((b.size for b in useful), default=0) + + +def _merge_intervals(intervals: list[tuple[int, int]]) -> list[tuple[int, int]]: + """겹치는 구간을 합쳐 비중복 구간 목록으로. set(range(...)) 는 긴 원고에서 + 메모리를 크게 먹으므로 구간 병합으로 처리한다.""" + if not intervals: + return [] + ordered = sorted(intervals) + merged = [ordered[0]] + for start, end in ordered[1:]: + last_start, last_end = merged[-1] + if start <= last_end: + merged[-1] = (last_start, max(last_end, end)) + else: + merged.append((start, end)) + return merged + + +def _covered_length(intervals: list[tuple[int, int]]) -> int: + return sum(end - start for start, end in _merge_intervals(intervals)) class PersistentCorpusIndex: MATRIX_FILE = "lexical.npz" META_FILE = "index.json" + #: 정밀 비교(SequenceMatcher) 기본 상한. 호출자가 설정값으로 덮어쓴다. + DEFAULT_EVIDENCE_LIMIT = 20 def __init__(self, store_path: str | Path, index_dir: str | Path): self.store = CorpusStore(store_path) @@ -195,11 +251,28 @@ 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) -> list[PersistentHit]: + def query(self, text: str, top_k: int = 50, min_score: float = 0.0, + evidence_limit: int | None = None) -> list[PersistentHit]: + """후보만 필요할 때 쓰는 얇은 래퍼. union coverage 가 필요하면 search().""" + return self.search(text, top_k, min_score, evidence_limit).hits + + def search( + self, + text: str, + top_k: int = 50, + min_score: float = 0.0, + evidence_limit: int | None = None, + ) -> PersistentQueryResult: + """후보 검색 + 상위 evidence_limit 개에 대해서만 정밀 비교. + + SequenceMatcher 는 O(질의청크 × 세그먼트) 라 후보 전체에 돌리면 요청당 + 수십 초가 된다. 정밀 비교 대상을 evidence_limit 로 제한하는 것이 CPU + 상한이며, 나머지 후보는 score 만 채워 reranked=False 로 표시한다. + """ if self._matrix is None: self.load() if not text.strip() or self._matrix is None or self._matrix.shape[0] == 0: - return [] + return PersistentQueryResult([], 0.0, 0, len(text), evidence_limit or 0, False, {}) vectorizer = _vectorizer( int(self._meta["n_features"]), self._meta["vectorizer_config"] ) @@ -228,7 +301,14 @@ class PersistentCorpusIndex: 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) + + limit = self.DEFAULT_EVIDENCE_LIMIT if evidence_limit is None else evidence_limit + limit = max(0, limit) hits: list[PersistentHit] = [] + union_intervals: list[tuple[int, int]] = [] + document_intervals: dict[str, list[tuple[int, int]]] = {} + reranked_count = 0 + for i in indexes: score = float(scores[int(i)]) if score < min_score: @@ -237,18 +317,42 @@ class PersistentCorpusIndex: record = records.get(segment_id) if not record: continue + + if reranked_count >= limit: + # CPU 상한. 정밀 비교 없이 score 만 채운다. + hits.append(self._to_hit(record, score, [], 0.0, 0, reranked=False)) + continue + chunk_start, chunk_text = query_chunks[int(best_chunk[int(i)])] - evidence, _, longest = _evidence_spans(chunk_text, record.text) + evidence, intervals, longest = _evidence_spans(chunk_text, record.text) for span in evidence: span["start"] += chunk_start span["end"] += chunk_start - covered = sum(span["end"] - span["start"] for span in evidence) - coverage = min(1.0, covered / max(1, len(text))) + shifted = [(s + chunk_start, e + chunk_start) for s, e in intervals] + union_intervals.extend(shifted) + document_intervals.setdefault(record.document_id, []).extend(shifted) + # 이 세그먼트 단독 기여분 (질의 전체 길이 대비) + coverage = min(1.0, _covered_length(shifted) / max(1, len(text))) hits.append(self._to_hit(record, score, evidence, coverage, longest)) - return hits + reranked_count += 1 + + covered_chars = _covered_length(union_intervals) + return PersistentQueryResult( + hits=hits, + union_coverage=min(1.0, covered_chars / max(1, len(text))), + covered_chars=covered_chars, + query_chars=len(text), + evidence_limit=limit, + evidence_truncated=any(not h.reranked for h in hits), + document_coverage={ + document_id: min(1.0, _covered_length(intervals) / max(1, len(text))) + for document_id, intervals in document_intervals.items() + }, + ) @staticmethod - def _to_hit(record: SegmentRecord, score: float, evidence: list[dict], coverage: float, longest: int) -> PersistentHit: + def _to_hit(record: SegmentRecord, score: float, evidence: list[dict], + coverage: float, longest: int, reranked: bool = True) -> PersistentHit: return PersistentHit( segment_id=record.segment_id, document_id=record.document_id, @@ -264,4 +368,7 @@ class PersistentCorpusIndex: source_char_start=record.char_start, source_char_end=record.char_end, reference_text=record.text, + reference_lemmas=record.lemmas, + reference_elements=record.elements, + reranked=reranked, ) diff --git a/app/engine/provenance.py b/app/engine/provenance.py index 34be0d4..089ec29 100644 --- a/app/engine/provenance.py +++ b/app/engine/provenance.py @@ -15,7 +15,9 @@ from pathlib import Path from typing import Iterable, Iterator -SCHEMA_VERSION = 1 +#: v2 — 참조 lemma/요소 특징 캐시 컬럼 추가 (#5). 기존 v1 DB 는 ALTER TABLE 로 +#: 자동 승격되며, 캐시가 비어 있으면 질의 시 계산 후 백필된다. +SCHEMA_VERSION = 2 def text_sha256(text: str) -> str: @@ -49,6 +51,11 @@ class SegmentRecord: char_end: int | None = None source_locator: str | None = None metadata: dict = field(default_factory=dict) + #: 인덱싱 시점에 계산해 둔 참조 lemma 열 (#5). None 이면 미계산 상태이며, + #: 질의 경로가 계산 후 백필한다. 형태소 분석을 요청마다 반복하지 않기 위한 캐시. + lemmas: list[str] | None = None + #: 인물/모티프 등 최소 요소 특징. ExtractedElements 를 dict 로 직렬화한 형태. + elements: dict | None = None @property def text_sha256(self) -> str: @@ -110,11 +117,20 @@ class CorpusStore: CREATE INDEX IF NOT EXISTS idx_segments_hash ON segments(text_sha256); """ ) + self._migrate(con) con.execute( "INSERT OR REPLACE INTO corpus_meta(key, value) VALUES('schema_version', ?)", (str(SCHEMA_VERSION),), ) + @staticmethod + def _migrate(con: sqlite3.Connection) -> None: + """기존 DB 를 파괴 없이 승격. v1 → v2 는 컬럼 추가만 필요하다.""" + existing = {row["name"] for row in con.execute("PRAGMA table_info(segments)")} + for column in ("lemmas_json", "elements_json"): + if column not in existing: + con.execute(f"ALTER TABLE segments ADD COLUMN {column} TEXT") + def upsert_document(self, record: DocumentRecord) -> None: self.upsert_documents([record]) @@ -150,8 +166,9 @@ class CorpusStore: """ INSERT OR IGNORE INTO segments( segment_id,document_id,ordinal,text,text_sha256,coordinate_scope, - page_number,paragraph_number,char_start,char_end,source_locator,metadata_json - ) VALUES(?,?,?,?,?,?,?,?,?,?,?,?) + page_number,paragraph_number,char_start,char_end,source_locator, + metadata_json,lemmas_json,elements_json + ) VALUES(?,?,?,?,?,?,?,?,?,?,?,?,?,?) """, ( r.segment_id, @@ -166,6 +183,9 @@ class CorpusStore: r.char_end, r.source_locator, json.dumps(r.metadata, ensure_ascii=False, sort_keys=True), + json.dumps(r.lemmas, ensure_ascii=False) if r.lemmas is not None else None, + json.dumps(r.elements, ensure_ascii=False, sort_keys=True) + if r.elements is not None else None, ), ) if con.total_changes > before: @@ -186,21 +206,66 @@ class CorpusStore: """ ) for row in rows: - metadata = json.loads(row["metadata_json"] or "{}") - metadata.setdefault("document_title", row["title"]) - yield SegmentRecord( - segment_id=row["segment_id"], - document_id=row["document_id"], - text=row["text"], - ordinal=row["ordinal"], - coordinate_scope=row["coordinate_scope"], - page_number=row["page_number"], - paragraph_number=row["paragraph_number"], - char_start=row["char_start"], - char_end=row["char_end"], - source_locator=row["source_locator"], - metadata=metadata, - ) + yield self._row_to_segment(row) + + @staticmethod + def _row_to_segment(row: sqlite3.Row) -> SegmentRecord: + metadata = json.loads(row["metadata_json"] or "{}") + metadata.setdefault("document_title", row["title"]) + keys = row.keys() + lemmas = None + elements = None + # v1 DB 를 그대로 읽는 경로에서도 죽지 않도록 컬럼 존재를 확인한다. + if "lemmas_json" in keys and row["lemmas_json"]: + lemmas = json.loads(row["lemmas_json"]) + if "elements_json" in keys and row["elements_json"]: + elements = json.loads(row["elements_json"]) + return SegmentRecord( + segment_id=row["segment_id"], + document_id=row["document_id"], + text=row["text"], + ordinal=row["ordinal"], + coordinate_scope=row["coordinate_scope"], + page_number=row["page_number"], + paragraph_number=row["paragraph_number"], + char_start=row["char_start"], + char_end=row["char_end"], + source_locator=row["source_locator"], + metadata=metadata, + lemmas=lemmas, + elements=elements, + ) + + def update_segment_features( + self, features: Iterable[tuple[str, list[str], dict]] + ) -> int: + """(segment_id, lemmas, elements) 를 백필한다. 반환값은 갱신된 행 수.""" + rows = [ + ( + json.dumps(lemmas, ensure_ascii=False), + json.dumps(elements, ensure_ascii=False, sort_keys=True), + segment_id, + ) + for segment_id, lemmas, elements in features + ] + if not rows: + return 0 + self.initialize() + with self._connect() as con: + con.executemany( + "UPDATE segments SET lemmas_json=?, elements_json=? WHERE segment_id=?", + rows, + ) + return con.total_changes + + def count_missing_features(self) -> int: + if not self.path.exists(): + return 0 + self.initialize() + with self._connect() as con: + return int(con.execute( + "SELECT COUNT(*) FROM segments WHERE lemmas_json IS NULL" + ).fetchone()[0]) def get_segments(self, segment_ids: Iterable[str]) -> dict[str, SegmentRecord]: ids = list(dict.fromkeys(segment_ids)) @@ -218,16 +283,7 @@ class CorpusStore: batch, ) for row in rows: - metadata = json.loads(row["metadata_json"] or "{}") - metadata.setdefault("document_title", row["title"]) - result[row["segment_id"]] = SegmentRecord( - segment_id=row["segment_id"], document_id=row["document_id"], - text=row["text"], ordinal=row["ordinal"], - coordinate_scope=row["coordinate_scope"], - page_number=row["page_number"], paragraph_number=row["paragraph_number"], - char_start=row["char_start"], char_end=row["char_end"], - source_locator=row["source_locator"], metadata=metadata, - ) + result[row["segment_id"]] = self._row_to_segment(row) return result def stats(self) -> dict[str, int]: diff --git a/app/main.py b/app/main.py index 2c18dc4..14e7034 100644 --- a/app/main.py +++ b/app/main.py @@ -19,12 +19,60 @@ from app.jobs.store import JobStore logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(name)s: %(message)s") +class AuthConfigurationError(RuntimeError): + """운영 인증 설정이 모순될 때 기동을 막는다.""" + + +def validate_auth_settings(settings) -> None: + """REQUIRE_API_KEY=true 인데 키가 없으면 기동 실패 (fail-closed). + + '키를 깜빡해서 무인증으로 떠 있었다'가 가능한 구성을 없애는 것이 목적이다. + 미출간 원고를 다루는 서버라 조용한 무인증이 가장 위험하다. + """ + if settings.require_api_key and not settings.api_key.strip(): + raise AuthConfigurationError( + "REQUIRE_API_KEY=true 인데 API_KEY 가 비어 있습니다. " + "키를 설정하거나 REQUIRE_API_KEY=false 로 두십시오." + ) + + +def public_paths(settings) -> set[str]: + """인증 없이 접근 가능한 경로. 설정으로 좁힐 수 있다.""" + paths = {"/"} + if settings.public_health: + paths.add("/v1/health") + if settings.public_docs: + paths.update({"/docs", "/openapi.json", "/redoc"}) + return paths + + +def is_authorized(settings, path: str, supplied: str) -> bool: + """요청 허용 여부. 순수 함수라 앱 기동 없이 테스트할 수 있다. + + api_key 가 비어 있으면(개발 기본값) 인증을 걸지 않는다. 이 경우 기동 시 + critical 경고가 남는다. + """ + configured = settings.api_key.strip() + if not configured: + return True + if path in public_paths(settings): + return True + protected_docs = {"/docs", "/openapi.json", "/redoc"} + if not path.startswith("/v1") and path not in protected_docs: + return True + # compare_digest 는 비ASCII str 에서 TypeError 를 던진다. 한글/이모지 키를 + # 넣으면 전 요청이 500 이 되므로 반드시 bytes 로 비교한다. + return hmac.compare_digest(configured.encode("utf-8"), (supplied or "").encode("utf-8")) + + @asynccontextmanager async def lifespan(app: FastAPI): settings = get_settings() + validate_auth_settings(settings) if not settings.api_key.strip(): logging.critical( "API_KEY is empty: all /v1 endpoints are unauthenticated. " + "Set REQUIRE_API_KEY=true with a key before exposing this server. " "Do not expose unpublished manuscripts publicly until client key rollout is complete." ) app.state.settings = settings @@ -67,12 +115,9 @@ app.include_router(api_router) @app.middleware("http") async def optional_api_key_auth(request: Request, call_next): """API_KEY가 설정된 운영 환경에서만 API 인증을 강제한다.""" - configured = _settings.api_key.strip() - public_paths = {"/", "/v1/health", "/docs", "/openapi.json", "/redoc"} - if configured and request.url.path not in public_paths and request.url.path.startswith("/v1"): - supplied = request.headers.get("x-api-key", "") - if not hmac.compare_digest(configured, supplied): - return JSONResponse(status_code=401, content={"detail": "Invalid or missing API key"}) + settings = getattr(request.app.state, "settings", None) or _settings + if not is_authorized(settings, request.url.path, request.headers.get("x-api-key", "")): + return JSONResponse(status_code=401, content={"detail": "Invalid or missing API key"}) return await call_next(request) _STATIC_DIR = Path(__file__).resolve().parent / "static" diff --git a/docs/IMPLEMENTATION_RUNBOOK.md b/docs/IMPLEMENTATION_RUNBOOK.md index 96ca605..caec4d1 100644 --- a/docs/IMPLEMENTATION_RUNBOOK.md +++ b/docs/IMPLEMENTATION_RUNBOOK.md @@ -47,6 +47,7 @@ PERSISTENT_INDEX_DIR=/app/data/runtime/index PERSISTENT_SIMILARITY_THRESHOLD=0.65 # 바이칼과 키 전달을 합의한 뒤 활성화 API_KEY= +REQUIRE_API_KEY=true ``` `0.65`는 영속 문자 n-gram 후보 검색의 보수적인 시작값일 뿐 운영 확정값이 아니다. @@ -127,3 +128,74 @@ API 인증·방화벽·키 회전을 별도 운영 작업으로 완료해야 한 포함하지 않는다. 일반 PyPI의 최신 torch가 CUDA 런타임 수 GB를 함께 설치할 수 있기 때문이다. 레거시 KoSimCSE가 반드시 필요한 별도 이미지에서만 해당 Python 버전에 맞는 공식 CPU 전용 torch wheel을 먼저 설치한 뒤 sentence-transformers를 추가한다. + +## 인증 fail-closed (#1) + +| 설정 | 기본 | 의미 | +|---|---|---| +| `API_KEY` | 빈 값 | 비어 있으면 **인증이 걸리지 않는다**. 기동 시 critical 로그가 남는다. | +| `REQUIRE_API_KEY` | `false` | `true` 인데 `API_KEY` 가 비면 **앱이 기동에 실패**한다(`AuthConfigurationError`). | +| `PUBLIC_HEALTH` | `true` | `/v1/health` 를 무인증 공개할지. 모니터링이 키를 못 넣으면 `true` 유지. | +| `PUBLIC_DOCS` | `true` | `/docs`, `/openapi.json`, `/redoc` 공개 여부. | + +- **King 실제 활성화는 바이칼 측 키 전달 후로 보류**한다. 그때까지 `REQUIRE_API_KEY=false` + 로 두되, 서버를 외부에 노출하지 않는다. 키를 받으면 `API_KEY` 설정과 동시에 + `REQUIRE_API_KEY=true` 로 올려 "키를 깜빡한 채 무인증으로 떠 있는" 상태를 원천 차단한다. +- **운영 키는 반드시 ASCII 로 발급**한다. HTTP 헤더는 비ASCII 를 전송할 수 없어 + 한글 키는 인증 자체가 불가능하다(서버는 500 대신 401 을 반환한다). +- `PUBLIC_DOCS=false` 이면 `/docs`, `/openapi.json`, `/redoc`에도 API 키가 + 필요하다. 외부 노출 환경에서는 리버스 프록시 차단도 함께 적용하는 편이 안전하다. + +## coverage 정의와 CPU 상한 (#3/#5) + +두 가지 coverage 를 구분한다. 혼동하면 긴 원고에서 위험도가 항상 낮게 나온다. + +- `matches[].matched_coverage` — **세그먼트 1건**의 일치 구간 / 질의 전체 길이. + 분모가 원고 전체라 30만 자 원고에서는 한 세그먼트가 최대 수천분의 1에 그친다. + 개별 후보의 기여도를 볼 때만 쓴다. +- `score_semantics.union_coverage` — **정밀 비교한 후보 전체**의 일치 구간을 질의 + 좌표에서 **합집합**으로 묶은 비율(중복 구간 1회만 계산). "이 원고의 몇 %가 등록 + 코퍼스와 겹치는가"에 답하는 값이며, `PERSISTENT_MIN_COVERAGE` 게이트와 판례 + 위험도(`legal_risk`)는 **이 값만** 사용한다. + +CPU 상한은 `PERSISTENT_RERANK_TOP_K`(기본 20) 하나로 통제한다. 후보 검색은 전량 +행렬곱으로 하되, `SequenceMatcher` 정밀 비교는 상위 N건에만 돌린다. 나머지 후보는 +`reranked=false` 로 반환되며 `evidence`/`coverage`/`longest_span` 이 0 이고 score 만 +의미가 있다. 잘림이 발생하면 `score_semantics.evidence_truncated=true` 로 노출된다. +이 값을 올리면 요청당 지연이 선형으로 증가한다. + +참조 lemma/요소는 `scripts/build_persistent_index.py` 가 인덱싱 때 DB 에 사전계산해 +둔다(`--skip-precompute` 로 생략 가능). 캐시가 없으면 질의 시 계산 후 자동 백필된다. +500 세그먼트×937자 기준 사전계산 시 요청 지연 0.499s → 0.415s (약 17%). + +## 점수 의미와 잠정 임계값 (#9) + +`confidence` / `matches[].similarity` 는 hashing 어휘 점수와 lemma 겹침을 설정 +가중치로 섞은 **검색 랭킹 점수**이며 침해 확률이 아니다. 응답의 `score_semantics` +가 이를 명시한다. + +- `threshold_source` — `server_default` / `request_override` +- `threshold_calibrated` — `SIMILARITY_THRESHOLD_CALIBRATED` 설정값. 79권 상호비교 + FP 분포를 측정하기 전까지 `false` 로 두고, `provisional=true` 로 노출된다. +- `matches[].match_reasons` — 후보가 채택된 이유. `score_threshold`(결합점수 초과), + `exact_span`(연속 일치 ≥ `PERSISTENT_MIN_EXACT_SPAN`), `coverage`(union coverage + ≥ `PERSISTENT_MIN_COVERAGE`). 이때 후보 채택용 coverage는 해당 출처 문서의 + 세그먼트끼리만 합산한다. 서로 다른 출처의 일치를 합쳐 개별 후보를 통과시키지 + 않는다. 임계값이 아니라 연속 일치 때문에 올라온 후보도 구분할 수 있다. + +임계값 수치는 캘리브레이션 전까지 **임의로 바꾸지 않는다.** 기존 값을 유지하고 +provisional 플래그로만 알린다. + +## 법적 맥락 입력 (#10) + +`POST /v1/plagiarism/detect` 의 `legal_context` 는 선택 필드다. 엔진은 이 사실들을 +**추론하지 않으며**, 미제공 시 `legal_risk.missing_factors` 에 그대로 남는다. + +```json +{"doc_id":"x","text":"...","legal_context":{ + "work_type":"literary","access_evidence":true, + "protected_expression_reviewed":true,"rights_verified":true}} +``` + +판례는 등록된 것만 반환한다(`precedent_ids`). 랭킹은 ① 태그 교집합 수 ② +`work_type` 일치 ③ 사건번호 순이며, 생성형 인용은 어떤 경로로도 발생하지 않는다. diff --git a/scripts/build_persistent_index.py b/scripts/build_persistent_index.py index 4a84306..0d54046 100644 --- a/scripts/build_persistent_index.py +++ b/scripts/build_persistent_index.py @@ -12,6 +12,36 @@ if __package__ in (None, ""): sys.path.insert(0, str(Path(__file__).resolve().parents[1])) from app.engine.persistent_index import PersistentCorpusIndex +from app.engine.provenance import CorpusStore + + +def precompute_features(store: CorpusStore, batch: int = 500) -> int: + """참조 lemma/요소를 미리 계산해 DB 에 저장 (#5). + + 이 작업을 인덱싱 때 1회 해두면, 탐지 요청마다 후보 세그먼트를 형태소 + 분석하던 비용이 사라진다. 이미 채워진 세그먼트는 건너뛴다. + """ + from app.engine.extractor import get_extractor + from app.engine.structural import extract_lemmas + + extractor = get_extractor() + pending: list[tuple[str, list[str], dict]] = [] + updated = 0 + for segment in store.iter_segments(): + if segment.lemmas is not None and segment.elements is not None: + continue + pending.append(( + segment.segment_id, + extract_lemmas(segment.text), + extractor.extract(segment.text).model_dump(), + )) + if len(pending) >= batch: + updated += store.update_segment_features(pending) + print(f" 특징 계산 {updated}건…", flush=True) + pending = [] + if pending: + updated += store.update_segment_features(pending) + return updated def main() -> int: @@ -19,10 +49,22 @@ def main() -> int: p.add_argument("--database", type=Path, required=True) p.add_argument("--index-dir", type=Path, required=True) p.add_argument("--features", type=int, default=2**20) + p.add_argument( + "--skip-precompute", action="store_true", + help="참조 lemma/요소 사전계산을 건너뛴다(질의 시 계산 후 백필됨)", + ) args = p.parse_args() if not args.database.exists(): p.error(f"database does not exist: {args.database}") + + store = CorpusStore(args.database) + precomputed = 0 + if not args.skip_precompute: + precomputed = precompute_features(store) + result = PersistentCorpusIndex(args.database, args.index_dir).sync(args.features) + result["precomputed_features"] = precomputed + result["missing_features"] = store.count_missing_features() print(json.dumps(result, ensure_ascii=False, indent=2)) return 0 diff --git a/tests/test_ai_training_split.py b/tests/test_ai_training_split.py new file mode 100644 index 0000000..4bade16 --- /dev/null +++ b/tests/test_ai_training_split.py @@ -0,0 +1,167 @@ +"""AI 탐지기 학습 데이터 분리 검증 (#F). + +핵심 불변식 두 가지: + 1) fit=train / threshold=val / report=test 로 역할이 섞이지 않는다. + 2) 어떤 source_group 도 두 split 에 동시에 나타나지 않는다(누출 금지). +""" + +from __future__ import annotations + +import json +import subprocess +import sys +from pathlib import Path + +import pytest + +ROOT = Path(__file__).resolve().parents[1] +sys.path.insert(0, str(ROOT)) + +from scripts.build_ai_training_dataset import Record, assign_splits, summarize # noqa: E402 + +sklearn = pytest.importorskip("sklearn", reason="scikit-learn 미설치") +pytest.importorskip("joblib", reason="joblib 미설치") + + +# --------------------------------------------------------------------------- +# 분할 자체의 불변식 (sklearn 불필요 부분) +# --------------------------------------------------------------------------- + +def _records(n_groups_per_label: int = 6, per_group: int = 6) -> list[Record]: + out: list[Record] = [] + for label in (0, 1): + origin = "human" if label == 0 else "ai" + for g in range(n_groups_per_label): + group = f"{origin}:group-{g}" + for i in range(per_group): + out.append(Record( + text=f"{origin} 문단 {g}-{i}", label=label, origin=origin, + source_group=group, book=group, + )) + return out + + +def test_assign_splits_never_shares_a_group(): + records = _records() + assignment = assign_splits(records, (0.7, 0.15, 0.15), seed=7) + for rec in records: + rec.split = assignment[rec.source_group] + + by_split: dict[str, set[str]] = {} + for rec in records: + by_split.setdefault(rec.split, set()).add(rec.source_group) + splits = list(by_split) + for i, a in enumerate(splits): + for b in splits[i + 1:]: + assert not (by_split[a] & by_split[b]), f"{a}/{b} 그룹 중복 = 누출" + + assert summarize(records)["group_overlap_between_splits"] == [] + + +def test_assign_splits_is_deterministic_for_same_seed(): + a = assign_splits(_records(), (0.7, 0.15, 0.15), seed=11) + b = assign_splits(_records(), (0.7, 0.15, 0.15), seed=11) + assert a == b + + +def test_assign_splits_covers_all_three_splits(): + assignment = assign_splits(_records(), (0.7, 0.15, 0.15), seed=3) + assert set(assignment.values()) == {"train", "val", "test"} + + +# --------------------------------------------------------------------------- +# 학습 CLI end-to-end +# --------------------------------------------------------------------------- + +def _write_dataset(path: Path) -> None: + human = "비가 왔다. 나는 그날 학교에 가지 않았고 대신 뒷산에 올라가 온종일 앉아 있었다. 춥지는 않았다. 형이 왔다." + ai = "그날의 기억은 오래도록 남아, 지금까지도 선명하게 떠오르는 장면이 되었다. 아침의 공기는 서늘했고, 발걸음은 조용히 이어졌다." + rows = [] + for label, body, origin in ((0, human, "human"), (1, ai, "ai")): + for g in range(6): + split = "train" if g < 4 else ("val" if g == 4 else "test") + for i in range(8): + rows.append({ + "text": (body + f" 변형 {g}-{i}. ") * 3, + "label": label, "origin": origin, + "source_group": f"{origin}:g{g}", "book": f"{origin}-{g}", + "split": split, + }) + path.write_text( + "\n".join(json.dumps(r, ensure_ascii=False) for r in rows), encoding="utf-8" + ) + + +def _run_trainer(data: Path, out: Path, *extra: str) -> subprocess.CompletedProcess: + return subprocess.run( + [sys.executable, "scripts/train_ai_detector.py", "--data", str(data), + "--out", str(out), *extra], + cwd=ROOT, capture_output=True, text=True, + ) + + +def test_trainer_uses_train_val_test_roles(tmp_path): + data = tmp_path / "ds.jsonl" + out = tmp_path / "model.joblib" + _write_dataset(data) + + proc = _run_trainer(data, out) + assert proc.returncode == 0, proc.stderr[-2000:] + + metrics = json.loads((tmp_path / "model.metrics.json").read_text(encoding="utf-8")) + # 세 역할이 모두 기록되어야 한다 + assert metrics["n_train"] > 0 and metrics["n_val"] > 0 and metrics["n_test"] > 0 + assert {"train", "validation", "test"} <= set(metrics) + # 임계값은 val 에서 뽑혔음이 지표에 남아야 한다 + assert "validation_low_point" in metrics["cuts"] + assert "validation_high_point" in metrics["cuts"] + # 학습에 쓰인 표본 수와 보고 표본 수가 서로 다른 집합이어야 한다 + assert metrics["n_train"] != metrics["n_test"] or metrics["n_val"] != metrics["n_test"] + + +def test_trainer_rejects_group_overlap_between_splits(tmp_path): + data = tmp_path / "leaky.jsonl" + out = tmp_path / "model.joblib" + _write_dataset(data) + + rows = [json.loads(line) for line in data.read_text(encoding="utf-8").splitlines()] + for row in rows: + # train 그룹 하나를 test 에도 등장시켜 누출을 주입 + if row["source_group"] == "human:g0" and row["split"] == "train": + row["split"] = "test" + break + data.write_text( + "\n".join(json.dumps(r, ensure_ascii=False) for r in rows), encoding="utf-8" + ) + + proc = _run_trainer(data, out) + assert proc.returncode == 2, "그룹 누출은 학습을 중단시켜야 한다" + assert "누출" in proc.stderr or "중복" in proc.stderr + + +def test_trainer_fails_on_single_class(tmp_path): + data = tmp_path / "one.jsonl" + _write_dataset(data) + rows = [json.loads(line) for line in data.read_text(encoding="utf-8").splitlines()] + kept = [r for r in rows if r["label"] == 0] + data.write_text( + "\n".join(json.dumps(r, ensure_ascii=False) for r in kept), encoding="utf-8" + ) + proc = _run_trainer(data, tmp_path / "m.joblib") + assert proc.returncode == 2 + assert "단일 클래스" in proc.stderr + + +def test_trainer_fails_when_a_split_is_empty(tmp_path): + data = tmp_path / "noval.jsonl" + _write_dataset(data) + rows = [json.loads(line) for line in data.read_text(encoding="utf-8").splitlines()] + for row in rows: + if row["split"] == "val": + row["split"] = "train" + data.write_text( + "\n".join(json.dumps(r, ensure_ascii=False) for r in rows), encoding="utf-8" + ) + proc = _run_trainer(data, tmp_path / "m.joblib") + assert proc.returncode == 2 + assert "빈 split" in proc.stderr diff --git a/tests/test_api.py b/tests/test_api.py index 0368144..a34d643 100644 --- a/tests/test_api.py +++ b/tests/test_api.py @@ -56,3 +56,69 @@ def test_batch_flow(): job_id = resp.json()["job_id"] status = client.get(f"/v1/plagiarism/batch/{job_id}") assert status.status_code == 200 + + +def test_detect_exposes_score_semantics(): + """#9 — 점수/임계값의 의미가 응답에 명시되어야 한다.""" + with TestClient(app) as client: + resp = client.post( + "/v1/plagiarism/detect", + json={"doc_id": "s-1", "text": "어린왕자는 작은 별에서 온 소년이다."}, + ) + assert resp.status_code == 200 + sem = resp.json()["score_semantics"] + assert sem["threshold_source"] == "server_default" + assert sem["threshold_calibrated"] is False + assert sem["provisional"] is True, "캘리브레이션 전에는 잠정값으로 노출" + assert 0.0 <= sem["union_coverage"] <= 1.0 + assert sem["query_chars"] > 0 + assert "침해 확률이 아닙니다" in sem["note"] + + +def test_detect_reports_request_threshold_override(): + with TestClient(app) as client: + resp = client.post( + "/v1/plagiarism/detect", + json={"doc_id": "s-2", "text": "앤 셜리는 초록 지붕 집에 입양된 소녀다.", + "options": {"threshold": 0.4}}, + ) + sem = resp.json()["score_semantics"] + assert sem["threshold_source"] == "request_override" + assert sem["threshold_used"] == 0.4 + + +def test_detect_accepts_legal_context(): + """#10 — 사람이 확인한 사실을 전달하면 missing_factors 에서 빠진다.""" + with TestClient(app) as client: + base = client.post( + "/v1/plagiarism/detect", + json={"doc_id": "l-1", "text": "홍길동은 활빈당을 만들어 재물을 나누었다."}, + ).json()["legal_risk"] + assert base["access_evidence"] == "not_provided" + assert len(base["missing_factors"]) == 3 + + supplied = client.post( + "/v1/plagiarism/detect", + json={ + "doc_id": "l-2", "text": "홍길동은 활빈당을 만들어 재물을 나누었다.", + "legal_context": { + "work_type": "literary", "access_evidence": True, + "protected_expression_reviewed": True, "rights_verified": True, + }, + }, + ).json()["legal_risk"] + assert supplied["access_evidence"] == "provided" + assert supplied["protected_expression"] == "reviewed" + assert supplied["missing_factors"] == [] + + +def test_matches_carry_match_reasons(): + with TestClient(app) as client: + resp = client.post( + "/v1/plagiarism/detect", + json={"doc_id": "r-1", "text": "어린왕자는 작은 별에서 온 소년이다. 그는 여우를 만난다.", + "options": {"threshold": 0.01}}, + ) + for match in resp.json()["matches"]: + assert match["match_reasons"], "채택 이유가 비어 있으면 안 된다" + assert set(match["match_reasons"]) <= {"score_threshold", "exact_span", "coverage"} diff --git a/tests/test_auth_middleware.py b/tests/test_auth_middleware.py new file mode 100644 index 0000000..49d18f8 --- /dev/null +++ b/tests/test_auth_middleware.py @@ -0,0 +1,144 @@ +"""API 키 인증 미들웨어 (#1). + +앱 전체를 띄우지 않고 순수 함수로 검증한다. 엔진 기동(코퍼스 인덱싱)을 타면 +테스트가 느려지고 인증 로직과 무관한 이유로 깨진다. +""" + +from __future__ import annotations + +import pytest + +from app.core.config import Settings +from app.main import ( + AuthConfigurationError, + is_authorized, + public_paths, + validate_auth_settings, +) + + +def _settings(**overrides) -> Settings: + base = {"api_key": "", "require_api_key": False, + "public_health": True, "public_docs": True} + base.update(overrides) + return Settings(**base) + + +# --------------------------------------------------------------------------- +# fail-closed 기동 검증 +# --------------------------------------------------------------------------- + +def test_require_api_key_without_key_fails_startup(): + with pytest.raises(AuthConfigurationError) as exc: + validate_auth_settings(_settings(require_api_key=True, api_key="")) + assert "REQUIRE_API_KEY" in str(exc.value) + + +def test_require_api_key_with_whitespace_only_key_fails(): + with pytest.raises(AuthConfigurationError): + validate_auth_settings(_settings(require_api_key=True, api_key=" ")) + + +def test_require_api_key_with_key_starts_fine(): + validate_auth_settings(_settings(require_api_key=True, api_key="secret")) + + +def test_default_settings_start_without_key(): + """기본값(개발)에서는 기동을 막지 않는다 — 기존 동작 보존.""" + validate_auth_settings(_settings()) + + +# --------------------------------------------------------------------------- +# 기존 기본 동작 보존 +# --------------------------------------------------------------------------- + +def test_no_key_configured_allows_everything(): + s = _settings(api_key="") + assert is_authorized(s, "/v1/plagiarism/detect", "") is True + assert is_authorized(s, "/v1/corpus", "") is True + + +def test_key_configured_rejects_missing_and_wrong_key(): + s = _settings(api_key="secret") + assert is_authorized(s, "/v1/plagiarism/detect", "") is False + assert is_authorized(s, "/v1/plagiarism/detect", "wrong") is False + assert is_authorized(s, "/v1/plagiarism/detect", "secret") is True + + +def test_corpus_write_paths_are_protected(): + s = _settings(api_key="secret") + for path in ("/v1/corpus", "/v1/corpus/file", "/v1/corpus/doc-1", "/v1/plagiarism/batch"): + assert is_authorized(s, path, "") is False, path + + +# --------------------------------------------------------------------------- +# 공개 경로 설정 +# --------------------------------------------------------------------------- + +def test_health_public_by_default(): + s = _settings(api_key="secret") + assert "/v1/health" in public_paths(s) + assert is_authorized(s, "/v1/health", "") is True + + +def test_health_can_be_protected(): + s = _settings(api_key="secret", public_health=False) + assert "/v1/health" not in public_paths(s) + assert is_authorized(s, "/v1/health", "") is False + assert is_authorized(s, "/v1/health", "secret") is True + + +def test_docs_public_by_default_and_can_be_closed(): + s = _settings(api_key="secret") + assert "/openapi.json" in public_paths(s) + closed = _settings(api_key="secret", public_docs=False) + assert "/openapi.json" not in public_paths(closed) + for path in ("/docs", "/openapi.json", "/redoc"): + assert is_authorized(closed, path, "") is False + assert is_authorized(closed, path, "secret") is True + + +def test_root_console_stays_public(): + s = _settings(api_key="secret") + assert is_authorized(s, "/", "") is True + + +# --------------------------------------------------------------------------- +# 회귀: 비ASCII 키가 TypeError 로 500 을 내지 않아야 한다 +# --------------------------------------------------------------------------- + +def test_non_ascii_key_does_not_raise(): + s = _settings(api_key="비밀키-한글🔑") + assert is_authorized(s, "/v1/plagiarism/detect", "비밀키-한글🔑") is True + assert is_authorized(s, "/v1/plagiarism/detect", "틀린키") is False + assert is_authorized(s, "/v1/plagiarism/detect", "") is False + + +def test_non_ascii_configured_key_returns_401_not_500(): + """서버에 한글 키를 설정해도 500 이 아니라 401 이어야 한다. + + HTTP 헤더는 비ASCII 를 전송할 수 없으므로 이런 키는 사실상 인증 불가지만, + 최소한 서버가 TypeError 로 터지면 안 된다. (운영 키는 ASCII 로 발급할 것) + """ + from fastapi import FastAPI, Request + from fastapi.responses import JSONResponse + from fastapi.testclient import TestClient + + from app.main import is_authorized as guard + + settings = _settings(api_key="한글키") + app = FastAPI() + + @app.middleware("http") + async def auth(request: Request, call_next): + if not guard(settings, request.url.path, request.headers.get("x-api-key", "")): + return JSONResponse(status_code=401, content={"detail": "Invalid or missing API key"}) + return await call_next(request) + + @app.get("/v1/thing") + async def thing(): + return {"ok": True} + + with TestClient(app) as client: + assert client.get("/v1/thing").status_code == 401 + assert client.get("/v1/thing", headers={"x-api-key": "ascii-guess"}).status_code == 401 diff --git a/tests/test_coverage_and_features.py b/tests/test_coverage_and_features.py new file mode 100644 index 0000000..263aa82 --- /dev/null +++ b/tests/test_coverage_and_features.py @@ -0,0 +1,337 @@ +"""union coverage (#3), 참조 특징 캐시 (#5), 점수 의미 (#9), 법적 맥락 (#10). + +scipy/sklearn 이 없으면 인덱스 관련 테스트는 skip 된다. +""" + +from __future__ import annotations + +import pytest + +from app.engine.legal_risk import LegalRiskEngine, Precedent +from app.engine.persistent_index import _covered_length, _merge_intervals +from app.engine.provenance import CorpusStore, DocumentRecord, SegmentRecord + +scipy = pytest.importorskip("scipy", reason="scipy 미설치") + +from app.engine.persistent_index import PersistentCorpusIndex # noqa: E402 + + +def _segment(segment_id: str, text: str, **kw) -> SegmentRecord: + return SegmentRecord( + segment_id=segment_id, document_id=kw.pop("document_id", "doc-1"), + text=text, ordinal=segment_id, char_start=0, char_end=len(text), + source_locator=f"book.json#{segment_id}", **kw, + ) + + +def _index_with(tmp_path, segments: list[SegmentRecord], title="원본"): + db = tmp_path / "corpus.sqlite3" + store = CorpusStore(db) + for doc_id in {s.document_id for s in segments}: + store.upsert_document(DocumentRecord(document_id=doc_id, title=title)) + store.add_segments(segments) + index = PersistentCorpusIndex(db, tmp_path / "index") + index.sync() + return store, index + + +# --------------------------------------------------------------------------- +# 구간 병합 (순수 함수) +# --------------------------------------------------------------------------- + +def test_merge_intervals_deduplicates_overlap(): + assert _merge_intervals([(0, 10), (5, 20), (30, 40)]) == [(0, 20), (30, 40)] + assert _covered_length([(0, 10), (5, 20)]) == 20 + assert _covered_length([(0, 10), (0, 10)]) == 10, "중복 구간을 두 번 세면 안 된다" + assert _covered_length([]) == 0 + + +def test_merge_intervals_handles_adjacent_and_nested(): + assert _merge_intervals([(0, 10), (10, 20)]) == [(0, 20)] + assert _merge_intervals([(0, 100), (10, 20)]) == [(0, 100)] + + +# --------------------------------------------------------------------------- +# #3 union coverage +# --------------------------------------------------------------------------- + +def test_union_coverage_sums_multiple_segments(tmp_path): + """서로 다른 세그먼트가 질의의 다른 부분과 일치하면 coverage 가 합산된다.""" + part_a = "바닷가 마을에서 파도 소리를 들으며 자란 기억이 아직도 선명하게 남아 있다. " * 3 + part_b = "군에 입대하던 날 아버지는 아무 말 없이 내 어깨를 두드려 주셨던 기억이 난다. " * 3 + _, index = _index_with(tmp_path, [ + _segment("seg-a", part_a), _segment("seg-b", part_b), + ]) + query = part_a + "완전히 무관한 중간 문단입니다. " * 20 + part_b + result = index.search(query, top_k=10) + + assert result.union_coverage > 0.4, "두 구간이 모두 반영되어야 한다" + assert result.covered_chars >= len(part_a) + assert result.query_chars == len(query) + # 개별 세그먼트 coverage 는 각자 union 보다 작다 + per_hit = [h.coverage for h in result.hits if h.reranked] + assert max(per_hit) < result.union_coverage + + +def test_document_coverage_does_not_leak_between_sources(tmp_path): + part_a = "바닷가 마을에서 파도 소리를 들으며 자랐다. " * 3 + part_b = "군에 입대하던 날 아버지가 내 어깨를 두드렸다. " * 3 + _, index = _index_with(tmp_path, [ + _segment("seg-a", part_a, document_id="doc-a"), + _segment("seg-b", part_b, document_id="doc-b"), + ]) + query = part_a + ("서로 무관한 중간 문장입니다. " * 20) + part_b + result = index.search(query, top_k=10) + + assert result.union_coverage > result.document_coverage["doc-a"] + assert result.union_coverage > result.document_coverage["doc-b"] + assert result.document_coverage["doc-a"] < 0.30 + assert result.document_coverage["doc-b"] < 0.30 + + +def test_union_coverage_low_for_long_unrelated_document(tmp_path): + _, index = _index_with(tmp_path, [_segment("seg-1", "바닷가 마을의 파도 소리를 기억한다.")]) + query = "전혀 다른 주제의 글입니다. 오늘 회의에서 분기 실적을 논의했습니다. " * 100 + result = index.search(query, top_k=5) + assert result.union_coverage < 0.1 + + +def test_short_full_copy_reaches_high_coverage(tmp_path): + text = "나는 어린 시절 바닷가 마을에서 살았고 매일 파도 소리를 들으며 잠들었다." + _, index = _index_with(tmp_path, [_segment("seg-1", text)]) + result = index.search(text, top_k=5) + assert result.union_coverage > 0.8 + + +def test_long_document_partial_copy_is_not_structurally_zero(tmp_path): + """긴 원고 안의 부분 복사가 coverage 에 실제로 잡히는지 (#3 회귀).""" + copied = "바닷가 마을에서 파도 소리를 들으며 자란 기억이 선명하다. " * 10 + _, index = _index_with(tmp_path, [_segment("seg-1", copied)]) + filler = "무관한 문장입니다. " * 300 + query = filler + copied + filler + result = index.search(query, top_k=5) + assert result.covered_chars >= len(copied) * 0.5 + assert result.union_coverage > 0.0 + + +# --------------------------------------------------------------------------- +# CPU 상한 +# --------------------------------------------------------------------------- + +def test_evidence_limit_caps_precise_comparison(tmp_path): + segments = [ + _segment(f"seg-{i}", f"바닷가 마을 이야기 {i}번 문단입니다. 파도 소리를 들었다. " * 3) + for i in range(8) + ] + _, index = _index_with(tmp_path, segments) + result = index.search("바닷가 마을 이야기 파도 소리를 들었다.", top_k=8, evidence_limit=3) + + reranked = [h for h in result.hits if h.reranked] + assert len(reranked) == 3 + assert result.evidence_truncated is True + for hit in result.hits: + if not hit.reranked: + assert hit.evidence == [] and hit.coverage == 0.0 and hit.longest_span == 0 + + +def test_query_wrapper_still_returns_hits(tmp_path): + """기존 호출부 호환 — query() 는 여전히 list[PersistentHit].""" + _, index = _index_with(tmp_path, [_segment("seg-1", "바닷가 마을의 파도 소리.")]) + hits = index.query("바닷가 마을의 파도 소리.", top_k=1) + assert isinstance(hits, list) and hits[0].segment_id == "seg-1" + + +# --------------------------------------------------------------------------- +# #5 참조 특징 캐시 + 마이그레이션 +# --------------------------------------------------------------------------- + +def test_v1_database_migrates_without_data_loss(tmp_path): + """lemmas_json/elements_json 없는 기존 DB 도 그대로 열려야 한다.""" + import sqlite3 + + db = tmp_path / "old.sqlite3" + con = sqlite3.connect(db) + con.executescript( + """ + CREATE TABLE corpus_meta (key TEXT PRIMARY KEY, value TEXT NOT NULL); + CREATE TABLE documents ( + document_id TEXT PRIMARY KEY, title TEXT NOT NULL, source_path TEXT, + source_sha256 TEXT, metadata_json TEXT NOT NULL DEFAULT '{}', + created_at TEXT DEFAULT CURRENT_TIMESTAMP, + updated_at TEXT DEFAULT CURRENT_TIMESTAMP); + CREATE TABLE segments ( + segment_id TEXT PRIMARY KEY, document_id TEXT NOT NULL, + ordinal TEXT NOT NULL, text TEXT NOT NULL, text_sha256 TEXT NOT NULL, + coordinate_scope TEXT NOT NULL, page_number INTEGER, + paragraph_number INTEGER, char_start INTEGER, char_end INTEGER, + source_locator TEXT, metadata_json TEXT NOT NULL DEFAULT '{}', + created_at TEXT DEFAULT CURRENT_TIMESTAMP, + UNIQUE(document_id, text_sha256)); + INSERT INTO documents(document_id,title) VALUES('doc-1','옛 책'); + INSERT INTO segments VALUES('seg-1','doc-1','1','옛 본문입니다.','h','episode', + NULL,NULL,0,7,NULL,'{}',CURRENT_TIMESTAMP); + """ + ) + con.commit() + con.close() + + store = CorpusStore(db) + store.initialize() # 마이그레이션 + segments = list(store.iter_segments()) + assert len(segments) == 1 + assert segments[0].text == "옛 본문입니다." + assert segments[0].lemmas is None + assert store.count_missing_features() == 1 + + +def test_feature_roundtrip_and_backfill(tmp_path): + store = CorpusStore(tmp_path / "corpus.sqlite3") + store.upsert_document(DocumentRecord(document_id="doc-1", title="책")) + store.add_segments([_segment("seg-1", "홍길동은 활빈당을 만들었다.")]) + assert store.count_missing_features() == 1 + + store.update_segment_features([("seg-1", ["홍길동", "활빈당"], {"characters": ["홍길동"]})]) + assert store.count_missing_features() == 0 + loaded = store.get_segments(["seg-1"])["seg-1"] + assert loaded.lemmas == ["홍길동", "활빈당"] + assert loaded.elements == {"characters": ["홍길동"]} + + +def test_precomputed_features_reach_the_hit(tmp_path): + store = CorpusStore(tmp_path / "corpus.sqlite3") + store.upsert_document(DocumentRecord(document_id="doc-1", title="책")) + store.add_segments([_segment( + "seg-1", "나는 어린 시절 바닷가 마을에서 살았다.", + lemmas=["바닷가", "마을", "살다"], + elements={"characters": [], "motifs": [], "genre": None, "keywords": ["바닷가"]}, + )]) + index = PersistentCorpusIndex(tmp_path / "corpus.sqlite3", tmp_path / "index") + index.sync() + hit = index.query("나는 어린 시절 바닷가 마을에서 살았다.", top_k=1)[0] + assert hit.reference_lemmas == ["바닷가", "마을", "살다"] + assert hit.reference_elements["keywords"] == ["바닷가"] + + +def test_detector_uses_cache_instead_of_recomputing(monkeypatch, tmp_path): + """캐시가 있으면 형태소 분석을 호출하지 않아야 한다 (#5 호출횟수 테스트).""" + from app.api.schemas import ExtractedElements + from app.engine import detector as det_module + from app.engine.persistent_index import PersistentHit + + calls = {"lemmas": 0, "extract": 0} + + def counting_lemmas(text, *a, **kw): + calls["lemmas"] += 1 + return ["x"] + + monkeypatch.setattr(det_module, "extract_lemmas", counting_lemmas) + + detector = det_module.PlagiarismDetector.__new__(det_module.PlagiarismDetector) + detector._feature_cache = {} + detector._persistent = None + + class _Extractor: + def extract(self, text): + calls["extract"] += 1 + return ExtractedElements() + + detector._extractor = _Extractor() + + cached_hit = PersistentHit( + segment_id="seg-1", document_id="doc-1", title="책", score=0.9, + evidence=[], coverage=0.0, longest_span=0, source_locator=None, + coordinate_scope="episode", page_number=None, paragraph_number=None, + source_char_start=None, source_char_end=None, reference_text="본문", + reference_lemmas=["미리", "계산"], + reference_elements={"characters": [], "motifs": [], "genre": None, "keywords": []}, + ) + lemmas, _ = detector.reference_features(cached_hit) + assert lemmas == ["미리", "계산"] + assert calls == {"lemmas": 0, "extract": 0}, "캐시가 있는데 재계산했다" + + uncached = PersistentHit( + segment_id="seg-2", document_id="doc-1", title="책", score=0.9, + evidence=[], coverage=0.0, longest_span=0, source_locator=None, + coordinate_scope="episode", page_number=None, paragraph_number=None, + source_char_start=None, source_char_end=None, reference_text="본문", + ) + detector.reference_features(uncached) + assert calls == {"lemmas": 1, "extract": 1} + # 두 번째 호출은 프로세스 캐시로 처리 + detector.reference_features(uncached) + assert calls == {"lemmas": 1, "extract": 1} + + +# --------------------------------------------------------------------------- +# #10 법적 맥락 + 판례 랭킹 +# --------------------------------------------------------------------------- + +def _precedents() -> list[Precedent]: + return [ + Precedent("2020다1", "가", "https://x/1", ("literary",), + ("reproduction",), (), "요지1"), + Precedent("2019다2", "나", "https://x/2", ("literary",), + ("reproduction", "derivative_work"), (), "요지2"), + Precedent("2018다3", "다", "https://x/3", ("musical",), + ("reproduction",), (), "요지3"), + ] + + +def test_precedent_ranking_prefers_more_tag_overlap(): + engine = LegalRiskEngine(_precedents()) + out = engine.assess( + max_similarity=0.9, coverage=0.5, longest_span=200, + legal_tags=["reproduction", "derivative_work"], work_type="literary", + ) + assert out.precedent_ids[0] == "2019다2", "태그 교집합이 큰 판례가 먼저" + assert "2018다3" not in out.precedent_ids, "work_type 이 다른 판례는 제외" + + +def test_only_registered_precedents_are_returned(): + engine = LegalRiskEngine(_precedents()) + out = engine.assess( + max_similarity=0.9, coverage=0.5, longest_span=200, + legal_tags=["reproduction"], work_type="literary", + ) + registered = {p.case_id for p in _precedents()} + assert set(out.precedent_ids) <= registered + + +def test_empty_precedent_db_reports_insufficient(): + out = LegalRiskEngine([]).assess( + max_similarity=0.9, coverage=0.9, longest_span=500, legal_tags=["reproduction"], + ) + assert out.status == "insufficient_precedent_data" + assert out.risk_level is None + assert out.precedent_ids == () + + +def test_legal_context_clears_missing_factors(): + engine = LegalRiskEngine(_precedents()) + default = engine.assess(max_similarity=0.5, coverage=0.1, longest_span=20, + legal_tags=["reproduction"]) + assert len(default.missing_factors) == 3 + assert default.access_evidence == "not_provided" + + supplied = engine.assess( + max_similarity=0.5, coverage=0.1, longest_span=20, legal_tags=["reproduction"], + access_evidence=True, protected_expression_reviewed=True, rights_verified=True, + ) + assert supplied.missing_factors == () + assert supplied.access_evidence == "provided" + assert supplied.protected_expression == "reviewed" + + +def test_legal_context_flows_through_detect_request(): + from app.api.schemas import DetectRequest, LegalContext + + req = DetectRequest( + doc_id="d", text="본문", + legal_context=LegalContext(work_type="musical", access_evidence=False, + protected_expression_reviewed=True), + ) + assert req.legal_context.work_type == "musical" + assert req.legal_context.access_evidence is False + assert req.legal_context.rights_verified is False + # 미제공이 기본 + assert DetectRequest(doc_id="d", text="본문").legal_context is None diff --git a/tests/test_review_regressions.py b/tests/test_review_regressions.py new file mode 100644 index 0000000..b7ac068 --- /dev/null +++ b/tests/test_review_regressions.py @@ -0,0 +1,144 @@ +"""이미 반영된 리뷰 항목의 회귀 방지 (#2 #4 #6 #7 #8).""" + +from __future__ import annotations + +import json + +import pytest + +from app.engine.provenance import CorpusStore, DocumentRecord, SegmentRecord + +pytest.importorskip("scipy", reason="scipy 미설치") + +from app.engine.persistent_index import ( # noqa: E402 + VECTORIZER_CONFIG, + PersistentCorpusIndex, +) + + +def _store_with(tmp_path, texts: list[str]) -> CorpusStore: + store = CorpusStore(tmp_path / "corpus.sqlite3") + store.upsert_document(DocumentRecord(document_id="doc-1", title="책")) + store.add_segments([ + SegmentRecord(segment_id=f"seg-{i}", document_id="doc-1", text=t, + ordinal=str(i), char_start=0, char_end=len(t)) + for i, t in enumerate(texts) + ]) + return store + + +# --- #2 원자적 교체 ------------------------------------------------------- + +def test_matrix_file_is_generation_scoped(tmp_path): + """세대별 파일명이라 교체 중 이전 인덱스가 덮어써지지 않는다.""" + store = _store_with(tmp_path, ["첫 번째 세그먼트 본문입니다."]) + index = PersistentCorpusIndex(tmp_path / "corpus.sqlite3", tmp_path / "index") + index.sync() + first = json.loads((tmp_path / "index" / "index.json").read_text())["matrix_file"] + + store.add_segments([SegmentRecord( + segment_id="seg-9", document_id="doc-1", text="두 번째 세그먼트 본문입니다.", + ordinal="9", char_start=0, char_end=10, + )]) + index.sync() + second = json.loads((tmp_path / "index" / "index.json").read_text())["matrix_file"] + + assert first != second, "세대가 바뀌면 파일명도 바뀌어야 한다" + assert (tmp_path / "index" / first).exists(), "이전 세대 파일이 살아 있어야 한다" + + +def test_replaced_index_object_is_self_consistent(tmp_path): + """load() 한 객체는 matrix 행수와 segment_ids 길이가 항상 일치한다.""" + _store_with(tmp_path, [f"세그먼트 {i} 본문입니다. 파도 소리." for i in range(5)]) + index = PersistentCorpusIndex(tmp_path / "corpus.sqlite3", tmp_path / "index") + index.sync() + fresh = PersistentCorpusIndex(tmp_path / "corpus.sqlite3", tmp_path / "index").load() + assert fresh._matrix.shape[0] == len(fresh._meta["segment_ids"]) == 5 + + +# --- #8 vectorizer 설정 고정 --------------------------------------------- + +def test_index_with_different_vectorizer_config_is_rejected(tmp_path): + _store_with(tmp_path, ["본문입니다. 파도 소리를 들었다."]) + index = PersistentCorpusIndex(tmp_path / "corpus.sqlite3", tmp_path / "index") + index.sync() + + meta_path = tmp_path / "index" / "index.json" + meta = json.loads(meta_path.read_text()) + meta["vectorizer_config"] = {**VECTORIZER_CONFIG, "ngram_range": [2, 6]} + meta_path.write_text(json.dumps(meta, ensure_ascii=False), encoding="utf-8") + + with pytest.raises(ValueError, match="재빌드"): + PersistentCorpusIndex(tmp_path / "corpus.sqlite3", tmp_path / "index").load() + + +def test_config_change_forces_rebuild_not_append(tmp_path): + _store_with(tmp_path, ["본문입니다. 파도 소리를 들었다."]) + index_dir = tmp_path / "index" + index = PersistentCorpusIndex(tmp_path / "corpus.sqlite3", index_dir) + index.sync() + # n_features 가 달라지면 append 가 성립하지 않아야 한다 + result = PersistentCorpusIndex(tmp_path / "corpus.sqlite3", index_dir).sync(n_features=2**16) + assert result["mode"] == "rebuild" + + +# --- #6 document_count 캐시 ---------------------------------------------- + +def test_document_count_served_from_meta_without_table_scan(tmp_path, monkeypatch): + _store_with(tmp_path, ["본문입니다."]) + index = PersistentCorpusIndex(tmp_path / "corpus.sqlite3", tmp_path / "index") + index.sync() + loaded = PersistentCorpusIndex(tmp_path / "corpus.sqlite3", tmp_path / "index").load() + + def explode(): + raise AssertionError("document_count 가 매 요청 DB 를 스캔하고 있다") + + monkeypatch.setattr(loaded.store, "document_count", explode) + assert loaded.document_count == 1 + + +# --- #7 API 업로드 청킹 --------------------------------------------------- + +def test_api_upload_is_chunked_not_single_segment(tmp_path, monkeypatch): + from app.core.config import Settings + from app.engine.detector import PlagiarismDetector + + settings = Settings( + use_persistent_index=True, + corpus_db_path=str(tmp_path / "corpus.sqlite3"), + persistent_index_dir=str(tmp_path / "index"), + precedents_path=str(tmp_path / "none.jsonl"), + ai_detector_model_path=str(tmp_path / "none.joblib"), + use_clustering=False, + use_lsh_filter=False, + ) + _store_with(tmp_path, ["씨앗 세그먼트입니다."]) + PersistentCorpusIndex(tmp_path / "corpus.sqlite3", tmp_path / "index").sync() + + detector = PlagiarismDetector(settings=settings) + assert detector.uses_persistent_index + + long_text = "긴 원고 문장입니다. 계속 이어집니다. " * 200 # 약 4,000자 + doc_id = detector.add_persistent_document(None, "업로드 원고", long_text) + + segments = [s for s in detector._persistent.store.iter_segments() + if s.document_id == doc_id] + assert len(segments) > 1, "문서 전체가 세그먼트 1개로 저장되면 부분 표절을 못 잡는다" + assert all(len(s.text) <= 1000 for s in segments) + # 오프셋이 원문을 정확히 복원해야 한다 + for seg in segments: + assert long_text.strip()[seg.char_start:seg.char_end] == seg.text + + +# --- #4 detect 가 이벤트 루프를 막지 않는지 ------------------------------- + +def test_detect_route_runs_in_threadpool(): + """라우트가 동기 detect 를 threadpool 로 넘기는지 (소스 계약 확인).""" + import inspect + + from app.api import routes + + source = inspect.getsource(routes.detect) + assert "run_in_threadpool" in source + for name in ("corpus_upload_json", "corpus_upload_file", "corpus_delete"): + assert "run_in_threadpool" in inspect.getsource(getattr(routes, name)), name