hbyang
e89cd23ed6
검토자가 침해 여부를 눈으로 판별하려면 원문이 있어야 하고, 어떤 조건으로 걸렸는지 알아야 한다. 두 가지를 워크북에 싣는다. - extract_suspected_excerpts.py: 코퍼스 DB(읽기 전용)에서 질의·상대 원문과 일치 구간을 뽑는다. 킹서버 기본 python 이 3.6 이라 이 파일만 구문을 맞췄다. - '판정 기준' 시트: 3개 OR 조건과 결합유사도 가중치, 조건별 충족 현황 - '원문 대조' 시트: 판정 사유 컬럼으로 각 건이 걸린 조건을 표시 판례 표기도 바로잡는다. USE_LLM_LEGAL_JUDGE=false 는 LLM 법적 판단만 끈 것이고 판례 검색은 규칙 기반으로 늘 동작한다. legal_risk.assess() 가 점수 하한 없이 상위 5건을 붙이므로 매칭 미탐지 건에도 판례가 채워진다. 한 줄에 뭉쳐 두면 모순으로 읽혀 판례 검색과 법적 판단을 분리해 적었다. 원문이 담긴 산출물은 gitignore 로 제외한다. 필요하면 위 스크립트로 코퍼스에서 다시 뽑을 수 있어 저장소에 영구 보존할 이유가 없다. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
139 lines
5.4 KiB
Python
139 lines
5.4 KiB
Python
"""판별 대상의 원문과 일치 구간을 뽑아 JSONL로 쓴다.
|
|
|
|
배치 결과 파일에는 원문이 없다. 검토자가 실제 침해인지 눈으로 판별하려면
|
|
질의 원문과 매칭 상대 원문, 그리고 둘 사이의 일치 구간이 필요하다.
|
|
코퍼스 DB는 읽기 전용으로만 연다.
|
|
|
|
킹서버의 기본 python3 가 구버전이라 이 파일만은 표준 라이브러리와 구문
|
|
호환 범위 안에서 쓴다 (타입 표기·f-string 미사용, __future__ import 없음).
|
|
저장소의 다른 모듈과 스타일이 다른 이유가 이것이다.
|
|
|
|
원문이 포함된 산출물이므로 취급에 주의한다.
|
|
"""
|
|
|
|
import argparse
|
|
import json
|
|
import sqlite3
|
|
from difflib import SequenceMatcher
|
|
|
|
# app.engine.persistent_index._evidence_spans 와 같은 기준을 쓴다.
|
|
MIN_MATCH = 12
|
|
SPAN_LIMIT = 10
|
|
|
|
|
|
def open_readonly(path):
|
|
con = sqlite3.connect("file:%s?mode=ro" % path, uri=True)
|
|
con.row_factory = sqlite3.Row
|
|
return con
|
|
|
|
|
|
def fetch_segments(con, segment_ids):
|
|
texts = {}
|
|
ids = [sid for sid in segment_ids if sid]
|
|
for start in range(0, len(ids), 500):
|
|
chunk = ids[start:start + 500]
|
|
placeholders = ",".join("?" * len(chunk))
|
|
rows = con.execute(
|
|
"SELECT segment_id, text FROM segments WHERE segment_id IN (%s)" % placeholders,
|
|
chunk,
|
|
)
|
|
for row in rows:
|
|
texts[row["segment_id"]] = row["text"]
|
|
return texts
|
|
|
|
|
|
def fetch_document_texts(con, document_ids):
|
|
"""문서 단위 질의(생활수기)용. 세그먼트를 순서대로 이어 붙인다."""
|
|
texts = {}
|
|
for document_id in document_ids:
|
|
rows = con.execute(
|
|
"SELECT text FROM segments WHERE document_id = ? ORDER BY ordinal, segment_id",
|
|
(document_id,),
|
|
).fetchall()
|
|
if rows:
|
|
texts[document_id] = "\n".join(row["text"] for row in rows)
|
|
return texts
|
|
|
|
|
|
def matching_spans(query, reference):
|
|
if not query or not reference:
|
|
return []
|
|
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))
|
|
selected = sorted(useful[:SPAN_LIMIT], key=lambda b: b.a)
|
|
return [query[b.a:b.a + b.size] for b in selected]
|
|
|
|
|
|
def main():
|
|
parser = argparse.ArgumentParser(description=__doc__)
|
|
parser.add_argument("--database", required=True)
|
|
parser.add_argument("--jsonl", required=True, help="배치 결과 JSONL")
|
|
parser.add_argument("--out-jsonl", required=True)
|
|
parser.add_argument("--max-chars", type=int, default=6000,
|
|
help="원문 컬럼 길이 상한. 0 이면 제한 없음")
|
|
parser.add_argument("--scope", choices=("all", "suspected"), default="all",
|
|
help="all=전 건(미탐지 포함), suspected=침해 의심 건만")
|
|
args = parser.parse_args()
|
|
|
|
with open(args.jsonl, encoding="utf-8") as handle:
|
|
all_rows = [json.loads(line) for line in handle if line.strip()]
|
|
if args.scope == "all":
|
|
targets = all_rows
|
|
else:
|
|
targets = [row for row in all_rows if row.get("침해 의심")]
|
|
flagged = sum(1 for row in targets if row.get("침해 의심"))
|
|
print("대상 %d건 (침해 의심 %d / 매칭 미탐지 %d)"
|
|
% (len(targets), flagged, len(targets) - flagged))
|
|
|
|
con = open_readonly(args.database)
|
|
try:
|
|
segment_ids = set()
|
|
document_ids = set()
|
|
for row in targets:
|
|
for key in ("query_segment_id", "매칭 상대 segment"):
|
|
if row.get(key):
|
|
segment_ids.add(row[key])
|
|
if not row.get("query_segment_id"):
|
|
document_ids.add(row["doc_id"])
|
|
segment_texts = fetch_segments(con, segment_ids)
|
|
document_texts = fetch_document_texts(con, document_ids)
|
|
finally:
|
|
con.close()
|
|
|
|
def clip(text):
|
|
if args.max_chars and len(text) > args.max_chars:
|
|
return text[:args.max_chars] + "\n…(이하 %d자 생략)" % (len(text) - args.max_chars)
|
|
return text
|
|
|
|
written = 0
|
|
missing = 0
|
|
with open(args.out_jsonl, "w", encoding="utf-8") as handle:
|
|
for row in targets:
|
|
query_key = row["query_segment_id"] or row["doc_id"]
|
|
query_text = segment_texts.get(query_key) or document_texts.get(query_key, "")
|
|
source_text = segment_texts.get(row.get("매칭 상대 segment") or "", "")
|
|
# 미탐지 건은 매칭 상대가 없는 것이 정상이므로 질의 원문만 확인한다.
|
|
if not query_text:
|
|
missing += 1
|
|
record = {
|
|
"query_key": row["query_key"],
|
|
"가명 제목": row["가명 제목"],
|
|
"침해 여부": row["침해 여부"],
|
|
"매칭 상대 제목": row.get("매칭 상대 제목"),
|
|
"원천 구분": row["원천 구분"],
|
|
"결합유사도": row["결합유사도"],
|
|
"근거 스팬 수": row.get("evidence 스팬 수"),
|
|
"일치 구간": matching_spans(query_text, source_text),
|
|
"검사 대상 원문": clip(query_text),
|
|
"매칭 상대 원문": clip(source_text),
|
|
}
|
|
handle.write(json.dumps(record, ensure_ascii=False) + "\n")
|
|
written += 1
|
|
print("wrote %s rows=%d 원문누락=%d" % (args.out_jsonl, written, missing))
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|