wuyang6
221 lines
9.2 KiB
Python
221 lines
9.2 KiB
Python
#!/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())
|