Add intervention data skill

2026-05-19 11:59:49 +08:00
parent 201058dfb2
commit 5684df3caf
7 changed files with 1316 additions and 0 deletions
@@ -0,0 +1,141 @@
+#!/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())