zk-data-agent/skills/intervention-data/scripts/audit_intervention_sources.py

#!/usr/bin/env python3
"""审计现有干预文件和当前规则的贴合度。

这个脚本用于回答两个问题：
1. 现有单句精确干预/正则干预文件自身是否存在明显格式问题。
2. 当前 label-master 校验、type_match_agent 下发表和历史数据是否匹配。

它不会修改任何文件，只输出 JSON 摘要和少量样例。
"""

from __future__ import annotations

import argparse
import re
from collections import Counter, defaultdict
from pathlib import Path
from typing import Any

from intervention_common import (
    DEFAULT_EXACT_SOURCE,
    DEFAULT_REGEX_SOURCE,
    build_source_summary,
    domain_compatible,
    infer_configured_dispatch_domain,
    json_dumps,
    load_exact_records,
    load_label_validator,
    load_regex_records,
)


def sample_append(bucket: list[dict[str, Any]], item: dict[str, Any], limit: int) -> None:
    if len(bucket) < limit:
        bucket.append(item)


def validate_unique_targets(targets: set[str]) -> dict[str, dict[str, Any]]:
    validator = load_label_validator()
    return {target: validator.validate(target) for target in sorted(targets)}


def audit_exact_records(summary: dict[str, Any], target_results: dict[str, dict[str, Any]], sample_limit: int) -> dict[str, Any]:
    records = load_exact_records(Path(summary["source_files"]["exact"]))
    empty_fields = Counter()
    target_invalid = Counter()
    configured_mismatch = Counter()
    duplicate_keys: dict[tuple[str, str], list[Any]] = defaultdict(list)
    samples = {
        "empty_fields": [],
        "invalid_targets": [],
        "configured_domain_mismatch": [],
        "conflicts": [],
    }

    for record in records:
        if not record.device:
            empty_fields["device"] += 1
        if not record.query:
            empty_fields["query"] += 1
        if not record.target:
            empty_fields["target"] += 1
        if not record.dispatch_domain:
            empty_fields["dispatch_domain"] += 1
        if any(not value for value in (record.device, record.query, record.target, record.dispatch_domain)):
            sample_append(samples["empty_fields"], record.__dict__, sample_limit)

        result = target_results.get(record.target)
        if result and not result.get("valid"):
            target_invalid[record.target] += 1
            sample_append(
                samples["invalid_targets"],
                {
                    "line_no": record.line_no,
                    "target": record.target,
                    "errors": result.get("errors", []),
                },
                sample_limit,
            )

        configured = infer_configured_dispatch_domain(record.target, record.device, summary)
        if configured and not domain_compatible(record.dispatch_domain, configured["dispatch_domain"]):
            configured_mismatch[configured["matched_key"]] += 1
            sample_append(
                samples["configured_domain_mismatch"],
                {
                    "line_no": record.line_no,
                    "device": record.device,
                    "query": record.query,
                    "target": record.target,
                    "dispatch_domain": record.dispatch_domain,
                    "configured": configured,
                },
                sample_limit,
            )

        duplicate_keys[(record.device, record.query)].append(record)

    conflicts = []
    duplicate_count = 0
    for (device, query), items in duplicate_keys.items():
        if len(items) <= 1:
            continue
        duplicate_count += len(items)
        variants = {(item.target, item.dispatch_domain) for item in items}
        if len(variants) > 1:
            conflict = {
                "device": device,
                "query": query,
                "lines": [item.line_no for item in items[:10]],
                "variants": sorted([{"target": target, "dispatch_domain": domain} for target, domain in variants], key=str),
            }
            conflicts.append(conflict)
            sample_append(samples["conflicts"], conflict, sample_limit)

    return {
        "total": len(records),
        "empty_fields": dict(empty_fields),
        "invalid_target_total": sum(target_invalid.values()),
        "invalid_target_top": dict(target_invalid.most_common(20)),
        "configured_domain_mismatch_total": sum(configured_mismatch.values()),
        "configured_domain_mismatch_top": dict(configured_mismatch.most_common(20)),
        "duplicate_row_count_by_device_query": duplicate_count,
        "conflict_key_count": len(conflicts),
        "samples": samples,
    }


