Blame for scripts/dataset/audit_multisource_raw10k_quality.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
DGX AI Factory - Multi-source Raw JSONL Quality Auditor

기능
- 입력 JSONL 자동 탐색 또는 --input 지정
- JSON/스키마/빈 응답/중복/오류 페이지/플레이스홀더/반복/코드펜스 검사
- systemd / CUDA 관련성 검사
- clean / review / rejected 분리
- 정규화된 clean 학습 파일과 JSON 보고서 생성

표준 라이브러리만 사용합니다.
"""

from __future__ import annotations

import argparse
import hashlib
import json
import math
import re
import sys
from collections import Counter, defaultdict
from dataclasses import dataclass, asdict
from pathlib import Path
from typing import Any, Iterable, Optional


DEFAULT_EXPERIMENT_DIR = Path(
    "/home/harness_user_1/dgx_ai_factory/102_qwen3_4b_mobile/"
    "experiments/qwen3_4b_multisource_raw10k_v1_1"
)

EXCLUDED_NAME_PARTS = (
    "clean",
    "review",
    "reject",
    "audit",
    "report",
    "filtered",
    "normalized",
)

PREFERRED_NAME_PARTS = (
    "raw",
    "collected",
    "collection",
    "dataset",
    "final",
    "train",
)

PROMPT_KEYS = (
    "instruction",
    "question",
    "prompt",
    "query",
    "user",
    "problem",
    "task",
    "input_text",
)

ANSWER_KEYS = (
    "output",
    "answer",
    "response",
    "assistant",
    "completion",
    "solution",
    "target",
    "output_text",
)

INPUT_KEYS = ("input", "context", "additional_input")

CATEGORY_KEYS = ("category", "domain", "topic", "label", "task_type")
SOURCE_KEYS = ("source", "dataset", "origin", "provider", "repo", "dataset_name")

SYSTEMD_TERMS = (
    "systemd", "systemctl", "journalctl", "unit file", "service file",
    "daemon-reload", "execstart", "wantedby", "restart=", "status=217",
    "/etc/systemd/system", "서비스 유닛", "시스템디",
)

CUDA_TERMS = (
    "cuda", "nvidia-smi", "nvcc", "cudnn", "nccl", "gpu", "vram",
    "out of memory", "oom", "torch.cuda", "cuda_visible_devices",
    "메모리 부족", "gpu 메모리", "cuda oom",
)

ERROR_PAYLOAD_TERMS = (
    "access denied",
    "forbidden",
    "not found",
    "too many requests",
    "rate limit exceeded",
    "internal server error",
    "bad gateway",
    "service unavailable",
    "dataset viewer is disabled",
    "dataset viewer is not supported",
    "configuration not found",
    "split not found",
    "invalid token",
)

BOILERPLATE_TERMS = (
    "as an ai language model",
    "i cannot browse",
    "i don't have access",
    "consult the documentation",
    "please refer to the documentation",
    "도움을 드릴 수 없습니다",
    "정확한 정보는 공식 문서를",
)

PLACEHOLDER_PATTERNS = (
    re.compile(r"\{[A-Za-z_][A-Za-z0-9_ -]{1,40}\}"),
    re.compile(r"<(?:placeholder|insert|your_|todo)[^>]*>", re.I),
    re.compile(r"\[(?:insert|placeholder|your |todo)[^\]]*\]", re.I),
    re.compile(r"\b(?:TODO|TBD|FIXME)\b"),
    re.compile(r"\$\{[^}]+\}"),
)

CHINESE_CHAR_RE = re.compile(r"[\u3400-\u4DBF\u4E00-\u9FFF]")
HTML_PAGE_RE = re.compile(r"(?is)<(?:!doctype\s+html|html|head|body)\b")
URL_RE = re.compile(r"https?://\S+")
WORD_RE = re.compile(r"[A-Za-z0-9가-힣_./:+#-]+")


@dataclass
class AuditResult:
    line_no: int
    status: str
    score: int
    hard_reject: bool
    reasons: list[str]
    prompt: str
    answer: str
    context: str
    category_declared: str
    category_inferred: str
    source: str
    exact_hash: str
    prompt_hash: str


def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(
        description="Raw JSONL 데이터 품질 감사 및 필터링"
    )
    parser.add_argument(
        "--experiment-dir",
        type=Path,
        default=DEFAULT_EXPERIMENT_DIR,
        help=f"실험 디렉터리 기본값: {DEFAULT_EXPERIMENT_DIR}",
    )
    parser.add_argument(
        "--input",
        type=Path,
        default=None,
        help="검사할 JSONL 파일. 생략하면 실험 디렉터리에서 자동 탐색합니다.",
    )
    parser.add_argument(
        "--output-dir",
        type=Path,
        default=None,
        help="출력 디렉터리. 기본값: <experiment-dir>/quality_audit",
    )
    parser.add_argument(
        "--min-clean-score",
        type=int,
        default=80,
        help="clean 최소 점수. 기본값 80",
    )
    parser.add_argument(
        "--min-review-score",
        type=int,
        default=55,
        help="review 최소 점수. 기본값 55",
    )
    parser.add_argument(
        "--max-samples",
        type=int,
        default=30,
        help="보고서에 포함할 상태별 예시 최대 개수",
    )
    return parser.parse_args()


def compact_json(value: Any) -> str:
    try:
        return json.dumps(value, ensure_ascii=False, sort_keys=True)
    except Exception:
        return repr(value)


def safe_string(value: Any) -> str:
    if value is None:
        return ""
    if isinstance(value, str):
        return value.strip()
    if isinstance(value, (int, float, bool)):
        return str(value)
    if isinstance(value, list):
        parts = [safe_string(x) for x in value]
        return "\n".join(x for x in parts if x).strip()
    if isinstance(value, dict):
        return compact_json(value)
    return str(value).strip()


def first_nonempty(obj: dict[str, Any], keys: Iterable[str]) -> str:
    for key in keys:
        if key in obj:
            text = safe_string(obj.get(key))
            if text:
                return text
    return ""


def extract_from_messages(messages: Any) -> tuple[str, str, str]:
    if not isinstance(messages, list):
        return "", "", ""

    user_parts: list[str] = []
    assistant_parts: list[str] = []
    context_parts: list[str] = []

    for item in messages:
        if not isinstance(item, dict):
            continue
        role = safe_string(item.get("role")).lower()
        content = item.get("content")
        if isinstance(content, list):
            text_chunks: list[str] = []
            for block in content:
                if isinstance(block, dict):
                    if block.get("type") in {"text", "input_text", "output_text"}:
                        text_chunks.append(safe_string(block.get("text")))
                    else:
                        text_chunks.append(safe_string(block))
                else:
                    text_chunks.append(safe_string(block))
            text = "\n".join(x for x in text_chunks if x).strip()
        else:
            text = safe_string(content)

        if not text:
            continue
        if role in {"user", "human"}:
            user_parts.append(text)
        elif role in {"assistant", "gpt", "model"}:
            assistant_parts.append(text)
        elif role in {"system", "developer", "context"}:
            context_parts.append(text)

    prompt = "\n\n".join(user_parts).strip()
    answer = "\n\n".join(assistant_parts).strip()
    context = "\n\n".join(context_parts).strip()
    return prompt, answer, context


def extract_record(obj: Any) -> tuple[str, str, str, str, str]:
    if not isinstance(obj, dict):
        return "", "", "", "", ""

    prompt, answer, context = extract_from_messages(obj.get("messages"))

    if not prompt:
        prompt = first_nonempty(obj, PROMPT_KEYS)
    if not answer:
        answer = first_nonempty(obj, ANSWER_KEYS)
    if not context:
        context = first_nonempty(obj, INPUT_KEYS)

    # Hugging Face의 흔한 nested 구조를 추가 지원합니다.
    if (not prompt or not answer) and isinstance(obj.get("data"), dict):
        nested = obj["data"]
        if not prompt:
            prompt = first_nonempty(nested, PROMPT_KEYS)
        if not answer:
            answer = first_nonempty(nested, ANSWER_KEYS)
        if not context:
            context = first_nonempty(nested, INPUT_KEYS)

    if (not prompt or not answer) and isinstance(obj.get("conversation"), list):
        p2, a2, c2 = extract_from_messages(obj.get("conversation"))
        prompt = prompt or p2
        answer = answer or a2
        context = context or c2

    category = first_nonempty(obj, CATEGORY_KEYS).lower()
    source = first_nonempty(obj, SOURCE_KEYS)
    return prompt, answer, context, category, source


def normalize_for_hash(text: str) -> str:
    text = text.lower()
    text = re.sub(r"https?://\S+", "<url>", text)
    text = re.sub(r"\s+", " ", text)
    text = re.sub(r"[`*_>#|~-]+", "", text)
    return text.strip()


def sha256_text(text: str) -> str:
    return hashlib.sha256(text.encode("utf-8", errors="replace")).hexdigest()


def infer_category(prompt: str, answer: str, declared: str) -> str:
    text = f"{declared}\n{prompt}\n{answer}".lower()
    systemd_score = sum(1 for term in SYSTEMD_TERMS if term in text)
    cuda_score = sum(1 for term in CUDA_TERMS if term in text)

    if systemd_score == 0 and cuda_score == 0:
        return "other"
    if systemd_score > cuda_score:
        return "systemd"
    if cuda_score > systemd_score:
        return "cuda"
    if "systemd" in declared:
        return "systemd"
    if "cuda" in declared or "gpu" in declared:
        return "cuda"
    return "mixed"


def repeated_line_ratio(text: str) -> float:
    lines = [re.sub(r"\s+", " ", x.strip()) for x in text.splitlines() if x.strip()]
    if len(lines) < 4:
        return 0.0
    counts = Counter(lines)
    repeated = sum(count - 1 for count in counts.values() if count > 1)
    return repeated / len(lines)


def token_repetition_ratio(text: str) -> float:
    words = WORD_RE.findall(text.lower())
    if len(words) < 30:
        return 0.0
    unique = len(set(words))
    return 1.0 - (unique / len(words))


def answer_is_url_only(answer: str) -> bool:
    stripped = URL_RE.sub("", answer)
    stripped = re.sub(r"[\s\-–—:;,.()\[\]{}]+", "", stripped)
    return bool(URL_RE.search(answer)) and len(stripped) < 20


def category_relevance(category: str, prompt: str, answer: str) -> tuple[int, list[str]]:
    text = f"{prompt}\n{answer}".lower()
    reasons: list[str] = []
    penalty = 0

    if category == "systemd":
        hits = sum(1 for term in SYSTEMD_TERMS if term in text)
        if hits == 0:
            penalty += 30
            reasons.append("systemd_relevance_zero")
        elif hits == 1:
            penalty += 10
            reasons.append("systemd_relevance_weak")
    elif category == "cuda":
        hits = sum(1 for term in CUDA_TERMS if term in text)
        if hits == 0:
            penalty += 30
            reasons.append("cuda_relevance_zero")
        elif hits == 1:
            penalty += 10
            reasons.append("cuda_relevance_weak")

    return penalty, reasons


def audit_record(
    obj: Any,
    line_no: int,
    seen_exact: set[str],
    seen_prompt: dict[str, int],
    min_clean_score: int,
    min_review_score: int,
) -> AuditResult:
    prompt, answer, context, declared, source = extract_record(obj)
    inferred = infer_category(prompt, answer, declared)

    normalized_prompt = normalize_for_hash(prompt)
    normalized_answer = normalize_for_hash(answer)
    exact_hash = sha256_text(normalized_prompt + "\n---\n" + normalized_answer)
    prompt_hash = sha256_text(normalized_prompt)

    score = 100
    reasons: list[str] = []
    hard_reject = False

    if not isinstance(obj, dict):
        hard_reject = True
        score = 0
        reasons.append("record_not_object")

    if not prompt:
        hard_reject = True
        score = 0
        reasons.append("missing_prompt")

    if not answer:
        hard_reject = True
        score = 0
        reasons.append("missing_answer")

    if prompt and len(prompt.strip()) < 8:
        score -= 25
        reasons.append("prompt_too_short")
    elif prompt and len(prompt.strip()) < 20:
        score -= 10
        reasons.append("prompt_short")

    if answer and len(answer.strip()) < 20:
        hard_reject = True
        score = min(score, 20)
        reasons.append("answer_too_short")
    elif answer and len(answer.strip()) < 80:
        score -= 25
        reasons.append("answer_short")
    elif answer and len(answer.strip()) < 160:
        score -= 10
        reasons.append("answer_somewhat_short")

    combined = f"{prompt}\n{answer}".strip()
    combined_lower = combined.lower()

    if HTML_PAGE_RE.search(combined):
        hard_reject = True
        score = min(score, 10)
        reasons.append("html_error_page")

    error_hits = [term for term in ERROR_PAYLOAD_TERMS if term in combined_lower]
    if error_hits and len(answer) < 800:
        hard_reject = True
        score = min(score, 15)
        reasons.append("api_or_viewer_error_payload")

    boilerplate_hits = [term for term in BOILERPLATE_TERMS if term in combined_lower]
    if boilerplate_hits:
        score -= 25
        reasons.append("generic_ai_boilerplate")

    if any(pattern.search(combined) for pattern in PLACEHOLDER_PATTERNS):
        score -= 35
        reasons.append("unfilled_placeholder")

    if answer.count("```") % 2 != 0:
        score -= 15
        reasons.append("unclosed_code_fence")

    if answer_is_url_only(answer):
        hard_reject = True
        score = min(score, 20)
        reasons.append("reference_or_url_only")

    chinese_count = len(CHINESE_CHAR_RE.findall(combined))
    if chinese_count >= 3:
        score -= 20
        reasons.append("possible_chinese_leakage")
    elif chinese_count > 0:
        score -= 8
        reasons.append("han_character_present")

    line_repeat = repeated_line_ratio(answer)
    if line_repeat >= 0.35:
        score -= 30
        reasons.append("severe_repeated_lines")
    elif line_repeat >= 0.15:
        score -= 15
        reasons.append("repeated_lines")

    token_repeat = token_repetition_ratio(answer)
    if len(WORD_RE.findall(answer)) >= 80 and token_repeat >= 0.78:
        score -= 15
        reasons.append("low_lexical_diversity")

    # 완전히 같은 prompt+answer는 불량 중복으로 제외합니다.
    if exact_hash in seen_exact:
        hard_reject = True
        score = 0
        reasons.append("exact_duplicate")
    else:
        seen_exact.add(exact_hash)

    # 같은 질문에 다른 답변이 반복되면 검토 대상으로 보냅니다.
    prior_prompt_line = seen_prompt.get(prompt_hash)
    if prior_prompt_line is not None and "exact_duplicate" not in reasons:
        score -= 20
        reasons.append("duplicate_prompt")
    else:
        seen_prompt[prompt_hash] = line_no

    declared_for_check = declared
    if "systemd" in declared_for_check:
        declared_for_check = "systemd"
    elif "cuda" in declared_for_check or "gpu" in declared_for_check:
        declared_for_check = "cuda"
    elif inferred in {"systemd", "cuda"}:
        declared_for_check = inferred
    else:
        declared_for_check = ""

    relevance_penalty, relevance_reasons = category_relevance(
        declared_for_check, prompt, answer
    )
    score -= relevance_penalty
    reasons.extend(relevance_reasons)

    # 지나치게 긴 단일 레코드는 학습 전 별도 검토가 안전합니다.
    if len(answer) > 30000:
        score -= 20
        reasons.append("answer_extremely_long")
    elif len(answer) > 12000:
        score -= 10
        reasons.append("answer_very_long")

    score = max(0, min(100, score))

    if hard_reject or score < min_review_score:
        status = "rejected"
    elif score < min_clean_score:
        status = "review"
    else:
        status = "clean"

    return AuditResult(
        line_no=line_no,
        status=status,
        score=score,
        hard_reject=hard_reject,
        reasons=sorted(set(reasons)),
        prompt=prompt,
        answer=answer,
        context=context,
        category_declared=declared,
        category_inferred=inferred,
        source=source,
        exact_hash=exact_hash,
        prompt_hash=prompt_hash,
    )


def count_lines_fast(path: Path) -> int:
    count = 0
    with path.open("rb") as handle:
        for _ in handle:
            count += 1
    return count


def discover_input(experiment_dir: Path) -> Path:
    candidates: list[Path] = []

    search_roots = [
        experiment_dir,
        experiment_dir / "data",
        experiment_dir / "raw",
        experiment_dir / "outputs",
        experiment_dir.parent.parent / "data",
    ]

    seen: set[Path] = set()
    for root in search_roots:
        if not root.exists():
            continue
        for path in root.rglob("*.jsonl"):
            try:
                resolved = path.resolve()
            except OSError:
                continue
            if resolved in seen:
                continue
            seen.add(resolved)

            name_lower = path.name.lower()
            if any(part in name_lower for part in EXCLUDED_NAME_PARTS):
                continue
            candidates.append(path)

    if not candidates:
        node-7.example.invalid FileNotFoundError(
            f"JSONL 입력 파일을 찾지 못했습니다: {experiment_dir}\n"
            "--input /절대/경로/파일.jsonl 로 지정하십시오."
        )

    ranked: list[tuple[int, int, float, Path]] = []
    for path in candidates:
        name_lower = path.name.lower()
        preferred = sum(1 for part in PREFERRED_NAME_PARTS if part in name_lower)
        try:
            size = path.stat().st_size
            mtime = path.stat().st_mtime
        except OSError:
            continue
        ranked.append((preferred, size, mtime, path))

    if not ranked:
        node-7.example.invalid FileNotFoundError("읽을 수 있는 JSONL 후보가 없습니다.")

    ranked.sort(reverse=True)
    return ranked[0][3]


def jsonl_write(handle, obj: Any) -> None:
    handle.write(json.dumps(obj, ensure_ascii=False) + "\n")


def safe_original(obj: Any) -> dict[str, Any]:
    if isinstance(obj, dict):
        return obj
    return {"_raw_value": obj}


def sample_view(result: AuditResult) -> dict[str, Any]:
    return {
        "line_no": result.line_no,
        "status": result.status,
        "score": result.score,
        "hard_reject": result.hard_reject,
        "reasons": result.reasons,
        "category_declared": result.category_declared,
        "category_inferred": result.category_inferred,
        "source": result.source,
        "prompt_preview": result.prompt[:500],
        "answer_preview": result.answer[:800],
    }


def determine_gate(
    total: int,
    clean: int,
    review: int,
    rejected: int,
    hard_rejects: int,
    duplicate_count: int,
) -> tuple[str, list[str]]:
    if total <= 0:
        return "FAIL", ["no_valid_input_rows"]

    clean_rate = clean / total
    rejected_rate = rejected / total
    hard_rate = hard_rejects / total
    dup_rate = duplicate_count / total

    reasons: list[str] = []

    if clean_rate < 0.65:
        reasons.append("clean_rate_below_65_percent")
    if rejected_rate > 0.20:
        reasons.append("rejected_rate_above_20_percent")
    if hard_rate > 0.12:
        reasons.append("hard_reject_rate_above_12_percent")
    if dup_rate > 0.10:
        reasons.append("duplicate_rate_above_10_percent")

    if (
        clean_rate >= 0.80
        and rejected_rate <= 0.10
        and hard_rate <= 0.05
        and dup_rate <= 0.05
    ):
        return "PASS", ["quality_gate_passed"]

    if clean_rate >= 0.65 and hard_rate <= 0.12:
        return "HOLD", reasons or ["manual_review_required"]

    return "FAIL", reasons or ["quality_gate_failed"]


def main() -> int:
    args = parse_args()
    experiment_dir = args.experiment_dir.expanduser().resolve()
    input_path = (
        args.input.expanduser().resolve()
        if args.input is not None
        else discover_input(experiment_dir)
    )

    if not input_path.is_file():
        print(f"[ERROR] 입력 파일이 없습니다: {input_path}", file=sys.stderr)
        return 2

    output_dir = (
        args.output_dir.expanduser().resolve()
        if args.output_dir is not None
        else experiment_dir / "quality_audit"
    )
    output_dir.mkdir(parents=True, exist_ok=True)

    clean_path = output_dir / "clean.jsonl"
    review_path = output_dir / "review.jsonl"
    rejected_path = output_dir / "rejected.jsonl"
    normalized_path = output_dir / "clean_normalized_train.jsonl"
    report_path = output_dir / "quality_audit_report.json"
    summary_path = output_dir / "quality_audit_summary.txt"

    total_lines = count_lines_fast(input_path)
    print(f"[INPUT] {input_path}")
    print(f"[TOTAL_LINES] {total_lines}")
    print(f"[OUTPUT_DIR] {output_dir}")

    status_counts: Counter[str] = Counter()
    reason_counts: Counter[str] = Counter()
    inferred_counts: Counter[str] = Counter()
    declared_counts: Counter[str] = Counter()
    source_counts: Counter[str] = Counter()
    score_buckets: Counter[str] = Counter()
    samples: dict[str, list[dict[str, Any]]] = defaultdict(list)

    seen_exact: set[str] = set()
    seen_prompt: dict[str, int] = {}

    invalid_json_count = 0
    processed_count = 0
    hard_rejects = 0

    with (
        input_path.open("r", encoding="utf-8", errors="replace") as src,
        clean_path.open("w", encoding="utf-8") as clean_out,
        review_path.open("w", encoding="utf-8") as review_out,
        rejected_path.open("w", encoding="utf-8") as rejected_out,
        normalized_path.open("w", encoding="utf-8") as normalized_out,
    ):
        for line_no, raw_line in enumerate(src, start=1):
            raw_line = raw_line.strip()

            if not raw_line:
                invalid_json_count += 1
                status_counts["rejected"] += 1
                hard_rejects += 1
                reason_counts["blank_line"] += 1
                rejected_obj = {
                    "_audit": {
                        "line_no": line_no,
                        "status": "rejected",
                        "score": 0,
                        "hard_reject": True,
                        "reasons": ["blank_line"],
                    },
                    "_raw_line": "",
                }
                jsonl_write(rejected_out, rejected_obj)
                continue

            try:
                obj = json.loads(raw_line)
            except json.JSONDecodeError as exc:
                invalid_json_count += 1
                status_counts["rejected"] += 1
                hard_rejects += 1
                reason_counts["invalid_json"] += 1
                rejected_obj = {
                    "_audit": {
                        "line_no": line_no,
                        "status": "rejected",
                        "score": 0,
                        "hard_reject": True,
                        "reasons": ["invalid_json"],
                        "json_error": str(exc),
                    },
                    "_raw_line": raw_line[:5000],
                }
                jsonl_write(rejected_out, rejected_obj)
                if len(samples["rejected"]) < args.max_samples:
                    samples["rejected"].append(
                        {
                            "line_no": line_no,
                            "status": "rejected",
                            "score": 0,
                            "hard_reject": True,
                            "reasons": ["invalid_json"],
                            "raw_preview": raw_line[:800],
                        }
                    )
                continue

            result = audit_record(
                obj=obj,
                line_no=line_no,
                seen_exact=seen_exact,
                seen_prompt=seen_prompt,
                min_clean_score=args.min_clean_score,
                min_review_score=args.min_review_score,
            )
            processed_count += 1
            status_counts[result.status] += 1
            if result.hard_reject:
                hard_rejects += 1

            for reason in result.reasons:
                reason_counts[reason] += 1

            inferred_counts[result.category_inferred or "unknown"] += 1
            declared_counts[result.category_declared or "unknown"] += 1
            source_counts[result.source or "unknown"] += 1

            if result.score >= 90:
                score_buckets["90-100"] += 1
            elif result.score >= 80:
                score_buckets["80-89"] += 1
            elif result.score >= 70:
                score_buckets["70-79"] += 1
            elif result.score >= 55:
                score_buckets["55-69"] += 1
            else:
                score_buckets["0-54"] += 1

            wrapped = safe_original(obj).copy()
            wrapped["_audit"] = {
                "line_no": result.line_no,
                "status": result.status,
                "score": result.score,
                "hard_reject": result.hard_reject,
                "reasons": result.reasons,
                "category_declared": result.category_declared,
                "category_inferred": result.category_inferred,
                "source": result.source,
                "exact_hash": result.exact_hash,
                "prompt_hash": result.prompt_hash,
            }

            if result.status == "clean":
                jsonl_write(clean_out, wrapped)
                normalized = {
                    "instruction": result.prompt,
                    "input": result.context,
                    "output": result.answer,
                    "category": (
                        result.category_declared
                        if result.category_declared
                        else result.category_inferred
                    ),
                    "source": result.source,
                    "quality_score": result.score,
                    "quality_reasons": result.reasons,
                    "source_line": result.line_no,
                }
                jsonl_write(normalized_out, normalized)
            elif result.status == "review":
                jsonl_write(review_out, wrapped)
            else:
                jsonl_write(rejected_out, wrapped)

            if len(samples[result.status]) < args.max_samples:
                samples[result.status].append(sample_view(result))

            if total_lines > 0 and (line_no % 500 == 0 or line_no == total_lines):
                pct = (line_no / total_lines) * 100
                print(
                    f"[PROGRESS] {line_no}/{total_lines} ({pct:.1f}%) "
                    f"clean={status_counts['clean']} "
                    f"review={status_counts['review']} "
                    f"rejected={status_counts['rejected']}"
                )

    total = sum(status_counts.values())
    duplicate_count = (
        reason_counts["exact_duplicate"] + reason_counts["duplicate_prompt"]
    )
    gate_status, gate_reasons = determine_gate(
        total=total,
        clean=status_counts["clean"],
        review=status_counts["review"],
        rejected=status_counts["rejected"],
        hard_rejects=hard_rejects,
        duplicate_count=duplicate_count,
    )

    def rate(value: int) -> float:
        return round((value / total * 100), 4) if total else 0.0

    report = {
        "status": gate_status,
        "gate_reasons": gate_reasons,
        "input_file": str(input_path),
        "output_dir": str(output_dir),
        "thresholds": {
            "min_clean_score": args.min_clean_score,
            "min_review_score": args.min_review_score,
        },
        "counts": {
            "total_lines": total_lines,
            "total_classified": total,
            "parsed_records": processed_count,
            "invalid_or_blank_lines": invalid_json_count,
            "clean": status_counts["clean"],
            "review": status_counts["review"],
            "rejected": status_counts["rejected"],
            "hard_rejects": hard_rejects,
            "duplicate_flags": duplicate_count,
        },
        "rates_percent": {
            "clean": rate(status_counts["clean"]),
            "review": rate(status_counts["review"]),
            "rejected": rate(status_counts["rejected"]),
            "hard_rejects": rate(hard_rejects),
            "duplicate_flags": rate(duplicate_count),
        },
        "reason_counts": dict(reason_counts.most_common()),
        "category_inferred_counts": dict(inferred_counts.most_common()),
        "category_declared_counts": dict(declared_counts.most_common()),
        "source_counts": dict(source_counts.most_common()),
        "score_buckets": dict(score_buckets),
        "samples": samples,
        "outputs": {
            "clean": str(clean_path),
            "review": str(review_path),
            "rejected": str(rejected_path),
            "clean_normalized_train": str(normalized_path),
            "report": str(report_path),
            "summary": str(summary_path),
        },
    }

    report_path.write_text(
        json.dumps(report, ensure_ascii=False, indent=2),
        encoding="utf-8",
    )

    summary_lines = [
        "==== DGX RAW DATA QUALITY AUDIT ====",
        f"STATUS={gate_status}",
        f"INPUT={input_path}",
        f"TOTAL={total}",
        f"CLEAN={status_counts['clean']} ({rate(status_counts['clean']):.2f}%)",
        f"REVIEW={status_counts['review']} ({rate(status_counts['review']):.2f}%)",
        f"REJECTED={status_counts['rejected']} ({rate(status_counts['rejected']):.2f}%)",
        f"HARD_REJECTS={hard_rejects} ({rate(hard_rejects):.2f}%)",
        f"DUPLICATE_FLAGS={duplicate_count} ({rate(duplicate_count):.2f}%)",
        f"INVALID_OR_BLANK={invalid_json_count}",
        f"GATE_REASONS={','.join(gate_reasons)}",
        "",
        "==== TOP REASONS ====",
    ]
    summary_lines.extend(
        f"{reason}={count}" for reason, count in reason_counts.most_common(20)
    )
    summary_lines.extend(
        [
            "",
            "==== INFERRED CATEGORIES ====",
            *(
                f"{category}={count}"
                for category, count in inferred_counts.most_common()
            ),
            "",
            "==== OUTPUTS ====",
            f"CLEAN_FILE={clean_path}",
            f"REVIEW_FILE={review_path}",
            f"REJECTED_FILE={rejected_path}",
            f"NORMALIZED_TRAIN_FILE={normalized_path}",
            f"REPORT_FILE={report_path}",
        ]
    )

    summary_text = "\n".join(summary_lines) + "\n"
    summary_path.write_text(summary_text, encoding="utf-8")
    print()
    print(summary_text)

    if gate_status == "PASS":
        return 0
    if gate_status == "HOLD":
        return 10
    return 20


if __name__ == "__main__":
    node-7.example.invalid SystemExit(main())