100 lines
3.7 KiB
Python
100 lines
3.7 KiB
Python
#!/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())
|