#!/usr/bin/env python3 """校验单句精确干预和正则干预候选 TSV。""" from __future__ import annotations import argparse import json import re import sys from pathlib import Path from typing import Any from intervention_common import ( DEFAULT_EXACT_SOURCE, DEFAULT_REGEX_SOURCE, build_source_summary, domain_compatible, historical_targets, infer_configured_dispatch_domain, infer_dispatch_domain_for_device, load_exact_records, load_regex_records, normalize_device, normalize_target, parse_candidate_lines, validate_target, json_dumps, ) def load_input_text(path: str | None) -> str: if path: return Path(path).read_text(encoding="utf-8") return sys.stdin.read() def validate_exact_rows(rows: list[list[str]], args: argparse.Namespace) -> dict[str, Any]: summary = build_source_summary(Path(args.exact_source), Path(args.regex_source)) devices = set(summary["devices"]) known_historical_targets = historical_targets(summary) existing: dict[tuple[str, str], list[Any]] = {} for record in load_exact_records(Path(args.exact_source)): existing.setdefault((record.device, record.query), []).append(record) results = [] for index, row in enumerate(rows, start=1): errors: list[str] = [] warnings: list[str] = [] if len(row) != 4: errors.append(f"单句精确干预必须是 4 列，当前 {len(row)} 列") results.append({"line": index, "valid": False, "errors": errors, "warnings": warnings, "row": row}) continue device, query, target, dispatch_domain = [cell.strip() for cell in row] device = normalize_device(device) target = normalize_target(target) if device not in devices: errors.append(f"设备不在历史枚举中：{device}，可选：{', '.join(sorted(devices))}") if not query: errors.append("query 不能为空") if "\t" in query or "\n" in query or "\r" in query: errors.append("query 不能包含制表符或换行") target_result = validate_target(target) if not target_result["valid"]: if target in known_historical_targets: warnings.append("target 未通过 label-master 知识库校验，但历史干预文件中已存在；建议后续补齐标签知识库") target_result["normalized_output"] = target else: errors.extend([f"target 不合法：{message}" for message in target_result.get("errors", [])]) if not dispatch_domain: errors.append("下发domain 不能为空") inferred = infer_dispatch_domain_for_device(target, device, summary) if inferred and dispatch_domain and dispatch_domain != inferred["dispatch_domain"]: message = ( f"下发domain 与推断映射不一致：当前 {dispatch_domain}，" f"推断为 {inferred['dispatch_domain']}，来源 {inferred.get('source', 'historical')}，置信度 {inferred.get('confidence')}" ) if args.strict_domain: errors.append(message) else: warnings.append(message) if not inferred: warnings.append("未在历史精确干预表中找到该 target 的下发domain 映射，需要人工确认") configured = infer_configured_dispatch_domain(target, device, summary) if configured and dispatch_domain and not domain_compatible(dispatch_domain, configured["dispatch_domain"]): warnings.append( f"下发domain 与 type_match_agent 配置不兼容：当前 {dispatch_domain}，" f"配置 {configured['matched_key']} -> {configured['dispatch_domain']}" ) if not configured: warnings.append("type_match_agent 下发表中未找到该 Agent tag 的设备/通用映射") conflicts = existing.get((device, query), []) for item in conflicts: if item.target == target and item.dispatch_domain == dispatch_domain: warnings.append(f"历史文件已存在相同单句干预：line {item.line_no}") else: errors.append( f"历史文件存在同设备同 query 的不同干预：line {item.line_no}，" f"{item.target} / {item.dispatch_domain}" ) results.append( { "line": index, "valid": not errors, "errors": errors, "warnings": warnings, "normalized": { "device": device, "query": query, "target": target_result.get("normalized_output", target), "dispatch_domain": dispatch_domain, }, "configured_dispatch_domain": configured, } ) return summarize_results(results) def validate_regex_rows(rows: list[list[str]], args: argparse.Namespace) -> dict[str, Any]: summary = build_source_summary(Path(args.exact_source), Path(args.regex_source)) devices = set(summary["devices"]) known_historical_targets = historical_targets(summary) existing: dict[tuple[str, str], list[Any]] = {} for record in load_regex_records(Path(args.regex_source)): existing.setdefault((record.device, record.pattern), []).append(record) results = [] for index, row in enumerate(rows, start=1): errors: list[str] = [] warnings: list[str] = [] if len(row) != 3: errors.append(f"正则干预必须是 3 列，当前 {len(row)} 列") results.append({"line": index, "valid": False, "errors": errors, "warnings": warnings, "row": row}) continue device, pattern, target = [cell.strip() for cell in row] device = normalize_device(device) target = normalize_target(target) if device not in devices: errors.append(f"设备不在历史枚举中：{device}，可选：{', '.join(sorted(devices))}") if not pattern: errors.append("正则不能为空") else: try: re.compile(pattern) except re.error as exc: errors.append(f"正则无法编译：{exc}") if "query#" not in pattern: errors.append("正则必须包含 query# 片段") if not pattern.startswith("^") or not pattern.endswith("$"): warnings.append("历史常见正则通常使用 ^...$ 完整锚定，建议确认是否需要锚定") target_result = validate_target(target) if not target_result["valid"]: if target in known_historical_targets: warnings.append("target 未通过 label-master 知识库校验，但历史干预文件中已存在；建议后续补齐标签知识库") target_result["normalized_output"] = target else: errors.extend([f"target 不合法：{message}" for message in target_result.get("errors", [])]) conflicts = existing.get((device, pattern), []) for item in conflicts: if item.target == target: warnings.append(f"历史文件已存在相同正则干预：line {item.line_no}") else: errors.append( f"历史文件存在同设备同正则的不同干预：line {item.line_no}，{item.target}" ) results.append( { "line": index, "valid": not errors, "errors": errors, "warnings": warnings, "normalized": { "device": device, "pattern": pattern, "target": target_result.get("normalized_output", target), }, } ) return summarize_results(results) def summarize_results(results: list[dict[str, Any]]) -> dict[str, Any]: return { "valid": all(item["valid"] for item in results), "total": len(results), "invalid": sum(1 for item in results if not item["valid"]), "warning_count": sum(len(item.get("warnings", [])) for item in results), "results": results, } def main() -> int: parser = argparse.ArgumentParser(description="校验干预 TSV 候选") parser.add_argument("--mode", choices=["exact", "regex"], required=True, help="干预类型") parser.add_argument("--input", help="候选 TSV 文件路径；不传则读取 stdin") parser.add_argument("--exact-source", default=str(DEFAULT_EXACT_SOURCE), help="历史单句精确干预 TSV") parser.add_argument("--regex-source", default=str(DEFAULT_REGEX_SOURCE), help="历史正则干预 TSV") parser.add_argument("--strict-domain", action="store_true", help="下发domain 和历史多数映射不一致时直接报错") args = parser.parse_args() text = load_input_text(args.input) rows = parse_candidate_lines(text) if args.mode == "exact": payload = validate_exact_rows(rows, args) else: payload = validate_regex_rows(rows, args) print(json_dumps(payload)) return 0 if payload["valid"] else 1 if __name__ == "__main__": raise SystemExit(main())