BIDI_EXTRACTION_AI/BiddingKG/dl/rules/validator.py

# -*- coding: utf-8 -*-
"""规则校验器（RULE VALIDATOR）。

按 ARCHITECTURE.md Phase 7 要求，规则生成后必须运行三项校验：
1. **Schema 校验** — YAML 文件结构符合 ``schema/pattern_schema.json``
2. **黄金样例** — 正例必须匹配，负例必须不匹配
3. **正则性能** — 编译时间 < 100ms，单次匹配时间 < 1ms

使用方式::

    python -m BiddingKG.dl.rules.validator
    # 或
    from BiddingKG.dl.rules.validator import RuleValidator
    RuleValidator.run_all()

依赖方向：rules 只依赖 common，不反向依赖业务层。
"""
from __future__ import absolute_import

import os
import re
import time
import json
from typing import List, Dict, Tuple

import yaml

from BiddingKG.dl.rules.loader import load_rule_yaml

# ----------------------------------------------------------------------------
# 路径常量
# ----------------------------------------------------------------------------
_RULES_DIR = os.path.dirname(os.path.abspath(__file__))
_PATTERNS_DIR = os.path.join(_RULES_DIR, "patterns")
_SCHEMA_DIR = os.path.join(_RULES_DIR, "schema")

# 性能阈值
_COMPILE_TIME_THRESHOLD_MS = 100    # 正则编译时间阈值
_MATCH_TIME_THRESHOLD_MS = 1        # 单次匹配时间阈值
_PERF_MATCH_ITERATIONS = 1000       # 性能测试迭代次数

# JSON Schema 可选依赖
try:
    import jsonschema
    _HAS_JSONSCHEMA = True
except ImportError:
    _HAS_JSONSCHEMA = False


