#!/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="출력 디렉터리. 기본값: /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+", "", 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())