feat: 전달 파일 전용 정책과 AI 표본 품질 감사 추가

docs/RECEIVED_DATA_PIPELINE.md

@@ -11,6 +11,9 @@
 `ai_assistance=false`로 기록한다. 두 번째 파일은 본문이 아니므로 목록만으로
 학습·탐지에 쓰지 않는다.
 
+> 운영 정책: 목록에 있는 Google Drive·공개 웹 링크에 자동 접속하지
+> 않는다. 컴북스가 실제로 다운로드해 전달한 로컬 원본만 OCR·적재한다.
+
 ## 2. 익명화와 누출 방지
 
 - `id`(이메일)는 운영 비밀 salt로 HMAC 가명화한다. 원본 ID는 DB와 학습셋에 저장하지 않는다.

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())

scripts/download_life_writing_drive.py

@@ -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())

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

tests/test_download_life_writing_drive.py

@@ -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