class RuleValidator:
    """规则校验器。"""

    def __init__(self):
        self.schema = self._load_schema()
        self.results = {
            "schema": [],       # [(file, passed, message)]
            "golden": [],       # [(pattern_id, passed, message)]
            "performance": [],  # [(pattern_id, passed, compile_ms, match_ms, message)]
        }

    # ------------------------------------------------------------------------
    # Schema 加载
    # ------------------------------------------------------------------------
    def _load_schema(self) -> dict:
        path = os.path.join(_SCHEMA_DIR, "pattern_schema.json")
        with open(path, encoding="utf-8") as f:
            return json.load(f)

    # ------------------------------------------------------------------------
    # 1. Schema 校验
    # ------------------------------------------------------------------------
    def validate_schema(self) -> bool:
        """校验所有 YAML 文件是否符合 JSON Schema。"""
        print("=" * 70)
        print("1. Schema 校验")
        print("=" * 70)
        all_passed = True
        for fname in sorted(os.listdir(_PATTERNS_DIR)):
            if not fname.endswith((".yaml", ".yml")):
                continue
            fpath = os.path.join(_PATTERNS_DIR, fname)
            with open(fpath, encoding="utf-8") as f:
                data = load_rule_yaml(f)
            passed, msg = self._validate_one_schema(data, fname)
            status = "OK  " if passed else "FAIL"
            print("  %s %s: %s" % (status, fname, msg if not passed else ""))
            self.results["schema"].append((fname, passed, msg))
            if not passed:
                all_passed = False
        n_pass = sum(1 for _, p, _ in self.results["schema"] if p)
        n_total = len(self.results["schema"])
        print("  ---")
        print("  Schema: %d/%d passed" % (n_pass, n_total))
        return all_passed

    def _validate_one_schema(self, data: dict, fname: str) -> Tuple[bool, str]:
        """校验单个 YAML 文件。"""
        if _HAS_JSONSCHEMA:
            try:
                jsonschema.validate(data, self.schema)
                return True, ""
            except jsonschema.ValidationError as e:
                return False, str(e.message)
            except jsonschema.SchemaError as e:
                return False, "Schema error: %s" % e
        else:
            return self._validate_basic(data, fname)

    def _validate_basic(self, data: dict, fname: str) -> Tuple[bool, str]:
        """无 jsonschema 时的基础结构校验。"""
        if not isinstance(data, dict):
            return False, "顶层应为 dict"
        for field in ("version", "category"):
            if field not in data:
                return False, "缺少必填字段: %s" % field
        if "patterns" not in data and "keyword_rules" not in data:
            return False, "缺少 patterns 或 keyword_rules"
        if "patterns" in data:
            for p in data["patterns"]:
                if "id" not in p or "pattern" not in p:
                    return False, "pattern 缺少 id 或 pattern 字段"
        if "keyword_rules" in data:
            for r in data["keyword_rules"]:
                if "id" not in r or "label" not in r or "keywords" not in r:
                    return False, "keyword_rule 缺少 id/label/keywords 字段"
        return True, ""

    # ------------------------------------------------------------------------
    # 2. 黄金样例
    # ------------------------------------------------------------------------
    def validate_golden(self) -> bool:
        """运行所有黄金样例。"""
        print()
        print("=" * 70)
        print("2. 黄金样例校验")
        print("=" * 70)
        all_passed = True
        for fname in sorted(os.listdir(_PATTERNS_DIR)):
            if not fname.endswith((".yaml", ".yml")):
                continue
            fpath = os.path.join(_PATTERNS_DIR, fname)
            with open(fpath, encoding="utf-8") as f:
                data = load_rule_yaml(f)
            patterns = data.get("patterns") or []
            for p in patterns:
                pid = p["id"]
                examples = p.get("golden_examples") or []
                if not examples:
                    continue
                passed, msg = self._run_golden_for_pattern(p)
                status = "OK  " if passed else "FAIL"
                print("  %s %s (%d examples): %s" % (
                    status, pid, len(examples), msg if not passed else ""))
                self.results["golden"].append((pid, passed, msg))
                if not passed:
                    all_passed = False
        n_pass = sum(1 for _, p, _ in self.results["golden"] if p)
        n_total = len(self.results["golden"])
        print("  ---")
        print("  Golden: %d/%d patterns passed" % (n_pass, n_total))
        return all_passed

    def _run_golden_for_pattern(self, p: dict) -> Tuple[bool, str]:
        """对单个 pattern 运行黄金样例。"""
        pattern_str = p["pattern"]
        method = p.get("method", "search")
        flags = 0
        for f in p.get("flags", []):
            flags |= getattr(re, f, 0)
        try:
            compiled = re.compile(pattern_str, flags)
        except re.error as e:
            return False, "正则编译失败: %s" % e

        examples = p.get("golden_examples") or []
        for ex in examples:
            text = ex["input"]
            should_match = ex["should_match"]
            if method == "match":
                m = compiled.match(text)
            elif method == "fullmatch":
                m = compiled.fullmatch(text)
            else:
                m = compiled.search(text)
            matched = m is not None
            if matched != should_match:
                return False, "输入 %r 期望 match=%s 实际 match=%s" % (
                    text, should_match, matched)
            # 检查期望的匹配组内容
            if "expected_group" in ex and m:
                actual = m.group(0)
                if actual != ex["expected_group"]:
                    return False, "输入 %r 期望 group(0)=%r 实际=%r" % (
                        text, ex["expected_group"], actual)
        return True, ""

    # ------------------------------------------------------------------------
    # 3. 正则性能
    # ------------------------------------------------------------------------
    def validate_performance(self) -> bool:
        """正则性能测试。"""
        print()
        print("=" * 70)
        print("3. 正则性能测试")
        print("=" * 70)
        all_passed = True
        for fname in sorted(os.listdir(_PATTERNS_DIR)):
            if not fname.endswith((".yaml", ".yml")):
                continue
            fpath = os.path.join(_PATTERNS_DIR, fname)
            with open(fpath, encoding="utf-8") as f:
                data = load_rule_yaml(f)
            patterns = data.get("patterns") or []
            for p in patterns:
                pid = p["id"]
                examples = p.get("golden_examples") or []
                passed, compile_ms, match_ms, msg = self._perf_one_pattern(p, examples)
                status = "OK  " if passed else "WARN"
                detail = "" if passed else " " + msg
                print("  %s %s: compile=%.2fms, match=%.3fms%s" % (
                    status, pid, compile_ms, match_ms, detail))
                self.results["performance"].append(
                    (pid, passed, compile_ms, match_ms, msg))
                if not passed:
                    all_passed = False
        n_pass = sum(1 for _, p, _, _, _ in self.results["performance"] if p)
        n_total = len(self.results["performance"])
        print("  ---")
        print("  Performance: %d/%d passed (thresholds: compile<%dms, match<%dms)" % (
            n_pass, n_total, _COMPILE_TIME_THRESHOLD_MS, _MATCH_TIME_THRESHOLD_MS))
        return all_passed

    def _perf_one_pattern(self, p: dict, examples: list) -> Tuple[bool, float, float, str]:
        """测试单个 pattern 的性能。"""
        pattern_str = p["pattern"]
        flags = 0
        for f in p.get("flags", []):
            flags |= getattr(re, f, 0)

        # 编译时间
        t0 = time.perf_counter()
        try:
            compiled = re.compile(pattern_str, flags)
        except re.error as e:
            return False, 0, 0, "编译失败: %s" % e
        t1 = time.perf_counter()
        compile_ms = (t1 - t0) * 1000

        # 匹配时间（取第一个正例做迭代）
        match_ms = 0
        test_text = ""
        for ex in examples:
            if ex.get("should_match", True):
                test_text = ex["input"]
                break
        if test_text:
            method = p.get("method", "search")
            matcher = getattr(compiled, method)
            t0 = time.perf_counter()
            for _ in range(_PERF_MATCH_ITERATIONS):
                matcher(test_text)
            t1 = time.perf_counter()
            match_ms = (t1 - t0) / _PERF_MATCH_ITERATIONS * 1000

        msg = ""
        passed = True
        if compile_ms > _COMPILE_TIME_THRESHOLD_MS:
            msg += "编译时间 %.2fms > %dms; " % (compile_ms, _COMPILE_TIME_THRESHOLD_MS)
            passed = False
        if match_ms > _MATCH_TIME_THRESHOLD_MS:
            msg += "匹配时间 %.3fms > %dms; " % (match_ms, _MATCH_TIME_THRESHOLD_MS)
            passed = False
        return passed, compile_ms, match_ms, msg.strip("; ")

    # ------------------------------------------------------------------------
    # 汇总
    # ------------------------------------------------------------------------
    def run_all(self) -> bool:
        """运行全部三项校验，返回是否全部通过。"""
        print("BiddingKG 规则校验")
        print("patterns 目录: %s" % _PATTERNS_DIR)
        print("schema: %s" % ("jsonschema" if _HAS_JSONSCHEMA else "basic 结构校验"))
        print()

        s = self.validate_schema()
        g = self.validate_golden()
        p = self.validate_performance()

        print()
        print("=" * 70)
        print("SUMMARY")
        print("=" * 70)
        n_schema = sum(1 for _, p, _ in self.results["schema"] if p)
        n_golden = sum(1 for _, p, _ in self.results["golden"] if p)
        n_perf = sum(1 for _, p, _, _, _ in self.results["performance"] if p)
        print("  Schema:      %d/%d" % (n_schema, len(self.results["schema"])))
        print("  Golden:      %d/%d" % (n_golden, len(self.results["golden"])))
        print("  Performance: %d/%d" % (n_perf, len(self.results["performance"])))
        print("  ---")
        all_passed = s and g and p
        if all_passed:
            print("  ALL CHECKS PASSED")
        else:
            print("  SOME CHECKS FAILED")
        return all_passed


# ----------------------------------------------------------------------------
# 命令行入口
# ----------------------------------------------------------------------------
if __name__ == "__main__":
    validator = RuleValidator()
    passed = validator.run_all()
    exit(0 if passed else 1)