# -*- coding: utf-8 -*- """规则加载器（RULE LOADER）。 按 ARCHITECTURE.md Phase 7 设计，从 ``dl/rules/patterns/*.yaml`` 加载正则规则 和关键词分类规则，编译为 ``re.Pattern`` 对象并缓存，供业务层使用。 使用方式:: from BiddingKG.dl.rules.loader import RuleLoader # 1. 获取编译后的正则对象（与 re.compile() 返回类型一致） pattern = RuleLoader.get_pattern("failure_keywords") if pattern.search(text): ... # 2. 使用 YAML 中指定的匹配方法（search/match/fullmatch） if RuleLoader.match("rank_first", text): return "win_tenderer" # 3. 关键词分类 label = RuleLoader.classify("money_source", "财政拨款") # -> "财政资金,上级拨款" 依赖方向：rules 只依赖 common，不反向依赖业务层。 """ from __future__ import absolute_import import os import re import threading from typing import Dict, List, Optional, Any, Pattern as RegexPattern import yaml # ---------------------------------------------------------------------------- # YAML Loader：on/off/yes/no 不解析为布尔（YAML 1.1 陷阱） # ---------------------------------------------------------------------------- # PyYAML 按 YAML 1.1 把裸标量 on/off/yes/no 解析为 bool，导致 guard 条件的 # `on: behind` 键变成 True、cond["on"] 取不到值而静默回退默认切片。 # 此处仅保留 true/false 的布尔解析，on/off/yes/no 保持字符串。 import re as _re class _RuleYamlLoader(yaml.SafeLoader): """规则 YAML 专用 Loader（on/off/yes/no 保持字符串）。""" _RuleYamlLoader.yaml_implicit_resolvers = { key: [(tag, regexp) for tag, regexp in resolvers if tag != 'tag:yaml.org,2002:bool'] for key, resolvers in yaml.SafeLoader.yaml_implicit_resolvers.items() } _RuleYamlLoader.add_implicit_resolver( 'tag:yaml.org,2002:bool', _re.compile(r'^(?:true|True|TRUE|false|False|FALSE)$'), list('tTfF'), ) def load_rule_yaml(stream): """加载规则 YAML（on/off/yes/no 保持字符串，其余语义同 safe_load）。""" return yaml.load(stream, Loader=_RuleYamlLoader) # ---------------------------------------------------------------------------- # 路径常量 # ---------------------------------------------------------------------------- _RULES_DIR = os.path.dirname(os.path.abspath(__file__)) _PATTERNS_DIR = os.path.join(_RULES_DIR, "patterns") # ---------------------------------------------------------------------------- # 正则标志映射 # ---------------------------------------------------------------------------- _FLAG_MAP = { "IGNORECASE": re.IGNORECASE, "MULTILINE": re.MULTILINE, "DOTALL": re.DOTALL, "UNICODE": re.UNICODE, "VERBOSE": re.VERBOSE, } class RuleLoader: """规则加载器（单例缓存）。 所有方法均为类方法，通过类级缓存避免重复加载。 线程安全：首次加载使用锁保护。 """ # 类级缓存 _patterns: Dict[str, RegexPattern] = {} # id → compiled regex _pattern_meta: Dict[str, dict] = {} # id → {method, flags, description, ...} _rules: List[dict] = [] # 全量规则 dict（Phase C 引擎消费，保持 YAML 顺序） _classifiers: Dict[str, dict] = {} # file_stem → {config, rules} _loaded: bool = False _lock = threading.Lock() # ------------------------------------------------------------------------ # 加载 # ------------------------------------------------------------------------ @classmethod def _ensure_loaded(cls): """惰性加载所有 YAML 规则文件（线程安全）。""" if cls._loaded: return with cls._lock: if cls._loaded: # double-check return cls._load_all() cls._loaded = True @classmethod def _load_all(cls): """扫描 patterns/ 目录，加载所有 .yaml 文件。""" if not os.path.isdir(_PATTERNS_DIR): return for fname in sorted(os.listdir(_PATTERNS_DIR)): if not fname.endswith((".yaml", ".yml")): continue fpath = os.path.join(_PATTERNS_DIR, fname) stem = os.path.splitext(fname)[0] with open(fpath, encoding="utf-8") as f: data = load_rule_yaml(f) if not data: continue cls._load_patterns(data, stem) cls._load_classifier(data, stem) @classmethod def _load_patterns(cls, data: dict, stem: str): """加载正则规则。""" patterns = data.get("patterns") or [] for p in patterns: pid = p["id"] pattern_str = p["pattern"] flags = 0 for f in p.get("flags", []): flags |= _FLAG_MAP.get(f, 0) compiled = re.compile(pattern_str, flags) cls._patterns[pid] = compiled cls._pattern_meta[pid] = { "method": p.get("method", "search"), "description": p.get("description", ""), "category": data.get("category", "OTHER"), "source_file": stem, "pattern_string": pattern_str, } # Phase C：保留完整规则结构（stage/guard/set_prob 等）供引擎消费 rule = dict(p) rule["pattern_compiled"] = compiled rule["category"] = data.get("category", "OTHER") rule["source_file"] = stem cls._rules.append(rule) @classmethod def _load_classifier(cls, data: dict, stem: str): """加载关键词分类规则。""" rules = data.get("keyword_rules") if not rules: return config = data.get("classifier_config", {}) cls._classifiers[stem] = { "config": { "min_matches": config.get("min_matches", 1), "max_matches": config.get("max_matches", 99), "default_label": config.get("default_label", ""), "separator": config.get("separator", ","), "sort": config.get("sort", True), }, "rules": rules, } # ------------------------------------------------------------------------ # 公共 API：正则 # ------------------------------------------------------------------------ @classmethod def get_pattern(cls, pattern_id: str) -> RegexPattern: """获取编译后的正则对象。 返回 ``re.Pattern`` 实例，可直接调用 ``.search()``, ``.match()``, ``.findall()``, ``.finditer()`` 等方法，与 ``re.compile()`` 返回类型一致。 :param pattern_id: 规则 ID（YAML 中的 ``id`` 字段） :return: 编译后的 ``re.Pattern`` 对象 :raises KeyError: 规则 ID 不存在 """ cls._ensure_loaded() if pattern_id not in cls._patterns: raise KeyError("未找到规则 ID: %s（可用: %s）" % ( pattern_id, ", ".join(sorted(cls._patterns.keys())))) return cls._patterns[pattern_id] @classmethod def match(cls, pattern_id: str, text: str) -> Optional[re.Match]: """使用 YAML 中指定的匹配方法进行匹配。 根据 YAML 中的 ``method`` 字段自动选择： - ``search`` → ``pattern.search(text)`` - ``match`` → ``pattern.match(text)`` - ``fullmatch`` → ``pattern.fullmatch(text)`` :param pattern_id: 规则 ID :param text: 待匹配文本 :return: ``re.Match`` 对象或 ``None`` """ cls._ensure_loaded() pattern = cls.get_pattern(pattern_id) method = cls._pattern_meta.get(pattern_id, {}).get("method", "search") if method == "match": return pattern.match(text) elif method == "fullmatch": return pattern.fullmatch(text) else: return pattern.search(text) @classmethod def search(cls, pattern_id: str, text: str) -> Optional[re.Match]: """强制使用 search 方法匹配。""" cls._ensure_loaded() return cls.get_pattern(pattern_id).search(text) @classmethod def get_all_pattern_ids(cls) -> List[str]: """返回所有已加载的规则 ID。""" cls._ensure_loaded() return sorted(cls._patterns.keys()) @classmethod def get_pattern_meta(cls, pattern_id: str) -> dict: """返回规则的元信息（方法、描述、类别等）。""" cls._ensure_loaded() return cls._pattern_meta.get(pattern_id, {}) # ------------------------------------------------------------------------ # 公共 API：完整规则（Phase C RoleRuleEngine 消费） # ------------------------------------------------------------------------ @classmethod def get_rules(cls, category: Optional[str] = None, staged: Optional[bool] = None) -> List[dict]: """返回完整规则列表（保持 YAML 文件内顺序）。 每条规则为 dict，除 YAML 原有字段（id/stage/model_labels/applies_to/ window/guard/guard_not/or_trigger/set_label/set_prob/entity_filter/ golden_examples 等）外，附加： - ``pattern_compiled`` : 编译后的 ``re.Pattern`` - ``category`` : 规则文件类别（ROLE/MONEY/...） - ``source_file`` : 来源 YAML 文件名（去扩展名） :param category: 只返回该类别的规则（None = 全部） :param staged: True = 只返回带 ``stage`` 字段的规则（引擎修正规则）， False = 只返回不带 stage 的规则，None = 全部 """ cls._ensure_loaded() rules = cls._rules if category is not None: rules = [r for r in rules if r.get("category") == category] if staged is not None: rules = [r for r in rules if ("stage" in r) == staged] return list(rules) @classmethod def get_rule(cls, pattern_id: str) -> dict: """按规则 ID 返回完整规则 dict（含 pattern_compiled）。 :raises KeyError: 规则 ID 不存在 """ cls._ensure_loaded() for r in cls._rules: if r["id"] == pattern_id: return r raise KeyError("未找到规则 ID: %s" % pattern_id) # ------------------------------------------------------------------------ # 公共 API：关键词分类 # ------------------------------------------------------------------------ @classmethod def classify(cls, classifier_id: str, text: str) -> str: """关键词分类。 根据 ``keyword_rules`` 规则列表，对文本进行多标签分类。 命中数在 ``min_matches`` 和 ``max_matches`` 之间时返回逗号分隔的标签， 否则返回 ``default_label``。 :param classifier_id: 分类器 ID（YAML 文件名去掉扩展名，如 ``money_source``） :param text: 待分类文本 :return: 分类标签字符串 """ cls._ensure_loaded() if classifier_id not in cls._classifiers: raise KeyError("未找到分类器 ID: %s（可用: %s）" % ( classifier_id, ", ".join(sorted(cls._classifiers.keys())))) clf = cls._classifiers[classifier_id] config = clf["config"] rules = clf["rules"] matched_labels = [] for rule in rules: # 检查排除词 excluded = False for ex in rule.get("excludes", []): if re.search(ex, text): excluded = True break if excluded: continue # 检查正向关键词（任一命中即归类） for kw in rule["keywords"]: if re.search(kw, text): matched_labels.append(rule["label"]) break if config["sort"]: matched_labels.sort() n = len(matched_labels) if n >= config["min_matches"] and n <= config["max_matches"]: return config["separator"].join(matched_labels) else: return config["default_label"] @classmethod def get_classifier_ids(cls) -> List[str]: """返回所有已加载的分类器 ID。""" cls._ensure_loaded() return sorted(cls._classifiers.keys()) # ------------------------------------------------------------------------ # 调试 / 测试辅助 # ------------------------------------------------------------------------ @classmethod def reload(cls): """清除缓存并重新加载（用于测试或热更新）。""" with cls._lock: cls._patterns.clear() cls._pattern_meta.clear() cls._rules[:] = [] cls._classifiers.clear() cls._loaded = False @classmethod def get_stats(cls) -> dict: """返回加载统计信息。""" cls._ensure_loaded() return { "pattern_count": len(cls._patterns), "classifier_count": len(cls._classifiers), "pattern_ids": sorted(cls._patterns.keys()), "classifier_ids": sorted(cls._classifiers.keys()), }