From 480f00702f0148d329fef44babc13ce0b4abba6a Mon Sep 17 00:00:00 2001 From: hbyang Date: Fri, 21 Aug 2026 08:44:22 +0900 Subject: [PATCH] =?UTF-8?q?feat:=20=EC=A0=84=EB=8B=AC=20=ED=8C=8C=EC=9D=BC?= =?UTF-8?q?=20=EC=A0=84=EC=9A=A9=20=EC=A0=95=EC=B1=85=EA=B3=BC=20AI=20?= =?UTF-8?q?=ED=91=9C=EB=B3=B8=20=ED=92=88=EC=A7=88=20=EA=B0=90=EC=82=AC=20?= =?UTF-8?q?=EC=B6=94=EA=B0=80?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- docs/RECEIVED_DATA_PIPELINE.md | 3 + scripts/clean_ai_samples.py | 99 +++++++++++++++++++++++ scripts/download_life_writing_drive.py | 92 --------------------- tests/test_clean_ai_samples.py | 12 +++ tests/test_download_life_writing_drive.py | 9 --- 5 files changed, 114 insertions(+), 101 deletions(-) create mode 100644 scripts/clean_ai_samples.py delete mode 100644 scripts/download_life_writing_drive.py create mode 100644 tests/test_clean_ai_samples.py delete mode 100644 tests/test_download_life_writing_drive.py diff --git a/docs/RECEIVED_DATA_PIPELINE.md b/docs/RECEIVED_DATA_PIPELINE.md index 622ac7d..f3568a4 100644 --- a/docs/RECEIVED_DATA_PIPELINE.md +++ b/docs/RECEIVED_DATA_PIPELINE.md @@ -11,6 +11,9 @@ `ai_assistance=false`로 기록한다. 두 번째 파일은 본문이 아니므로 목록만으로 학습·탐지에 쓰지 않는다. +> 운영 정책: 목록에 있는 Google Drive·공개 웹 링크에 자동 접속하지 +> 않는다. 컴북스가 실제로 다운로드해 전달한 로컬 원본만 OCR·적재한다. + ## 2. 익명화와 누출 방지 - `id`(이메일)는 운영 비밀 salt로 HMAC 가명화한다. 원본 ID는 DB와 학습셋에 저장하지 않는다. diff --git a/scripts/clean_ai_samples.py b/scripts/clean_ai_samples.py new file mode 100644 index 0000000..e2c534a --- /dev/null +++ b/scripts/clean_ai_samples.py @@ -0,0 +1,99 @@ +#!/usr/bin/env python3 +"""AI 대조문 JSONL의 지시문 누출·인사말·중복을 제거하고 품질 보고서를 남긴다.""" + +from __future__ import annotations + +import argparse +import hashlib +import json +import re +from collections import Counter +from pathlib import Path + +PROMPT_LEAK_MARKERS = ("아래 소재", "본문만 출력", "문체 지시", "분량:") +ASSISTANT_PREFIXES = ("안녕하세요", "물론입니다", "네, ") + + +def hangul_ratio(text: str) -> float: + letters = [ch for ch in text if ch.isalpha()] + return sum("가" <= ch <= "힣" for ch in letters) / len(letters) if letters else 0.0 + + +def rejection_reason(text: str, min_chars: int, max_chars: int, min_hangul: float) -> str | None: + stripped = text.strip() + if len(stripped) < min_chars or len(stripped) > max_chars: + return "length" + if stripped.startswith(ASSISTANT_PREFIXES): + return "assistant_greeting" + if any(marker in stripped for marker in PROMPT_LEAK_MARKERS): + return "prompt_leak" + if hangul_ratio(stripped) < min_hangul: + return "low_hangul_ratio" + return None + + +def main() -> int: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--input", type=Path, action="append", required=True) + parser.add_argument("--out", type=Path, required=True) + parser.add_argument("--report", type=Path, required=True) + parser.add_argument("--min-chars", type=int, default=199) + parser.add_argument("--max-chars", type=int, default=1803) + parser.add_argument("--min-hangul-ratio", type=float, default=0.55) + args = parser.parse_args() + + counts: Counter[str] = Counter() + seen: set[str] = set() + accepted = [] + for path in args.input: + with path.open(encoding="utf-8") as handle: + for line_number, line in enumerate(handle, start=1): + if not line.strip(): + continue + counts["input"] += 1 + try: + row = json.loads(line) + except json.JSONDecodeError: + counts["invalid_json"] += 1 + continue + text = str(row.get("text") or "").strip() + reason = rejection_reason( + text, args.min_chars, args.max_chars, args.min_hangul_ratio + ) + if reason: + counts[reason] += 1 + continue + key = hashlib.sha1(re.sub(r"\s+", "", text).encode()).hexdigest() + if key in seen: + counts["duplicate"] += 1 + continue + seen.add(key) + row["quality_audit"] = "passed-v1" + row["source_file"] = path.name + row["source_line"] = line_number + accepted.append(row) + + args.out.parent.mkdir(parents=True, exist_ok=True) + with args.out.open("w", encoding="utf-8") as handle: + for row in accepted: + handle.write(json.dumps(row, ensure_ascii=False) + "\n") + counts["accepted"] = len(accepted) + report = { + "inputs": [str(path) for path in args.input], + "output": str(args.out), + "counts": dict(counts), + "quality_policy": { + "min_chars": args.min_chars, + "max_chars": args.max_chars, + "min_hangul_ratio": args.min_hangul_ratio, + "assistant_prefixes": list(ASSISTANT_PREFIXES), + "prompt_leak_markers": list(PROMPT_LEAK_MARKERS), + }, + } + args.report.write_text(json.dumps(report, ensure_ascii=False, indent=2), encoding="utf-8") + print(json.dumps(report["counts"], ensure_ascii=False)) + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/scripts/download_life_writing_drive.py b/scripts/download_life_writing_drive.py deleted file mode 100644 index 10cb61f..0000000 --- a/scripts/download_life_writing_drive.py +++ /dev/null @@ -1,92 +0,0 @@ -#!/usr/bin/env python3 -"""수령한 생활 수기집 목록의 Google Drive 공유 원본을 다운로드한다.""" - -from __future__ import annotations - -import argparse -import json -import re -import sys -from pathlib import Path - -FILE_ID = re.compile(r"drive\.google\.com/file/d/([^/]+)") -FOLDER_ID = re.compile(r"drive\.google\.com/drive/folders/([^/?]+)") - - -def parse_args() -> argparse.Namespace: - parser = argparse.ArgumentParser(description=__doc__) - parser.add_argument("manifest", type=Path) - parser.add_argument("--out-dir", type=Path, required=True) - parser.add_argument("--report", type=Path, required=True) - return parser.parse_args() - - -def drive_kind(url: str) -> tuple[str, str] | None: - if match := FILE_ID.search(url): - return "file", match.group(1) - if match := FOLDER_ID.search(url): - return "folder", match.group(1) - return None - - -def main() -> int: - args = parse_args() - try: - import gdown - from openpyxl import load_workbook - except ImportError: - print("gdown과 openpyxl이 필요합니다.", file=sys.stderr) - return 2 - - workbook = load_workbook(args.manifest, read_only=False, data_only=False) - sheet = workbook.worksheets[0] - results = [] - args.out_dir.mkdir(parents=True, exist_ok=True) - - for rownum in range(3, sheet.max_row + 1): - cell = sheet.cell(row=rownum, column=10) - url = cell.hyperlink.target if cell.hyperlink else "" - parsed = drive_kind(url or "") - if not parsed: - continue - kind, drive_id = parsed - target = args.out_dir / f"source_{rownum:03d}" - target.mkdir(parents=True, exist_ok=True) - result = { - "row": rownum, - "kind": kind, - "drive_id": drive_id, - "url": url, - "label": str(cell.value or ""), - "status": "failed", - "files": [], - } - try: - if kind == "folder": - gdown.download_folder( - url=url, output=str(target), quiet=False, remaining_ok=True, - ) - else: - gdown.download(id=drive_id, output=str(target) + "/", quiet=False) - files = sorted(str(p.relative_to(args.out_dir)) for p in target.rglob("*") if p.is_file()) - result["files"] = files - result["status"] = "downloaded" if files else "empty" - except Exception as exc: # noqa: BLE001 - result["error"] = str(exc) - results.append(result) - - report = { - "manifest": args.manifest.name, - "drive_links": len(results), - "downloaded": sum(r["status"] == "downloaded" for r in results), - "failed_or_empty": sum(r["status"] != "downloaded" for r in results), - "results": results, - } - args.report.parent.mkdir(parents=True, exist_ok=True) - args.report.write_text(json.dumps(report, ensure_ascii=False, indent=2), encoding="utf-8") - print(json.dumps({k: v for k, v in report.items() if k != "results"}, ensure_ascii=False)) - return 0 if report["downloaded"] else 1 - - -if __name__ == "__main__": - raise SystemExit(main()) diff --git a/tests/test_clean_ai_samples.py b/tests/test_clean_ai_samples.py new file mode 100644 index 0000000..3b9e9ac --- /dev/null +++ b/tests/test_clean_ai_samples.py @@ -0,0 +1,12 @@ +from scripts.clean_ai_samples import rejection_reason + + +def test_rejects_assistant_greeting_and_prompt_leak(): + body = "삶을 돌아보며 기억을 적었다. " * 30 + assert rejection_reason("안녕하세요. " + body, 100, 2000, 0.5) == "assistant_greeting" + assert rejection_reason(body + " 본문만 출력", 100, 2000, 0.5) == "prompt_leak" + + +def test_accepts_clean_korean_prose(): + body = "삶을 돌아보며 그날의 기억과 가족의 목소리를 기록했다. " * 20 + assert rejection_reason(body, 100, 2000, 0.5) is None diff --git a/tests/test_download_life_writing_drive.py b/tests/test_download_life_writing_drive.py deleted file mode 100644 index 2a40e22..0000000 --- a/tests/test_download_life_writing_drive.py +++ /dev/null @@ -1,9 +0,0 @@ -from scripts.download_life_writing_drive import drive_kind - - -def test_drive_kind_parses_file_and_folder(): - assert drive_kind("https://drive.google.com/file/d/abc123/view") == ("file", "abc123") - assert drive_kind("https://drive.google.com/drive/folders/xyz789?usp=drive_link") == ( - "folder", "xyz789" - ) - assert drive_kind("https://example.com") is None