def audit_regex_records(target_results: dict[str, dict[str, Any]], summary: dict[str, Any], sample_limit: int) -> dict[str, Any]:
    records = load_regex_records(Path(summary["source_files"]["regex"]))
    empty_fields = Counter()
    regex_errors = Counter()
    anchor_warnings = 0
    missing_query_fragment = 0
    target_invalid = Counter()
    duplicate_keys: dict[tuple[str, str], list[Any]] = defaultdict(list)
    samples = {
        "empty_fields": [],
        "regex_errors": [],
        "missing_query_fragment": [],
        "invalid_targets": [],
        "conflicts": [],
    }

    for record in records:
        if not record.device:
            empty_fields["device"] += 1
        if not record.pattern:
            empty_fields["pattern"] += 1
        if not record.target:
            empty_fields["target"] += 1
        if any(not value for value in (record.device, record.pattern, record.target)):
            sample_append(samples["empty_fields"], record.__dict__, sample_limit)

        if record.pattern:
            try:
                re.compile(record.pattern)
            except re.error as exc:
                regex_errors[str(exc)] += 1
                sample_append(
                    samples["regex_errors"],
                    {"line_no": record.line_no, "pattern": record.pattern, "error": str(exc)},
                    sample_limit,
                )
            if "query#" not in record.pattern:
                missing_query_fragment += 1
                sample_append(samples["missing_query_fragment"], record.__dict__, sample_limit)
            if not record.pattern.startswith("^") or not record.pattern.endswith("$"):
                anchor_warnings += 1

        result = target_results.get(record.target)
        if result and not result.get("valid"):
            target_invalid[record.target] += 1
            sample_append(
                samples["invalid_targets"],
                {
                    "line_no": record.line_no,
                    "target": record.target,
                    "errors": result.get("errors", []),
                },
                sample_limit,
            )

        duplicate_keys[(record.device, record.pattern)].append(record)

    conflicts = []
    duplicate_count = 0
    for (device, pattern), items in duplicate_keys.items():
        if len(items) <= 1:
            continue
        duplicate_count += len(items)
        variants = {item.target for item in items}
        if len(variants) > 1:
            conflict = {
                "device": device,
                "pattern": pattern,
                "lines": [item.line_no for item in items[:10]],
                "variants": sorted(variants),
            }
            conflicts.append(conflict)
            sample_append(samples["conflicts"], conflict, sample_limit)

    return {
        "total": len(records),
        "empty_fields": dict(empty_fields),
        "regex_error_total": sum(regex_errors.values()),
        "regex_error_top": dict(regex_errors.most_common(20)),
        "missing_query_fragment": missing_query_fragment,
        "anchor_warning_count": anchor_warnings,
        "invalid_target_total": sum(target_invalid.values()),
        "invalid_target_top": dict(target_invalid.most_common(20)),
        "duplicate_row_count_by_device_pattern": duplicate_count,
        "conflict_key_count": len(conflicts),
        "samples": samples,
    }


def main() -> int:
    parser = argparse.ArgumentParser(description="审计现有干预文件")
    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("--sample-limit", type=int, default=5, help="每类问题最多输出样例数")
    args = parser.parse_args()

    summary = build_source_summary(Path(args.exact_source), Path(args.regex_source))
    exact_records = load_exact_records(Path(args.exact_source))
    regex_records = load_regex_records(Path(args.regex_source))
    target_results = validate_unique_targets({record.target for record in exact_records + regex_records if record.target})

    payload = {
        "source_files": summary["source_files"],
        "devices": summary["devices"],
        "counts": summary["counts"],
        "target_validation": {
            "unique_targets": len(target_results),
            "valid_unique_targets": sum(1 for item in target_results.values() if item.get("valid")),
            "invalid_unique_targets": sum(1 for item in target_results.values() if not item.get("valid")),
        },
        "exact": audit_exact_records(summary, target_results, args.sample_limit),
        "regex": audit_regex_records(target_results, summary, args.sample_limit),
    }
    print(json_dumps(payload))
    has_hard_errors = bool(payload["exact"]["empty_fields"] or payload["regex"]["regex_error_total"])
    return 1 if has_hard_errors else 0


if __name__ == "__main__":
    raise SystemExit(main())