feat: 完成 issue #43 ④ 敏感度路由规则引擎(本地/云端分级)

This commit is contained in:
2026-08-05 02:18:26 +08:00
parent 2947dae434
commit 6bbcd810e7
3 changed files with 114 additions and 2 deletions
+58 -1
View File
@@ -157,6 +157,7 @@ class RouterRule:
pattern: str
target: str
description: str = ""
priority: int = 0 # 数字越小越优先(issue #43);相同按声明顺序
_compiled: Optional["re.Pattern[str]"] = field(default=None, repr=False, compare=False)
@classmethod
@@ -173,6 +174,10 @@ class RouterRule:
raise ValueError(f"router 规则 {name} 的 target 非法:{target!r}")
if kind not in ("keyword", "regex"):
raise ValueError(f"router 规则 {name} 的 kind 非法:{kind!r}")
try:
priority = int(m.get("priority", 0))
except (TypeError, ValueError):
raise ValueError(f"router 规则 {name} 的 priority 非法:{m.get('priority')!r}")
return cls(
name=name,
category=str(m.get("category", "general")),
@@ -180,6 +185,7 @@ class RouterRule:
pattern=pattern,
target=target,
description=str(m.get("description", "")),
priority=priority,
)
def _compiled_regex(self) -> "re.Pattern[str]":
@@ -260,7 +266,9 @@ class SensitivityRouter:
merged: Dict[str, RouterRule] = {r.name: r for r in self.ROUTER_DEFAULT_RULES}
for r in (rules or []):
merged[r.name] = r
self._rules: List[RouterRule] = list(merged.values())
# 按 priority 升序(稳定排序:同 priority 保持合并后的声明顺序)
self._rules: List[RouterRule] = sorted(
merged.values(), key=lambda r: r.priority)
self.default_target = default_target
self.audit = audit
self._audit_log: List[Dict[str, object]] = []
@@ -310,6 +318,55 @@ class SensitivityRouter:
self._record(decision)
return decision
# -- 规则引擎诊断(Issue #43) -----------------------------------------
def validate_rules(self) -> List[str]:
"""校验规则集合法性,返回问题列表(空 = 合法)。
检查项:规则名重复、pattern 空、target/kind 非法(from_mapping 已拦截)、
priority 排序无冲突(仅报告,不阻断)。
"""
problems: List[str] = []
seen: Dict[str, int] = {}
for rule in self._rules:
seen[rule.name] = seen.get(rule.name, 0) + 1
if not rule.pattern:
problems.append(f"规则 {rule.name} pattern 为空")
if rule.target not in (RouteTarget.LOCAL, RouteTarget.CLOUD,
RouteTarget.BLOCK):
problems.append(f"规则 {rule.name} target 非法:{rule.target!r}")
for name, count in seen.items():
if count > 1:
problems.append(f"规则名重复:{name}({count} 次)")
return problems
def stats(self) -> Dict[str, object]:
"""规则集统计(按 target / kind 分类)。"""
by_target: Dict[str, int] = {}
by_kind: Dict[str, int] = {}
for rule in self._rules:
by_target[rule.target] = by_target.get(rule.target, 0) + 1
by_kind[rule.kind] = by_kind.get(rule.kind, 0) + 1
return {"total": len(self._rules),
"by_target": by_target, "by_kind": by_kind}
def describe(self, query: str) -> Dict[str, object]:
"""命中链诊断:返回 query 命中的全部规则(不改变路由决策)。
用于规则调试/配置台预览:查看同一条 query 命中的多条规则,
理解实际决策是第一条命中的规则(按 priority 排序后)。
"""
hits = []
for rule in self._rules:
found = rule.find(query)
if found:
hits.append({
"name": rule.name, "target": rule.target,
"category": rule.category, "kind": rule.kind,
"priority": rule.priority, "matches": len(found),
})
return {"query": query, "hits": hits}
# -- 评估(Issue #49 雏形:路由准确率离线评估脚本入口) ----------------
def evaluate(self, samples: List[Dict[str, object]]) -> Dict[str, object]: