wuyang6
142 lines
4.8 KiB
Python
142 lines
4.8 KiB
Python
#!/usr/bin/env python3
|
|
"""生成单句精确干预或正则干预候选 TSV，并返回校验结果。"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import json
|
|
import subprocess
|
|
import sys
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
from intervention_common import (
|
|
DEFAULT_EXACT_SOURCE,
|
|
DEFAULT_REGEX_SOURCE,
|
|
build_query_pattern,
|
|
build_source_summary,
|
|
exact_tsv_line,
|
|
infer_configured_dispatch_domain,
|
|
infer_dispatch_domain_for_device,
|
|
json_dumps,
|
|
normalize_device,
|
|
normalize_target,
|
|
regex_tsv_line,
|
|
)
|
|
|
|
|
|
SCRIPT_DIR = Path(__file__).resolve().parent
|
|
VALIDATE_SCRIPT = SCRIPT_DIR / "validate_intervention_records.py"
|
|
|
|
|
|
def load_payload(path: str | None) -> dict[str, Any]:
|
|
text = Path(path).read_text(encoding="utf-8") if path else sys.stdin.read()
|
|
if not text.strip():
|
|
raise ValueError("输入 JSON 不能为空")
|
|
return json.loads(text)
|
|
|
|
|
|
def run_validator(mode: str, tsv: str, exact_source: str, regex_source: str) -> dict[str, Any]:
|
|
proc = subprocess.run(
|
|
[
|
|
sys.executable,
|
|
str(VALIDATE_SCRIPT),
|
|
"--mode",
|
|
mode,
|
|
"--exact-source",
|
|
exact_source,
|
|
"--regex-source",
|
|
regex_source,
|
|
],
|
|
input=tsv,
|
|
text=True,
|
|
stdout=subprocess.PIPE,
|
|
stderr=subprocess.PIPE,
|
|
check=False,
|
|
)
|
|
if not proc.stdout.strip():
|
|
return {
|
|
"valid": False,
|
|
"errors": [proc.stderr.strip() or f"校验脚本退出：{proc.returncode}"],
|
|
}
|
|
return json.loads(proc.stdout)
|
|
|
|
|
|
def generate_exact(payload: dict[str, Any], exact_source: str, regex_source: str) -> dict[str, Any]:
|
|
device = normalize_device(str(payload.get("device", "")))
|
|
query = str(payload.get("query", "")).strip()
|
|
target = normalize_target(str(payload.get("target") or payload.get("code") or ""))
|
|
dispatch_domain = str(payload.get("dispatch_domain") or payload.get("domain") or "").strip()
|
|
|
|
summary = build_source_summary(Path(exact_source), Path(regex_source))
|
|
inferred = None
|
|
configured = infer_configured_dispatch_domain(target, device, summary)
|
|
if not dispatch_domain:
|
|
inferred = infer_dispatch_domain_for_device(target, device, summary)
|
|
if inferred:
|
|
dispatch_domain = inferred["dispatch_domain"]
|
|
|
|
tsv = exact_tsv_line(device, query, target, dispatch_domain)
|
|
validation = run_validator("exact", tsv, exact_source, regex_source)
|
|
return {
|
|
"mode": "exact",
|
|
"tsv": tsv,
|
|
"record": {
|
|
"device": device,
|
|
"query": query,
|
|
"target": target,
|
|
"dispatch_domain": dispatch_domain,
|
|
},
|
|
"inferred_dispatch_domain": inferred,
|
|
"configured_dispatch_domain": configured,
|
|
"validation": validation,
|
|
}
|
|
|
|
|
|
def generate_regex(payload: dict[str, Any], exact_source: str, regex_source: str) -> dict[str, Any]:
|
|
device = normalize_device(str(payload.get("device", "")))
|
|
target = normalize_target(str(payload.get("target") or payload.get("label") or ""))
|
|
raw_pattern = str(payload.get("pattern") or payload.get("regex") or "").strip()
|
|
query = str(payload.get("query") or payload.get("current_query") or "").strip()
|
|
before_queries = payload.get("before_queries") or []
|
|
if not isinstance(before_queries, list):
|
|
raise ValueError("before_queries 必须是字符串数组")
|
|
pattern = raw_pattern or build_query_pattern(query, [str(item) for item in before_queries])
|
|
|
|
tsv = regex_tsv_line(device, pattern, target)
|
|
validation = run_validator("regex", tsv, exact_source, regex_source)
|
|
return {
|
|
"mode": "regex",
|
|
"tsv": tsv,
|
|
"record": {
|
|
"device": device,
|
|
"pattern": pattern,
|
|
"target": target,
|
|
},
|
|
"validation": validation,
|
|
}
|
|
|
|
|
|
def main() -> int:
|
|
parser = argparse.ArgumentParser(description="生成干预候选 TSV")
|
|
parser.add_argument("--input", help="输入 JSON 文件；不传则读取 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")
|
|
args = parser.parse_args()
|
|
|
|
payload = load_payload(args.input)
|
|
mode = str(payload.get("mode", "")).strip().lower()
|
|
if mode == "exact":
|
|
result = generate_exact(payload, args.exact_source, args.regex_source)
|
|
elif mode == "regex":
|
|
result = generate_regex(payload, args.exact_source, args.regex_source)
|
|
else:
|
|
raise ValueError("mode 必须是 exact 或 regex")
|
|
|
|
print(json_dumps(result))
|
|
return 0 if result.get("validation", {}).get("valid") else 1
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|