From e50a9424a1d54d76229a19193b3f3320a93c37a7 Mon Sep 17 00:00:00 2001 From: xsl Date: Fri, 14 Aug 2026 20:10:05 +0800 Subject: [PATCH] feat(rules): data-driven decision table engine with hard constraints MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit kb_rule DSL 解析 + 纯域求值（优先级分组/硬约束/用户禁忌压过专业推荐） + 400 条基准 + 黄金集雏形（T-P2-02，TD-10）。 实现：dsl.py 封闭词汇表 pydantic 校验（未识别键拒绝）+ match 纯逻辑 （等值/include/exclude/min-max，缺字段不命中）；engine.py 纯域求值 （priority 降序、confidence 次序、is_hard avoid → hard_blocks、高优先级 avoid 剔除冲突 recommend）；service.py+repo.py published 整表载入进程内 缓存 + CRUD + draft→published→archived 状态机（同 T6 语义）。 测试：DSL 解析与 match 矩阵、排序/硬约束/禁忌优先、状态机矩阵、 黄金集 11 组（含用户禁忌压过专业推荐与多规则叠加）、400 条 × 1000 次 基准（实测均值 0.26ms/次、p95 0.41ms，预算 50ms）；集成 3 用例 （published 载入命中拦截、draft 不载入、DSL 写入拒绝与状态机守卫）。 单测 310 通过（+72），集成 19 通过（+3）。 偏差说明： 1. 冲突仲裁按通用优先级语义实现（高优先级 avoid 剔除"严格更低"优先级的 冲突 recommend），任务文档的"用户禁忌（≥95）永远压过专业推荐（≤90）" 是该规则的特例，黄金集已含该场景用例。 2. include 子句按字段类型取语义：标量字段=命中列表任一值；集合字段= 列表全部在场（子集，精确命中）；文档未区分，此实现两类字段均可用。 3. 单测引擎文件命名为 tests/unit/test_rules_evaluation.py（而非 test_rules_engine.py）：与契约指定的集成文件 tests/integration/ test_rules_engine.py 同名会导致 pytest 收集冲突（tests 目录非包）。 4. 集成测试每用例独立 engine 并 dispose：pytest-asyncio 每用例一个事件 循环，直接用全局 session 工厂会让连接跨 loop 污染其它集成用例。 5. 规则不提供物理删除（命中链需可追溯，下线走 archived）；kb_rule 契约 无 reviewing 状态，状态机为 draft→published→archived（draft 可撤稿）。 --- packages/rules/README.md | 27 +- packages/rules/__init__.py | 8 +- packages/rules/dsl.py | 226 +++++++++++++++++ packages/rules/engine.py | 165 +++++++++++++ packages/rules/repo.py | 43 ++++ packages/rules/service.py | 240 ++++++++++++++++++ tests/integration/test_rules_engine.py | 250 +++++++++++++++++++ tests/unit/test_rules_dsl.py | 260 +++++++++++++++++++ tests/unit/test_rules_evaluation.py | 322 ++++++++++++++++++++++++ tests/unit/test_rules_golden.py | 330 +++++++++++++++++++++++++ tests/unit/test_rules_service.py | 122 +++++++++ 11 files changed, 1984 insertions(+), 9 deletions(-) create mode 100644 packages/rules/dsl.py create mode 100644 packages/rules/engine.py create mode 100644 packages/rules/repo.py create mode 100644 packages/rules/service.py create mode 100644 tests/integration/test_rules_engine.py create mode 100644 tests/unit/test_rules_dsl.py create mode 100644 tests/unit/test_rules_evaluation.py create mode 100644 tests/unit/test_rules_golden.py create mode 100644 tests/unit/test_rules_service.py diff --git a/packages/rules/README.md b/packages/rules/README.md index 5effa89..c1f6914 100644 --- a/packages/rules/README.md +++ b/packages/rules/README.md @@ -1,17 +1,28 @@ -# `packages/rules/` — 规则引擎（占位） +# `packages/rules` — 规则引擎（TD-10 自研数据驱动决策表） -架构 §8.1 / ADR-05：自研数据驱动决策表（TD-10）。 +架构 §8.1 / ADR-05：**结论由规则引擎产出，LLM 只负责表达**（防幻觉红线）。 -## 待实现 +## 模块 -- 从 `kb_rule` 读已发布规则，求值并输出**命中链** -- 硬约束（`is_hard`）校验：可拦截违规输出 -- p95 < 50ms（FR-KB-05） +- `dsl.py` —— kb_rule DSL 的 pydantic 解析校验 + 画像匹配（纯逻辑）。封闭词汇表： + 未识别的键一律拒绝而非忽略，规则不会静默失效。 +- `engine.py` —— 纯域求值：priority 降序（同序按 confidence）、`is_hard` 的 avoid + 命中进 `hard_blocks`（调用方必须据此做硬过滤）、高优先级 avoid 剔除严格低优先级 + 的冲突 recommend（用户禁忌 95 压过专业推荐 ≤90）。 +- `repo.py` —— kb_rule 表的唯一读写点（经共享 ORM 访问）。 +- `service.py` —— CRUD、draft→published→archived 状态机（同 T6 语义）、published + 规则进程内缓存（量级 ~400，整表载入，管理动作后 `invalidate()`）。 ## 设计约束 - **规则是数据不是代码**。求值器是代码，规则本身在 `kb_rule` 表里， 顾问在后台维护，改规则不发版（NFR-MAINT-05）。 -- 每次求值必须输出**命中了哪几条规则**，这是报告与 Agent 可解释性的来源， +- 每次求值必须输出**命中链**（`rule_hits`：rule_key/priority/category/is_hard/ + confidence/explanation），这是报告与 Agent 可解释性的来源， 也是 `pf_diagnosis.rule_hits` 与 `ag_message.rule_hits` 的数据来源。 -- 已发布规则集缓存在进程内 + Redis，TTL 5 分钟，发布时广播失效（架构 §10）。 +- p95 < 50ms（FR-KB-05）：求值是纯内存操作，400 条规则量级实测远低于预算 + （见 `tests/unit/test_rules_engine.py` 的基准用例）。 +- 黄金集雏形：`tests/unit/test_rules_golden.py`（固定画像 → 期望输出，含硬禁忌 + 与多规则叠加场景），改引擎行为必须先过黄金集。 +- 已发布规则集缓存在进程内；Redis TTL + 发布广播失效（架构 §10）为后续接入项， + 失效入口已收敛到 `service.invalidate_published_cache()`。 diff --git a/packages/rules/__init__.py b/packages/rules/__init__.py index 7188402..0820ad8 100644 --- a/packages/rules/__init__.py +++ b/packages/rules/__init__.py @@ -1 +1,7 @@ -"""规则引擎：决策表求值、硬约束校验、命中链输出。占位。""" +"""规则引擎（TD-10 自研数据驱动决策表）：DSL 校验、纯域求值、硬约束拦截、命中链输出。 + +- :mod:`packages.rules.dsl` —— kb_rule DSL 的 pydantic 解析与画像匹配（纯逻辑） +- :mod:`packages.rules.engine` —— 优先级排序 / 硬约束 / 用户禁忌压过专业推荐 +- :mod:`packages.rules.repo` —— kb_rule 表数据访问（唯一读写点） +- :mod:`packages.rules.service` —— CRUD、draft→published 状态机、已发布规则缓存 +""" diff --git a/packages/rules/dsl.py b/packages/rules/dsl.py new file mode 100644 index 0000000..9647fd9 --- /dev/null +++ b/packages/rules/dsl.py @@ -0,0 +1,226 @@ +"""kb_rule DSL 的解析校验与纯匹配（TD-10 决策表，知识库文档 §3.2⑩）。 + +规则是数据不是代码（NFR-MAINT-05）：本模块只回答两个问题——"一条规则是否 +合法"与"一条规则是否命中画像"，不碰任何 I/O。 + +**封闭词汇表**：未识别的键一律拒绝而非忽略。规则里写错一个键名却被静默忽略， +意味着它悄悄不生效——给用户的建议变味且无人察觉，比写入时报错危险得多。 +这是"专业可信"红线在数据侧的落地（FR-KB-02）。 + +条件子句语义（子句间全部 AND）： + +- 等值（str）：画像字段等于该值；画像字段是集合时退化为"包含该值"。 +- ``include``：标量字段=命中列表任一值；集合字段=列表全部在场（子集语义）。 +- ``exclude``：画像取值与列表没有任何交集。 +- ``min``/``max``：数值闭区间，仅对数值字段（temperature）有意义。 +- 画像缺字段（None）：该子句不匹配 → 整条规则不命中。 +""" + +from __future__ import annotations + +from dataclasses import dataclass, fields +from typing import Any, Self + +from pydantic import BaseModel, ConfigDict, Field, model_validator + +# ── 优先级分组（知识库文档 §3.2⑩：数字越大越优先）────────────────────────── +PRIORITY_COMPLIANCE = 100 # 合规硬约束（发布红线） +PRIORITY_USER_TABOO = 95 # 用户禁忌：永远压过专业推荐 +PRIORITY_BODY_SHAPE = 90 # 体型 +PRIORITY_COLOR = 85 # 色彩 +PRIORITY_OCCASION = 80 # 场景 +PRIORITY_TEMPERATURE = 75 # 温度 +PRIORITY_STYLE = 70 # 风格 +PRIORITY_PREFERENCE = 60 # 偏好 +PRIORITY_TREND = 50 # 趋势 + + +@dataclass(frozen=True, slots=True) +class RuleContext: + """求值输入：一份用户画像。字段缺失（None/空集合）= 相关子句不匹配。""" + + body_shape: str | None = None + season: str | None = None + face_shape: str | None = None + temperature: float | None = None + features: tuple[str, ...] = () + goals: tuple[str, ...] = () + occasion: str | None = None + style: str | None = None + user_taboos: tuple[str, ...] = () + + +#: 条件字段只能引用 RuleContext 上真实存在的字段——typo 在解析时就暴露， +#: 而不是等到线上作为"永不命中的死规则"被发现。 +CONTEXT_FIELDS: frozenset[str] = frozenset(f.name for f in fields(RuleContext)) + + +class ConditionClause(BaseModel): + """集合/区间子句：min/max=闭区间；include/exclude 按字段类型取语义。 + + - 标量字段（body_shape/season/…）：``include`` = 字段值命中列表任一值； + - 集合字段（features/goals/user_taboos）：``include`` = 列表全部在场 + （子集语义——决策表要精确命中，宁可少命中也不误命中，想要"任一满足" + 就拆成多条规则）； + - ``exclude`` 两种类型对称：与列表有任何交集即不匹配。 + """ + + model_config = ConfigDict(extra="forbid") + + include: tuple[str, ...] | None = None + exclude: tuple[str, ...] | None = None + min: float | None = None + max: float | None = None + + @model_validator(mode="after") + def _must_constrain_something(self) -> Self: + if self.include is None and self.exclude is None and self.min is None and self.max is None: + raise ValueError("条件子句至少要包含 include/exclude/min/max 之一") + if self.min is not None and self.max is not None and self.min > self.max: + raise ValueError("min 不能大于 max") + return self + + +#: 子句取值：等值匹配（str）或集合/区间结构（ConditionClause）。 +ClauseValue = str | ConditionClause + + +class RuleCondition(BaseModel): + """条件子句集合（子句间 AND）。 + + 对外序列化形状是决策表里的扁平 dict（如 ``{"body_shape": "A型"}``）， + 内部挂在 ``clauses`` 下是为了复用 pydantic 的键值校验。 + + 空 conditions 合法：表示无条件命中，供合规类兜底规则使用（例如 + "全场禁真兽皮"）；effects 仍必须至少写一项，规则不会变成空操作。 + """ + + model_config = ConfigDict(extra="forbid") + + clauses: dict[str, ClauseValue] = Field(default_factory=dict) + + @model_validator(mode="before") + @classmethod + def _wrap_flat_dict(cls, data: Any) -> Any: + # 允许直接传扁平 dict（DB JSONB / 后台表单的自然形状）。 + if isinstance(data, dict) and "clauses" not in data: + return {"clauses": data} + return data + + @model_validator(mode="after") + def _fields_must_exist_on_context(self) -> Self: + unknown = sorted(set(self.clauses) - CONTEXT_FIELDS) + if unknown: + raise ValueError(f"未识别的条件字段 {unknown}，可用字段：{sorted(CONTEXT_FIELDS)}") + return self + + +class SlotAttrs(BaseModel): + """单个品类的属性约束：slot + 每个属性维度上的取值集合。 + + ``attrs`` 为空表示对整个 slot 推荐/回避（如"回避泳装品类"）。维度名开放 + （shoulder/sleeve/silhouette/…由衣橱属性体系决定，不能收口），但每个维度 + 的取值列表必须非空——空列表约束不了任何东西，多半是漏填。 + """ + + model_config = ConfigDict(extra="forbid") + + slot: str = Field(min_length=1, max_length=32) + attrs: dict[str, tuple[str, ...]] = Field(default_factory=dict) + + @model_validator(mode="after") + def _values_must_not_be_empty(self) -> Self: + for dim, values in self.attrs.items(): + if not dim: + raise ValueError("属性维度名不能为空") + if not values: + raise ValueError(f"属性 {dim} 的取值列表不能为空") + return self + + +class RuleEffect(BaseModel): + """规则效果：推荐什么、回避什么、怎么向用户解释。""" + + model_config = ConfigDict(extra="forbid") + + recommend: tuple[SlotAttrs, ...] = () + avoid: tuple[SlotAttrs, ...] = () + explanation: str | None = Field(default=None, max_length=500) + + @model_validator(mode="after") + def _at_least_one_effect(self) -> Self: + if not (self.recommend or self.avoid or self.explanation): + raise ValueError("effects 至少要包含 recommend/avoid/explanation 之一") + return self + + +class Rule(BaseModel): + """一条已通过 DSL 校验的决策表规则（求值引擎的唯一输入类型）。""" + + model_config = ConfigDict(extra="forbid") + + rule_key: str = Field(min_length=1, max_length=64) + category: str = Field(min_length=1, max_length=32) + priority: int = Field(ge=0, le=100) + conditions: RuleCondition = Field(default_factory=RuleCondition) + effects: RuleEffect + is_hard: bool = False + confidence: float = Field(default=0.9, ge=0.0, le=1.0) + + +# ── 匹配（纯逻辑，重点单测对象）──────────────────────────────────────────── + + +def match(condition: RuleCondition, ctx: RuleContext) -> bool: + """所有子句 AND；画像缺字段（None）= 该子句不匹配 → 整条规则不命中。""" + for name, clause in condition.clauses.items(): + if not _clause_matches(clause, getattr(ctx, name, None)): + return False + return True + + +def _as_options(value: Any) -> tuple[str, ...] | None: + """把画像字段值规整成可做集合判断的取值元组；不适用的类型返回 None。""" + if isinstance(value, str): + return (value,) + if isinstance(value, (tuple, list, frozenset, set)): + return tuple(value) + return None + + +def _is_number(value: Any) -> bool: + # bool 是 int 的子类，但 True 不是任何字段的合法温度值。 + return isinstance(value, (int, float)) and not isinstance(value, bool) + + +def _clause_matches(clause: ClauseValue, value: Any) -> bool: + if value is None: + return False # 缺字段 = 不匹配该子句 + if isinstance(clause, str): + options = _as_options(value) + return options is not None and clause in options + if clause.include is not None or clause.exclude is not None: + if isinstance(value, str): + # 标量字段只能取一个值：include=命中列表任一值。 + if clause.include is not None and value not in clause.include: + return False + if clause.exclude is not None and value in clause.exclude: + return False + else: + options = _as_options(value) + if options is None: + return False + present = set(options) + # 集合字段：include=列表全部在场（精确命中）。 + if clause.include is not None and not set(clause.include) <= present: + return False + if clause.exclude is not None and set(clause.exclude) & present: + return False + if clause.min is not None or clause.max is not None: + if not _is_number(value): + return False + if clause.min is not None and value < clause.min: + return False + if clause.max is not None and value > clause.max: + return False + return True diff --git a/packages/rules/engine.py b/packages/rules/engine.py new file mode 100644 index 0000000..b5a8ce6 --- /dev/null +++ b/packages/rules/engine.py @@ -0,0 +1,165 @@ +"""纯域求值引擎（架构 §8.2 rules-engine，TD-10）：无 I/O。 + +防幻觉红线：**结论由规则引擎产出，LLM 只负责表达**。因此本模块是"钱与 +正确性"代码——输出必须完全确定、可解释、可回放。 + +求值流程： + +1. ``dsl.match`` 过滤命中规则（子句 AND，缺字段不命中）； +2. 按 priority 降序、confidence 降序排序（再按 rule_key 兜底，输出完全确定）； +3. 汇总 recommend/avoid（保序去重）与解释； +4. 硬约束：``is_hard`` 规则的 avoid 命中 → ``hard_blocks`` 非空。调用方必须 + 用其中的 avoid 约束对候选做硬过滤——禁忌违反 = 0 是发布红线； +5. 冲突仲裁：recommend 与 avoid 同 slot 且共享维度上取值相交视为冲突， + **高优先级 avoid 剔除严格低优先级 recommend**。由此：用户禁忌（95）压过 + 体型（90）及以下的专业推荐，合规（100）压过一切——"数字越大越优先"在 + 规则之间同样成立，而不是只写死 95 这一个档位。 +""" + +from __future__ import annotations + +from collections.abc import Sequence +from dataclasses import dataclass + +from packages.rules import dsl +from packages.rules.dsl import SlotAttrs + +#: recommend/avoid 条目的去重键：(slot, 排序后的 (维度, 排序后取值) 序列)。 +ItemKey = tuple[str, tuple[tuple[str, tuple[str, ...]], ...]] + + +@dataclass(frozen=True, slots=True) +class RuleHit: + """命中链的一环：报告与 Agent 可解释性的数据来源（架构 §8.1）。""" + + rule_key: str + priority: int + category: str + is_hard: bool + confidence: float + explanation: str | None + + +@dataclass(frozen=True, slots=True) +class HardBlock: + """一条必须执行的硬约束：命中规则元数据 + 待过滤的 avoid 约束。 + + 调用方拿 ``avoid`` 里的 slot/attrs 对候选单品做过滤（同 slot 且任一 + 维度取值落在列表内即剔除），而不是只拿 rule_key 记日志。 + """ + + rule_key: str + priority: int + category: str + confidence: float + explanation: str | None + avoid: tuple[SlotAttrs, ...] + + +@dataclass(frozen=True, slots=True) +class RuleVerdict: + """一次求值的完整结论。""" + + recommend: tuple[SlotAttrs, ...] + avoid: tuple[SlotAttrs, ...] + explanations: tuple[str, ...] + rule_hits: tuple[RuleHit, ...] + hard_blocks: tuple[HardBlock, ...] + + @property + def explanation(self) -> str: + """命中规则的解释按优先级序拼接，供 LLM 直接表达。""" + return "；".join(self.explanations) + + @property + def blocked(self) -> bool: + """是否触发硬拦截。""" + return bool(self.hard_blocks) + + +def _item_key(item: SlotAttrs) -> ItemKey: + return ( + item.slot, + tuple(sorted((dim, tuple(sorted(vals))) for dim, vals in item.attrs.items())), + ) + + +def _conflicts(rec: SlotAttrs, avoid: SlotAttrs) -> bool: + """同 slot 且至少一个共享维度上取值相交，才视为推荐与回避冲突。 + + 不同维度的 avoid（如"回避修身"之于"推荐垫肩"）是互补信号而非冲突： + 只有当推荐恰好落进回避圈定的事实集合时才需要仲裁。 + """ + if rec.slot != avoid.slot: + return False + return any( + dim in avoid.attrs and set(vals) & set(avoid.attrs[dim]) for dim, vals in rec.attrs.items() + ) + + +def evaluate(rules: Sequence[dsl.Rule], ctx: dsl.RuleContext) -> RuleVerdict: + """对一份画像求值全部规则，产出推荐/回避/命中链/硬拦截。""" + fired = [r for r in rules if dsl.match(r.conditions, ctx)] + fired.sort(key=lambda r: (-r.priority, -r.confidence, r.rule_key)) + + rule_hits = tuple( + RuleHit( + rule_key=r.rule_key, + priority=r.priority, + category=r.category, + is_hard=r.is_hard, + confidence=r.confidence, + explanation=r.effects.explanation, + ) + for r in fired + ) + hard_blocks = tuple( + HardBlock( + rule_key=r.rule_key, + priority=r.priority, + category=r.category, + confidence=r.confidence, + explanation=r.effects.explanation, + avoid=r.effects.avoid, + ) + for r in fired + # 硬约束的拦截语义只由 avoid 表达：is_hard 规则没写 avoid 就无物可拦。 + if r.is_hard and r.effects.avoid + ) + explanations = tuple(r.effects.explanation for r in fired if r.effects.explanation) + + # recommend 保序去重，并记住来源优先级供仲裁。 + recommend: list[tuple[int, SlotAttrs]] = [] + seen_rec: set[ItemKey] = set() + # avoid 保序去重；同一约束来自多条规则时取最高来源优先级参与仲裁。 + avoid: list[SlotAttrs] = [] + avoid_priority: dict[ItemKey, int] = {} + for r in fired: + for item in r.effects.recommend: + key = _item_key(item) + if key not in seen_rec: + seen_rec.add(key) + recommend.append((r.priority, item)) + for item in r.effects.avoid: + key = _item_key(item) + if key in avoid_priority: + avoid_priority[key] = max(avoid_priority[key], r.priority) + else: + avoid_priority[key] = r.priority + avoid.append(item) + + kept_recommend = tuple( + item + for prio, item in recommend + if not any( + avoid_priority[_item_key(blocker)] > prio and _conflicts(item, blocker) + for blocker in avoid + ) + ) + return RuleVerdict( + recommend=kept_recommend, + avoid=tuple(avoid), + explanations=explanations, + rule_hits=rule_hits, + hard_blocks=hard_blocks, + ) diff --git a/packages/rules/repo.py b/packages/rules/repo.py new file mode 100644 index 0000000..934a9d9 --- /dev/null +++ b/packages/rules/repo.py @@ -0,0 +1,43 @@ +"""kb_rule 表数据访问。**只有本模块读写 kb_rule**（表与 packages/kb 共享， +但规则包只经 ORM 访问，不碰 T6 的文件）。""" + +from __future__ import annotations + +from sqlalchemy import select +from sqlalchemy.ext.asyncio import AsyncSession + +from packages.shared.models import KbRule + +STATUS_PUBLISHED = "published" + + +async def list_published(session: AsyncSession) -> list[KbRule]: + """全量已发布规则。量级 ~400，直接整表载入；序即求值序（priority 降序）。""" + result = await session.execute( + select(KbRule) + .where(KbRule.status == STATUS_PUBLISHED) + .order_by(KbRule.priority.desc(), KbRule.rule_key) + ) + return list(result.scalars().all()) + + +async def get_by_key(session: AsyncSession, rule_key: str) -> KbRule | None: + result = await session.execute(select(KbRule).where(KbRule.rule_key == rule_key)) + return result.scalar_one_or_none() + + +async def list_by_status(session: AsyncSession, status: str) -> list[KbRule]: + result = await session.execute( + select(KbRule).where(KbRule.status == status).order_by(KbRule.rule_key) + ) + return list(result.scalars().all()) + + +async def list_all(session: AsyncSession) -> list[KbRule]: + result = await session.execute(select(KbRule).order_by(KbRule.rule_key)) + return list(result.scalars().all()) + + +def add_rule(session: AsyncSession, rule: KbRule) -> KbRule: + session.add(rule) + return rule diff --git a/packages/rules/service.py b/packages/rules/service.py new file mode 100644 index 0000000..4d46ef0 --- /dev/null +++ b/packages/rules/service.py @@ -0,0 +1,240 @@ +"""kb_rule 的 CRUD、draft→published 状态机与已发布规则缓存。 + +状态机（同 T6 语义）：draft → published → archived；archived 终态；draft 可 +直接 archived（撤稿）。只有 draft 可修改内容且每次修改 version+1——直接改 +published 规则等于线上建议瞬间变味，必须重新走发布动作。 + +规则不做物理删除：命中链（rule_hits）是报告与 Agent 可解释性的数据来源， +历史规则要可追溯，下线一律走 archived。 +""" + +from __future__ import annotations + +from datetime import UTC, datetime +from decimal import Decimal +from typing import Any + +from pydantic import ValidationError +from sqlalchemy.ext.asyncio import AsyncSession + +from packages.domain.ids import next_id +from packages.rules import dsl, engine, repo +from packages.shared.errors import AppError, ErrorCode +from packages.shared.logging import get_logger +from packages.shared.models import KbRule + +logger = get_logger(__name__) + +STATUS_DRAFT = "draft" +STATUS_PUBLISHED = "published" +STATUS_ARCHIVED = "archived" + +#: 合法状态跃迁全集（自跃迁也非法：状态机不接受无操作变更）。 +LEGAL_TRANSITIONS: frozenset[tuple[str, str]] = frozenset( + { + (STATUS_DRAFT, STATUS_PUBLISHED), + (STATUS_PUBLISHED, STATUS_ARCHIVED), + (STATUS_DRAFT, STATUS_ARCHIVED), + } +) + + +def ensure_transition(current: str, target: str) -> None: + """非法跃迁抛 STATE_CONFLICT——状态错误必须响亮失败，不能悄悄吞掉。""" + if (current, target) not in LEGAL_TRANSITIONS: + raise AppError( + ErrorCode.STATE_CONFLICT, + message=f"规则状态不能从 {current} 变更为 {target}", + data={"current": current, "target": target}, + ) + + +def validate_rule_payload(payload: dict[str, Any]) -> dsl.Rule: + """DSL 写入前校验：非法结构 100% 拒绝（未识别键不静默忽略，见 dsl 模块）。""" + try: + return dsl.Rule.model_validate(payload) + except ValidationError as exc: + raise AppError( + ErrorCode.VALIDATION_FAILED, + message="规则 DSL 不合法", + data={ + "errors": [ + f"{'.'.join(str(loc) for loc in err['loc'])}: {err['msg']}" + for err in exc.errors() + ] + }, + ) from exc + + +def row_to_rule(row: KbRule) -> dsl.Rule: + """ORM 行 → 域规则。 + + 已入库数据不合法属于知识库数据错误（5xxxx），必须立刻暴露——静默跳过 + 一条 published 规则等于线上建议悄悄缺了一块，没人会发现。 + """ + try: + return dsl.Rule.model_validate( + { + "rule_key": row.rule_key, + "category": row.category, + "priority": row.priority, + "conditions": row.conditions, + "effects": row.effects, + "is_hard": row.is_hard, + "confidence": float(row.confidence), + } + ) + except ValidationError as exc: + logger.error("kb_rule_dsl_invalid", rule_key=row.rule_key, errors=exc.errors()) + raise AppError( + ErrorCode.CONFIG_ERROR, + message="知识库中存在不合法的规则数据", + data={"rule_key": row.rule_key}, + ) from exc + + +class PublishedRules: + """进程内已发布规则缓存。 + + 量级 ~400、读极多写极少：整表载入 + 管理动作后失效即可满足 p95 < 50ms + （求值本身是纯内存操作）。跨进程失效广播（Redis，架构 §10）属后续接入 + 项，先留 ``invalidate()`` 单一入口。并发首载最多重复读一次库，无害。 + """ + + def __init__(self) -> None: + self._rules: tuple[dsl.Rule, ...] | None = None + + async def load(self, session: AsyncSession) -> tuple[dsl.Rule, ...]: + if self._rules is None: + rows = await repo.list_published(session) + self._rules = tuple(row_to_rule(row) for row in rows) + logger.info("rules_cache_loaded", count=len(self._rules)) + return self._rules + + def invalidate(self) -> None: + self._rules = None + + +_published_rules = PublishedRules() + + +def invalidate_published_cache() -> None: + """发布/下线/修改规则后调用，进程内缓存立刻重建。""" + _published_rules.invalidate() + + +class RuleService: + """kb_rule 的读写门面：CRUD + 状态机 + 缓存求值入口。""" + + def __init__(self, session: AsyncSession) -> None: + self._session = session + + # -- 读 ------------------------------------------------------------------- + + async def get(self, rule_key: str) -> KbRule: + return await self._require(rule_key) + + async def list(self, status: str | None = None) -> list[KbRule]: + if status is None: + return await repo.list_all(self._session) + return await repo.list_by_status(self._session, status) + + async def evaluate(self, ctx: dsl.RuleContext) -> engine.RuleVerdict: + """用进程内已发布规则缓存求值（命中链 + 硬拦截 + 冲突仲裁）。""" + rules = await _published_rules.load(self._session) + return engine.evaluate(rules, ctx) + + # -- 写 ------------------------------------------------------------------- + + async def create( + self, + payload: dict[str, Any], + *, + author: str | None = None, + evidence: str | None = None, + ) -> KbRule: + """新建 draft 规则。DSL 先校验再落库；rule_key 全局唯一。""" + rule = validate_rule_payload(payload) + if await repo.get_by_key(self._session, rule.rule_key) is not None: + raise AppError( + ErrorCode.BAD_REQUEST, + message="规则键已存在", + data={"rule_key": rule.rule_key}, + ) + row = repo.add_rule( + self._session, + KbRule( + rule_id=next_id(), + rule_key=rule.rule_key, + category=rule.category, + version=1, + status=STATUS_DRAFT, + priority=rule.priority, + conditions=rule.conditions.model_dump()["clauses"], + effects=rule.effects.model_dump(), + is_hard=rule.is_hard, + confidence=Decimal(str(rule.confidence)), + author=author, + evidence=evidence, + ), + ) + await self._session.flush() + logger.info("rule_created", rule_key=rule.rule_key, author=author) + return row + + async def update(self, rule_key: str, payload: dict[str, Any]) -> KbRule: + """修改规则内容（仅 draft）：version+1；rule_key 是命中链锚点，不可改。""" + rule = validate_rule_payload(payload) + if rule.rule_key != rule_key: + raise AppError( + ErrorCode.BAD_REQUEST, + message="rule_key 不可修改", + data={"rule_key": rule_key}, + ) + row = await self._require(rule_key) + if row.status != STATUS_DRAFT: + raise AppError( + ErrorCode.STATE_CONFLICT, + message="仅 draft 规则可修改，published 规则请先下线重发", + data={"rule_key": rule_key, "status": row.status}, + ) + row.category = rule.category + row.priority = rule.priority + row.conditions = rule.conditions.model_dump()["clauses"] + row.effects = rule.effects.model_dump() + row.is_hard = rule.is_hard + row.confidence = Decimal(str(rule.confidence)) + row.version += 1 + await self._session.flush() + logger.info("rule_updated", rule_key=rule_key, version=row.version) + return row + + async def publish(self, rule_key: str) -> KbRule: + row = await self._require(rule_key) + ensure_transition(row.status, STATUS_PUBLISHED) + row.status = STATUS_PUBLISHED + row.published_at = datetime.now(UTC) + await self._session.flush() + invalidate_published_cache() + logger.info("rule_published", rule_key=rule_key, version=row.version) + return row + + async def archive(self, rule_key: str) -> KbRule: + """下线/撤稿（draft 与 published 均可）。规则不物理删除，保留可追溯性。""" + row = await self._require(rule_key) + ensure_transition(row.status, STATUS_ARCHIVED) + row.status = STATUS_ARCHIVED + await self._session.flush() + invalidate_published_cache() + logger.info("rule_archived", rule_key=rule_key) + return row + + async def _require(self, rule_key: str) -> KbRule: + row = await repo.get_by_key(self._session, rule_key) + if row is None: + raise AppError( + ErrorCode.NOT_FOUND, + message="规则不存在", + data={"rule_key": rule_key}, + ) + return row diff --git a/tests/integration/test_rules_engine.py b/tests/integration/test_rules_engine.py new file mode 100644 index 0000000..63ecb47 --- /dev/null +++ b/tests/integration/test_rules_engine.py @@ -0,0 +1,250 @@ +"""规则引擎集成用例（真实 PostgreSQL）：published 载入、draft 不载入、硬约束拦截。 + +覆盖任务契约：插 3 条规则（含 1 条 is_hard）→ service 载入 → 求值命中与拦截 +正确；draft 规则不被载入。规则键带随机后缀 + 前缀清理，可与任意环境的存量 +数据共存。 +""" + +from __future__ import annotations + +import os +import uuid +from collections.abc import AsyncIterator +from contextlib import asynccontextmanager +from typing import Any + +import pytest + + +def _infra_available() -> bool: + return bool(os.getenv("MEIDA_RUN_INTEGRATION")) + + +pytestmark = [ + pytest.mark.integration, + pytest.mark.skipif( + not _infra_available(), + reason="set MEIDA_RUN_INTEGRATION=1 with PostgreSQL and Redis running", + ), +] + +_PREFIX = "R-IT7-" + + +def _key(suffix: str) -> str: + return f"{_PREFIX}{suffix}-{uuid.uuid4().hex[:6]}" + + +@asynccontextmanager +async def _session() -> AsyncIterator[Any]: + """每个用例独立 engine：pytest-asyncio 每个用例一个事件循环， + + 共享全局连接池会把上一个 loop 的连接带进下一个用例（Future attached to + a different loop）。用完 dispose，进程内全局池保持零污染。 + """ + from sqlalchemy.ext.asyncio import async_sessionmaker + + from packages.shared.config import get_settings + from packages.shared.db import create_engine + + engine = create_engine(get_settings()) + factory = async_sessionmaker(bind=engine, expire_on_commit=False, autoflush=False) + try: + async with factory() as session: + try: + yield session + await session.commit() + except Exception: + await session.rollback() + raise + finally: + await engine.dispose() + + +async def _cleanup() -> None: + from sqlalchemy import delete + + from packages.rules import service as rules_service + from packages.shared.models import KbRule + + rules_service.invalidate_published_cache() + async with _session() as session: + await session.execute(delete(KbRule).where(KbRule.rule_key.like(f"{_PREFIX}%"))) + + +def _body_payload(rule_key: str) -> dict[str, Any]: + return { + "rule_key": rule_key, + "category": "body_shape", + "priority": 90, + "confidence": 0.95, + "conditions": {"body_shape": "A型", "features": {"include": ["溜肩"]}}, + "effects": { + "recommend": [{"slot": "top", "attrs": {"shoulder": ["IT7垫肩"]}}], + "avoid": [{"slot": "top", "attrs": {"shoulder": ["IT7落肩"]}}], + "explanation": "IT7 体型规则", + }, + } + + +def _hard_payload(rule_key: str) -> dict[str, Any]: + return { + "rule_key": rule_key, + "category": "compliance", + "priority": 100, + "confidence": 1.0, + "is_hard": True, + "conditions": {"occasion": {"include": ["IT7晚宴"]}}, + "effects": { + "avoid": [{"slot": "accessory", "attrs": {"material": ["IT7真兽皮"]}}], + "explanation": "IT7 硬约束", + }, + } + + +def _draft_payload(rule_key: str) -> dict[str, Any]: + return { + "rule_key": rule_key, + "category": "body_shape", + "priority": 90, + "conditions": {"body_shape": "A型"}, + "effects": { + "recommend": [{"slot": "dress", "attrs": {"silhouette": ["IT7草稿独有"]}}], + "explanation": "IT7 未发布规则", + }, + } + + +async def test_published_loaded_draft_not_loaded_and_hard_blocks() -> None: + from packages.rules import dsl + from packages.rules import service as rules_service + + await _cleanup() + body_key, hard_key, draft_key = _key("BODY"), _key("HARD"), _key("DRAFT") + try: + async with _session() as session: + svc = rules_service.RuleService(session) + await svc.create(_body_payload(body_key), author="it7") + await svc.create(_hard_payload(hard_key), author="it7") + await svc.create(_draft_payload(draft_key), author="it7") # 保持 draft + await svc.publish(body_key) + await svc.publish(hard_key) + + ctx = dsl.RuleContext(body_shape="A型", features=("溜肩",), occasion="IT7晚宴") + + # 求值：published 载入（两条命中），draft 不载入 + async with _session() as session: + svc = rules_service.RuleService(session) + verdict = await svc.evaluate(ctx) + hit_keys = [h.rule_key for h in verdict.rule_hits] + assert body_key in hit_keys + assert hard_key in hit_keys + assert draft_key not in hit_keys + + # 硬约束：is_hard 的 avoid 命中 → hard_blocks 带出可过滤的约束内容 + assert verdict.blocked is True + block_keys = [b.rule_key for b in verdict.hard_blocks] + assert hard_key in block_keys + hard_block = next(b for b in verdict.hard_blocks if b.rule_key == hard_key) + assert hard_block.avoid[0].slot == "accessory" + assert hard_block.avoid[0].attrs["material"] == ("IT7真兽皮",) + + # 命中的推荐来自 body 规则；draft 的效果从未出现 + rec_values = { + value + for item in verdict.recommend + for values in item.attrs.values() + for value in values + } + assert "IT7垫肩" in rec_values + assert "IT7草稿独有" not in rec_values + + # 下线 body 规则 → 缓存失效 → 再求值不再命中（保留 hard 规则互不干扰） + async with _session() as session: + svc = rules_service.RuleService(session) + await svc.archive(body_key) + async with _session() as session: + svc = rules_service.RuleService(session) + verdict_after = await svc.evaluate(ctx) + after_keys = [h.rule_key for h in verdict_after.rule_hits] + assert body_key not in after_keys + assert hard_key in after_keys + assert draft_key not in after_keys + finally: + await _cleanup() + + +async def test_create_rejects_invalid_dsl() -> None: + + from packages.shared.errors import AppError, ErrorCode + + await _cleanup() + try: + async with _session() as session: + from packages.rules.service import RuleService + + svc = RuleService(session) + # 未识别的键（conclusions 是架构旧列名）必须被拒绝而不是静默忽略 + with pytest.raises(AppError) as excinfo: + await svc.create({**_body_payload(_key("BAD")), "conclusions": ["垫肩"]}) + assert excinfo.value.code == ErrorCode.VALIDATION_FAILED + finally: + await _cleanup() + + +async def test_state_machine_guards_on_real_rows() -> None: + + from packages.shared.errors import AppError, ErrorCode + + await _cleanup() + sm_key = _key("SM") + other_key = _key("SM2") + try: + async with _session() as session: + from packages.rules.service import RuleService + + svc = RuleService(session) + row = await svc.create(_body_payload(sm_key), author="it7") + assert row.status == "draft" + assert row.version == 1 + + published = await svc.publish(sm_key) + assert published.status == "published" + assert published.published_at is not None + + # published 不能再发布、不能直接改内容（要改必须先下线） + with pytest.raises(AppError) as republish: + await svc.publish(sm_key) + assert republish.value.code == ErrorCode.STATE_CONFLICT + with pytest.raises(AppError) as edit: + await svc.update(sm_key, _body_payload(sm_key)) + assert edit.value.code == ErrorCode.STATE_CONFLICT + + # draft 可以修改且 version+1；rule_key 不可改 + draft_row = await svc.create(_body_payload(other_key), author="it7") + # identity map：update 返回的是同一 ORM 对象，先快照修改前的版本号 + base_version = draft_row.version + updated = await svc.update( + other_key, + { + **_body_payload(other_key), + "priority": 85, + "effects": { + "recommend": [{"slot": "top", "attrs": {"shoulder": ["IT7改"]}}], + "explanation": "IT7 修改后", + }, + }, + ) + assert updated.version == base_version + 1 + assert updated.priority == 85 + with pytest.raises(AppError) as rename: + await svc.update(other_key, _body_payload(_key("OTHER"))) + assert rename.value.code == ErrorCode.BAD_REQUEST + + # archived 是终态 + await svc.archive(sm_key) + with pytest.raises(AppError) as rearchive: + await svc.archive(sm_key) + assert rearchive.value.code == ErrorCode.STATE_CONFLICT + finally: + await _cleanup() diff --git a/tests/unit/test_rules_dsl.py b/tests/unit/test_rules_dsl.py new file mode 100644 index 0000000..f1fec21 --- /dev/null +++ b/tests/unit/test_rules_dsl.py @@ -0,0 +1,260 @@ +"""kb_rule DSL 解析校验与 match 矩阵（TD-10，纯逻辑）。 + +核心验收点：**非法结构 100% 拒绝**——未识别的键（顶层/子句/条件字段）全部 +在解析时抛错，而不是被静默忽略后变成一条永不生效的死规则。 +""" + +from __future__ import annotations + +import pytest +from pydantic import ValidationError + +from packages.rules import dsl + +VALID_RULE = { + "rule_key": "R-BODY-A-002", + "category": "body_shape", + "priority": 90, + "conditions": { + "body_shape": "A型", + "features": {"include": ["溜肩"]}, + "goals": {"include": ["修饰溜肩"]}, + }, + "effects": { + "recommend": [ + {"slot": "top", "attrs": {"shoulder": ["垫肩", "自然肩"], "sleeve": ["泡泡袖"]}} + ], + "avoid": [{"slot": "top", "attrs": {"shoulder": ["落肩"], "silhouette": ["修身"]}}], + "explanation": "溜肩需要在肩部建立结构感", + }, + "is_hard": False, + "confidence": 0.95, +} + + +# ── 解析：合法结构 ───────────────────────────────────────────────────────── + + +def test_parse_doc_example() -> None: + rule = dsl.Rule.model_validate(VALID_RULE) + assert rule.rule_key == "R-BODY-A-002" + assert rule.priority == 90 + assert rule.is_hard is False + assert rule.confidence == pytest.approx(0.95) + # 等值子句解析为 str，集合子句解析为 ConditionClause + assert rule.conditions.clauses["body_shape"] == "A型" + features = rule.conditions.clauses["features"] + assert isinstance(features, dsl.ConditionClause) + assert features.include == ("溜肩",) + assert rule.effects.recommend[0].slot == "top" + assert rule.effects.avoid[0].attrs["shoulder"] == ("落肩",) + + +def test_parse_min_max_and_exclude_clause() -> None: + rule = dsl.Rule.model_validate( + { + **VALID_RULE, + "conditions": { + "temperature": {"min": 18, "max": 26}, + "season": {"exclude": ["夏"]}, + }, + } + ) + temp = rule.conditions.clauses["temperature"] + assert isinstance(temp, dsl.ConditionClause) + assert temp.min == 18.0 + assert temp.max == 26.0 + season = rule.conditions.clauses["season"] + assert isinstance(season, dsl.ConditionClause) + assert season.exclude == ("夏",) + + +def test_empty_conditions_means_match_all() -> None: + # 合规类兜底规则（"全场禁真兽皮"）无条件命中是合法语义。 + rule = dsl.Rule.model_validate({**VALID_RULE, "conditions": {}}) + assert rule.conditions.clauses == {} + assert dsl.match(rule.conditions, dsl.RuleContext()) is True + + +def test_explanation_only_effect_is_valid() -> None: + rule = dsl.Rule.model_validate( + {**VALID_RULE, "effects": {"explanation": "整体偏保守的通勤思路"}} + ) + assert rule.effects.recommend == () + assert rule.effects.avoid == () + + +def test_attrs_empty_means_whole_slot() -> None: + rule = dsl.Rule.model_validate( + {**VALID_RULE, "effects": {"avoid": [{"slot": "swimsuit"}], "explanation": "回避品类"}} + ) + assert rule.effects.avoid[0].attrs == {} + + +# ── 解析：非法结构 100% 拒绝 ─────────────────────────────────────────────── + + +@pytest.mark.parametrize( + "mutation", + [ + {"rule_key": 12345}, # 非 str + {"rule_key": "R-" + "x" * 63}, # 超出 64 字符（列宽） + {"category": "c" * 33}, # 超出 32 字符（列宽） + {"priority": 101}, # 超出分组上限 100 + {"priority": -1}, + {"confidence": 1.5}, + {"confidence": -0.1}, + ], +) +def test_invalid_top_level_values_rejected(mutation: dict) -> None: + with pytest.raises(ValidationError): + dsl.Rule.model_validate({**VALID_RULE, **mutation}) + + +def test_unknown_top_level_key_rejected() -> None: + with pytest.raises(ValidationError, match="unexpected"): + dsl.Rule.model_validate({**VALID_RULE, "unexpected": 1}) + + +def test_unknown_condition_field_rejected() -> None: + # 条件字段名 typo：静默忽略会让这条规则永不命中 + with pytest.raises(ValidationError, match="未识别的条件字段"): + dsl.Rule.model_validate({**VALID_RULE, "conditions": {"body_shap": "A型"}}) + + +def test_unknown_clause_key_rejected() -> None: + with pytest.raises(ValidationError): + dsl.Rule.model_validate({**VALID_RULE, "conditions": {"features": {"includes": ["溜肩"]}}}) + + +def test_empty_clause_rejected() -> None: + with pytest.raises(ValidationError, match="至少要包含"): + dsl.Rule.model_validate({**VALID_RULE, "conditions": {"features": {}}}) + + +def test_min_greater_than_max_rejected() -> None: + with pytest.raises(ValidationError, match="min 不能大于 max"): + dsl.Rule.model_validate( + {**VALID_RULE, "conditions": {"temperature": {"min": 26, "max": 18}}} + ) + + +def test_empty_attr_values_rejected() -> None: + with pytest.raises(ValidationError, match="取值列表不能为空"): + dsl.Rule.model_validate( + { + **VALID_RULE, + "effects": { + "recommend": [{"slot": "top", "attrs": {"shoulder": []}}], + "explanation": "x", + }, + } + ) + + +def test_empty_attr_dim_rejected() -> None: + with pytest.raises(ValidationError, match="维度名不能为空"): + dsl.Rule.model_validate( + { + **VALID_RULE, + "effects": { + "recommend": [{"slot": "top", "attrs": {"": ["垫肩"]}}], + "explanation": "x", + }, + } + ) + + +def test_empty_effects_rejected() -> None: + with pytest.raises(ValidationError, match="至少要包含"): + dsl.Rule.model_validate({**VALID_RULE, "effects": {}}) + + +def test_blank_explanation_alone_rejected() -> None: + # explanation="" 视同未填，效果为空的规则没有存在意义 + with pytest.raises(ValidationError, match="至少要包含"): + dsl.Rule.model_validate({**VALID_RULE, "effects": {"explanation": ""}}) + + +def test_non_dict_conditions_rejected() -> None: + with pytest.raises(ValidationError): + dsl.Rule.model_validate({**VALID_RULE, "conditions": ["body_shape"]}) + + +def test_effects_missing_rejected() -> None: + payload = dict(VALID_RULE) + del payload["effects"] + with pytest.raises(ValidationError): + dsl.Rule.model_validate(payload) + + +# ── match 矩阵 ───────────────────────────────────────────────────────────── + + +def _cond(**clauses: object) -> dsl.RuleCondition: + return dsl.RuleCondition.model_validate(clauses) + + +def test_match_equality_hit_and_miss() -> None: + cond = _cond(body_shape="A型") + assert dsl.match(cond, dsl.RuleContext(body_shape="A型")) is True + assert dsl.match(cond, dsl.RuleContext(body_shape="H型")) is False + assert dsl.match(cond, dsl.RuleContext(body_shape=None)) is False # 缺字段 + + +def test_match_equality_on_collection_field_is_membership() -> None: + cond = _cond(user_taboos="露脐") + assert dsl.match(cond, dsl.RuleContext(user_taboos=("露脐", "豹纹"))) is True + assert dsl.match(cond, dsl.RuleContext(user_taboos=())) is False + + +def test_match_include_requires_all_values() -> None: + cond = _cond(features={"include": ["溜肩", "宽胯"]}) + assert dsl.match(cond, dsl.RuleContext(features=("溜肩", "宽胯", "小腿粗"))) is True + # 子集语义：只命中一半不算命中 + assert dsl.match(cond, dsl.RuleContext(features=("溜肩",))) is False + assert dsl.match(cond, dsl.RuleContext(features=())) is False + + +def test_match_include_on_scalar_field() -> None: + cond = _cond(occasion={"include": ["面试", "通勤"]}) + assert dsl.match(cond, dsl.RuleContext(occasion="面试")) is True + assert dsl.match(cond, dsl.RuleContext(occasion="约会")) is False + + +def test_match_exclude() -> None: + cond = _cond(features={"exclude": ["溜肩"]}) + assert dsl.match(cond, dsl.RuleContext(features=("宽胯",))) is True + assert dsl.match(cond, dsl.RuleContext(features=("宽胯", "溜肩"))) is False + + +def test_match_min_max_closed_interval() -> None: + cond = _cond(temperature={"min": 18, "max": 26}) + assert dsl.match(cond, dsl.RuleContext(temperature=18.0)) is True # 闭区间含端点 + assert dsl.match(cond, dsl.RuleContext(temperature=26.0)) is True + assert dsl.match(cond, dsl.RuleContext(temperature=21.5)) is True + assert dsl.match(cond, dsl.RuleContext(temperature=17.9)) is False + assert dsl.match(cond, dsl.RuleContext(temperature=26.1)) is False + assert dsl.match(cond, dsl.RuleContext(temperature=None)) is False # 缺字段 + + +def test_match_min_only_and_max_only() -> None: + assert dsl.match(_cond(temperature={"min": 10}), dsl.RuleContext(temperature=10.0)) + assert not dsl.match(_cond(temperature={"min": 10}), dsl.RuleContext(temperature=9.9)) + assert dsl.match(_cond(temperature={"max": 10}), dsl.RuleContext(temperature=10.0)) + assert not dsl.match(_cond(temperature={"max": 10}), dsl.RuleContext(temperature=10.1)) + + +def test_match_clauses_are_anded() -> None: + cond = _cond(body_shape="A型", features={"include": ["溜肩"]}) + assert dsl.match(cond, dsl.RuleContext(body_shape="A型", features=("溜肩",))) is True + assert dsl.match(cond, dsl.RuleContext(body_shape="A型", features=())) is False + assert dsl.match(cond, dsl.RuleContext(body_shape="H型", features=("溜肩",))) is False + + +def test_match_include_and_min_max_combined_in_one_clause() -> None: + # 同一子句内 include 与区间也按 AND 处理 + cond = _cond(goals={"include": ["显瘦"], "exclude": ["显壮"]}) + assert dsl.match(cond, dsl.RuleContext(goals=("显瘦",))) is True + assert dsl.match(cond, dsl.RuleContext(goals=("显瘦", "显壮"))) is False + assert dsl.match(cond, dsl.RuleContext(goals=("显高",))) is False diff --git a/tests/unit/test_rules_evaluation.py b/tests/unit/test_rules_evaluation.py new file mode 100644 index 0000000..65dadac --- /dev/null +++ b/tests/unit/test_rules_evaluation.py @@ -0,0 +1,322 @@ +"""纯域求值引擎：排序、硬约束拦截、用户禁忌压过专业推荐、性能基准（T-P2-02）。""" + +from __future__ import annotations + +import time +from typing import Any + +from packages.rules import dsl, engine + + +def _rule( + rule_key: str, + priority: int, + *, + category: str = "test", + confidence: float = 0.9, + is_hard: bool = False, + conditions: dict[str, Any] | None = None, + recommend: tuple[dict[str, Any], ...] = (), + avoid: tuple[dict[str, Any], ...] = (), + explanation: str | None = None, +) -> dsl.Rule: + effects: dict[str, Any] = {} + if recommend: + effects["recommend"] = list(recommend) + if avoid: + effects["avoid"] = list(avoid) + if explanation: + effects["explanation"] = explanation + if not effects: + effects["explanation"] = "测试规则" # 空效果本身非法，测试助手兜底 + return dsl.Rule.model_validate( + { + "rule_key": rule_key, + "category": category, + "priority": priority, + "conditions": conditions or {}, + "effects": effects, + "is_hard": is_hard, + "confidence": confidence, + } + ) + + +# ── 排序与命中链 ─────────────────────────────────────────────────────────── + + +def test_hits_sorted_by_priority_then_confidence_then_key() -> None: + rules = [ + _rule("R-LOW", 50), + _rule("R-TIE-B", 90, confidence=0.9), + _rule("R-TOP", 95, confidence=0.5), + _rule("R-TIE-A", 90, confidence=0.9), + _rule("R-TIE-C", 90, confidence=0.8), + ] + verdict = engine.evaluate(rules, dsl.RuleContext()) + assert [h.rule_key for h in verdict.rule_hits] == [ + "R-TOP", + "R-TIE-A", + "R-TIE-B", + "R-TIE-C", + "R-LOW", + ] + assert [h.priority for h in verdict.rule_hits] == [95, 90, 90, 90, 50] + + +def test_rule_hit_carries_metadata_and_explanations_in_priority_order() -> None: + rules = [ + _rule("R-LOW", 50, explanation="趋势解释", category="trend", confidence=0.7), + _rule("R-HIGH", 90, explanation="体型解释", category="body_shape", confidence=0.95), + ] + verdict = engine.evaluate(rules, dsl.RuleContext()) + top = verdict.rule_hits[0] + assert (top.rule_key, top.priority, top.category, top.confidence) == ( + "R-HIGH", + 90, + "body_shape", + 0.95, + ) + assert top.is_hard is False + assert verdict.explanations == ("体型解释", "趋势解释") + assert verdict.explanation == "体型解释；趋势解释" + + +def test_no_rules_or_no_hits_yield_empty_verdict() -> None: + assert engine.evaluate([], dsl.RuleContext()).recommend == () + verdict = engine.evaluate( + [_rule("R-X", 90, conditions={"body_shape": "A型"})], + dsl.RuleContext(body_shape="H型"), + ) + assert verdict.rule_hits == () + assert verdict.blocked is False + + +# ── 硬约束 ───────────────────────────────────────────────────────────────── + + +def test_hard_rule_with_avoid_populates_hard_blocks() -> None: + avoid = {"slot": "shoes", "attrs": {"material": ["真皮"]}} + rules = [_rule("R-HARD", 100, is_hard=True, avoid=(avoid,), explanation="合规红线")] + verdict = engine.evaluate(rules, dsl.RuleContext()) + assert verdict.blocked is True + assert len(verdict.hard_blocks) == 1 + block = verdict.hard_blocks[0] + assert block.rule_key == "R-HARD" + assert block.avoid[0].slot == "shoes" + assert block.avoid[0].attrs["material"] == ("真皮",) + assert block.explanation == "合规红线" + + +def test_hard_rule_without_avoid_does_not_block() -> None: + # 拦截语义只由 avoid 表达：is_hard 规则只写 recommend 就无物可拦 + rules = [ + _rule( + "R-HARD-REC", + 100, + is_hard=True, + recommend=({"slot": "top", "attrs": {"collar": ["衬衫领"]}},), + ) + ] + verdict = engine.evaluate(rules, dsl.RuleContext()) + assert verdict.hard_blocks == () + assert verdict.blocked is False + + +def test_hard_rule_not_fired_does_not_block() -> None: + rules = [ + _rule( + "R-HARD", + 100, + is_hard=True, + conditions={"occasion": "面试"}, + avoid=({"slot": "top", "attrs": {"neckline": ["深V"]}},), + ) + ] + assert engine.evaluate(rules, dsl.RuleContext(occasion="通勤")).blocked is False + assert engine.evaluate(rules, dsl.RuleContext(occasion="面试")).blocked is True + + +def test_soft_rule_with_avoid_never_blocks() -> None: + rules = [_rule("R-SOFT", 90, avoid=({"slot": "top", "attrs": {"silhouette": ["修身"]}},))] + assert engine.evaluate(rules, dsl.RuleContext()).hard_blocks == () + + +# ── 冲突仲裁：高优先级 avoid 剔除严格低优先级的冲突 recommend ──────────────── + + +def test_user_taboo_overrides_professional_recommend() -> None: + # 用户禁忌（95）与专业推荐（90）同 slot 同维度取值相交 → 推荐被剔除 + rules = [ + _rule( + "R-PRO", + dsl.PRIORITY_BODY_SHAPE, + recommend=({"slot": "top", "attrs": {"shoulder": ["落肩"]}},), + ), + _rule( + "R-TABOO", + dsl.PRIORITY_USER_TABOO, + conditions={"user_taboos": {"include": ["oversize"]}}, + avoid=({"slot": "top", "attrs": {"shoulder": ["落肩"]}},), + ), + ] + verdict = engine.evaluate(rules, dsl.RuleContext(user_taboos=("oversize",))) + assert verdict.recommend == () + assert verdict.avoid != () # 禁忌约束保留在 avoid 里供下游过滤 + assert [h.rule_key for h in verdict.rule_hits] == ["R-TABOO", "R-PRO"] + + +def test_taboo_keeps_non_conflicting_recommend() -> None: + # 不同维度（length vs shoulder）不构成冲突：专业推荐保留 + rules = [ + _rule( + "R-PRO", + 90, + recommend=({"slot": "top", "attrs": {"shoulder": ["垫肩"]}},), + ), + _rule( + "R-TABOO", + 95, + avoid=({"slot": "top", "attrs": {"length": ["短款"]}},), + ), + ] + verdict = engine.evaluate(rules, dsl.RuleContext()) + assert [item.slot for item in verdict.recommend] == ["top"] + assert verdict.recommend[0].attrs["shoulder"] == ("垫肩",) + + +def test_taboo_does_not_override_higher_priority_recommend() -> None: + # 合规（100）> 用户禁忌（95）：只有"严格更低"优先级的推荐才会被剔除 + rules = [ + _rule( + "R-COMPLIANCE", + dsl.PRIORITY_COMPLIANCE, + recommend=({"slot": "top", "attrs": {"shoulder": ["落肩"]}},), + ), + _rule( + "R-TABOO", + dsl.PRIORITY_USER_TABOO, + avoid=({"slot": "top", "attrs": {"shoulder": ["落肩"]}},), + ), + ] + verdict = engine.evaluate(rules, dsl.RuleContext()) + assert len(verdict.recommend) == 1 + + +def test_lower_priority_recommend_vetoed_by_higher_soft_avoid() -> None: + # 通用优先级语义：体型 avoid（90）同样压过趋势 recommend（50） + rules = [ + _rule( + "R-TREND", + dsl.PRIORITY_TREND, + recommend=({"slot": "top", "attrs": {"silhouette": ["修身"]}},), + ), + _rule( + "R-BODY", + dsl.PRIORITY_BODY_SHAPE, + avoid=({"slot": "top", "attrs": {"silhouette": ["修身"]}},), + ), + ] + verdict = engine.evaluate(rules, dsl.RuleContext()) + assert verdict.recommend == () + + +def test_same_priority_avoid_does_not_veto_recommend() -> None: + # 同优先级不仲裁：推荐与回避并存，由下游按场景取舍 + rules = [ + _rule( + "R-A", + 90, + recommend=({"slot": "top", "attrs": {"sleeve": ["泡泡袖"]}},), + ), + _rule( + "R-B", + 90, + avoid=({"slot": "top", "attrs": {"sleeve": ["泡泡袖"]}},), + ), + ] + verdict = engine.evaluate(rules, dsl.RuleContext()) + assert len(verdict.recommend) == 1 + + +def test_different_slot_never_conflicts() -> None: + rules = [ + _rule( + "R-REC", + 50, + recommend=({"slot": "bottom", "attrs": {"silhouette": ["紧身"]}},), + ), + _rule( + "R-AVOID", + 100, + is_hard=True, + avoid=({"slot": "top", "attrs": {"silhouette": ["紧身"]}},), + ), + ] + verdict = engine.evaluate(rules, dsl.RuleContext()) + assert len(verdict.recommend) == 1 + assert verdict.blocked is True # 拦截归拦截，推荐归推荐 + + +# ── 汇总与去重 ───────────────────────────────────────────────────────────── + + +def test_identical_recommend_deduped_and_order_follows_priority() -> None: + item = {"slot": "top", "attrs": {"collar": ["衬衫领"]}} + rules = [ + _rule("R-STYLE", 70, recommend=(item,)), + _rule("R-OCCASION", 80, recommend=(item,)), + ] + verdict = engine.evaluate(rules, dsl.RuleContext()) + assert len(verdict.recommend) == 1 + assert len(verdict.avoid) == 0 + + +def test_identical_avoid_deduped() -> None: + avoid = {"slot": "top", "attrs": {"silhouette": ["修身"]}} + rules = [ + _rule("R-A", 90, avoid=(avoid,)), + _rule("R-B", 60, avoid=(avoid,)), + ] + verdict = engine.evaluate(rules, dsl.RuleContext()) + assert len(verdict.avoid) == 1 + + +# ── 性能基准（T-P2-02：p95 < 50ms @ 400 条规则）───────────────────────────── + + +def test_benchmark_400_rules_well_under_50ms_budget() -> None: + """400 条规则 × 1000 次求值。 + + 断言均值 < 5ms/次（对 50ms 预算留 10 倍裕量，CI 机器差异不至于 flaky）， + 并直接断言 p95 < 50ms 对应验收指标。求值是纯内存操作，规则从缓存来， + 不含任何 I/O。 + """ + shapes = ("A型", "H型", "X型", "O型") + rules = [ + _rule( + f"R-BENCH-{i:03d}", + 50 + (i % 6) * 10, + category="bench", + confidence=0.5 + (i % 10) / 20, + conditions={"body_shape": shapes[i % 4], "season": {"exclude": ["夏"]}}, + recommend=({"slot": "top", "attrs": {"collar": [f"领型{i % 5}"]}},), + avoid=({"slot": "bottom", "attrs": {"silhouette": ["紧身"]}},), + explanation=f"基准规则 {i}", + ) + for i in range(400) + ] + assert len(rules) == 400 + ctx = dsl.RuleContext(body_shape="A型", season="冬", temperature=18.0) + + engine.evaluate(rules, ctx) # 预热：首次调用的函数定位等开销不摊给被测对象 + timings: list[float] = [] + for _ in range(1000): + start = time.perf_counter() + engine.evaluate(rules, ctx) + timings.append(time.perf_counter() - start) + + mean = sum(timings) / len(timings) + p95 = sorted(timings)[int(len(timings) * 0.95) - 1] + assert mean < 0.005, f"平均单次求值 {mean * 1000:.2f}ms，超出 5ms 基线" + assert p95 < 0.05, f"p95 单次求值 {p95 * 1000:.2f}ms，超出 50ms 预算" diff --git a/tests/unit/test_rules_golden.py b/tests/unit/test_rules_golden.py new file mode 100644 index 0000000..0da29ea --- /dev/null +++ b/tests/unit/test_rules_golden.py @@ -0,0 +1,330 @@ +"""规则引擎黄金集雏形（架构 §1081）：固定画像 → 期望输出。 + +改引擎行为必须先过黄金集——这是"专业可信"的回归防线。必含两类场景： +硬禁忌（用户禁忌压过专业推荐）与多规则叠加。规则库是知识库文档 §3.2⑩ 的 +迷你缩影（体型/场景/温度/风格/趋势/用户禁忌/合规硬约束各至少一条）。 +""" + +from __future__ import annotations + +from collections.abc import Iterable +from typing import Any + +from packages.rules import dsl, engine + +# ── 固定规则库（8 条）─────────────────────────────────────────────────────── + +_GOLDEN_RULES = [ + # 体型（90）：文档 §3.2⑩ 原始示例规则 + { + "rule_key": "R-BODY-A-002", + "category": "body_shape", + "priority": 90, + "confidence": 0.95, + "conditions": { + "body_shape": "A型", + "features": {"include": ["溜肩"]}, + "goals": {"include": ["修饰溜肩"]}, + }, + "effects": { + "recommend": [ + {"slot": "top", "attrs": {"shoulder": ["垫肩", "自然肩"], "sleeve": ["泡泡袖"]}} + ], + "avoid": [{"slot": "top", "attrs": {"shoulder": ["落肩"], "silhouette": ["修身"]}}], + "explanation": "溜肩需要用垫肩与泡泡袖在肩部建立结构感。", + }, + }, + # 体型（90，同优先级低置信度：验证二级排序） + { + "rule_key": "R-BODY-A-010", + "category": "body_shape", + "priority": 90, + "confidence": 0.90, + "conditions": {"body_shape": "A型"}, + "effects": { + "recommend": [{"slot": "bottom", "attrs": {"silhouette": ["直筒", "阔腿"]}}], + "avoid": [{"slot": "bottom", "attrs": {"silhouette": ["紧身"]}}], + "explanation": "A型下半身宜直线轮廓，避免紧身放大胯部。", + }, + }, + # 场景（80） + { + "rule_key": "R-OCCASION-INTERVIEW-001", + "category": "occasion", + "priority": 80, + "confidence": 0.90, + "conditions": {"occasion": "面试"}, + "effects": { + "recommend": [ + {"slot": "top", "attrs": {"collar": ["衬衫领"]}}, + {"slot": "bottom", "attrs": {"silhouette": ["直筒"]}}, + ], + "avoid": [{"slot": "top", "attrs": {"neckline": ["深V"]}}], + "explanation": "面试场合需要干练可信的轮廓。", + }, + }, + # 温度（75）：闭区间 + { + "rule_key": "R-TEMP-COMFORT-18-26", + "category": "temperature", + "priority": 75, + "confidence": 0.85, + "conditions": {"temperature": {"min": 18, "max": 26}}, + "effects": { + "recommend": [{"slot": "outer", "attrs": {"thickness": ["薄款"]}}], + "avoid": [{"slot": "outer", "attrs": {"thickness": ["厚款"]}}], + "explanation": "18-26℃ 单层薄外套最舒适。", + }, + }, + # 风格（70）：与场景规则的推荐完全相同 → 验证去重 + { + "rule_key": "R-STYLE-OL-001", + "category": "style", + "priority": 70, + "confidence": 0.80, + "conditions": {"style": "通勤"}, + "effects": { + "recommend": [{"slot": "top", "attrs": {"collar": ["衬衫领"]}}], + "explanation": "通勤风格以衬衫领立住精神气。", + }, + }, + # 趋势（50）：推荐恰与用户禁忌冲突 → 必须被压过 + { + "rule_key": "R-TREND-SWEET-001", + "category": "trend", + "priority": 50, + "confidence": 0.70, + "conditions": {"style": "甜酷"}, + "effects": { + "recommend": [{"slot": "top", "attrs": {"length": ["短款"]}}], + "explanation": "短款上衣是当季甜酷风的核心单品。", + }, + }, + # 用户禁忌（95）：avoid 永远压过专业推荐 + { + "rule_key": "R-TABOO-NAVEL-001", + "category": "user_taboo", + "priority": 95, + "confidence": 1.0, + "conditions": {"user_taboos": {"include": ["露脐"]}}, + "effects": { + "avoid": [{"slot": "top", "attrs": {"length": ["短款", "露脐"]}}], + "explanation": "用户不接受露脐，已剔除相关单品。", + }, + }, + # 合规（100）：is_hard，avoid 命中即硬拦截 + { + "rule_key": "R-HARD-EXOTIC-001", + "category": "compliance", + "priority": 100, + "confidence": 1.0, + "is_hard": True, + "conditions": {"occasion": {"include": ["红毯", "晚宴"]}}, + "effects": { + "avoid": [ + {"slot": "accessory", "attrs": {"material": ["真兽皮"]}}, + {"slot": "shoes", "attrs": {"material": ["真皮"]}}, + ], + "explanation": "合规红线：不推荐真兽皮材质。", + }, + }, +] + +_RULES = [dsl.Rule.model_validate(r) for r in _GOLDEN_RULES] + + +# ── 断言辅助：把 SlotAttrs 规整成可比较的集合 ──────────────────────────────── + + +#: 单个 recommend/avoid 条目规整后的形状。 +NormItem = tuple[str, tuple[tuple[str, tuple[str, ...]], ...]] + + +def _item(slot: str, **dims: tuple[str, ...]) -> NormItem: + # 与 _norm 相同的规整规则（维度与取值都排序），期望值不必手写排序 + return (slot, tuple(sorted((d, tuple(sorted(v))) for d, v in dims.items()))) + + +def _norm(items: Iterable[dsl.SlotAttrs]) -> frozenset[NormItem]: + return frozenset( + (i.slot, tuple(sorted((d, tuple(sorted(v))) for d, v in i.attrs.items()))) for i in items + ) + + +def _evaluate(**ctx: Any) -> engine.RuleVerdict: + return engine.evaluate(_RULES, dsl.RuleContext(**ctx)) + + +def _hit_keys(verdict: engine.RuleVerdict) -> list[str]: + return [h.rule_key for h in verdict.rule_hits] + + +# ── 1. 基础命中 ───────────────────────────────────────────────────────────── + + +def test_a_shape_with_sloped_shoulders() -> None: + verdict = _evaluate(body_shape="A型", features=("溜肩",), goals=("修饰溜肩",)) + assert _hit_keys(verdict) == ["R-BODY-A-002", "R-BODY-A-010"] + assert _norm(verdict.recommend) == { + _item("top", shoulder=("垫肩", "自然肩"), sleeve=("泡泡袖",)), + _item("bottom", silhouette=("阔腿", "直筒")), + } + assert _norm(verdict.avoid) == { + _item("top", shoulder=("落肩",), silhouette=("修身",)), + _item("bottom", silhouette=("紧身",)), + } + assert verdict.hard_blocks == () + assert verdict.blocked is False + + +def test_missing_body_shape_does_not_fire_shape_rules() -> None: + # 缺字段 = 不匹配该子句：只有 features/goals 时体型规则不命中 + verdict = _evaluate(features=("溜肩",), goals=("修饰溜肩",)) + assert _hit_keys(verdict) == [] + assert verdict.recommend == () + assert verdict.avoid == () + assert verdict.explanation == "" + + +def test_include_is_subset_semantics() -> None: + # features 是 include 列表的超集 → 命中 + hit = _evaluate(body_shape="A型", features=("溜肩", "宽胯"), goals=("修饰溜肩",)) + assert "R-BODY-A-002" in _hit_keys(hit) + # goals 缺一个 → 不命中 + miss = _evaluate(body_shape="A型", features=("溜肩",), goals=("显瘦",)) + assert "R-BODY-A-002" not in _hit_keys(miss) + assert _hit_keys(miss) == ["R-BODY-A-010"] + + +# ── 2. 多规则叠加 ────────────────────────────────────────────────────────── + + +def test_body_plus_temperature_plus_occasion_stack() -> None: + verdict = _evaluate( + body_shape="A型", + features=("溜肩",), + goals=("修饰溜肩",), + temperature=20.0, + occasion="面试", + ) + assert _hit_keys(verdict) == [ + "R-BODY-A-002", + "R-BODY-A-010", + "R-OCCASION-INTERVIEW-001", + "R-TEMP-COMFORT-18-26", + ] + # 四条规则的推荐全部叠加，互不冲突（不同维度/不同 slot；两条 bottom 推荐 + # 的属性集不同，各自保留） + assert _norm(verdict.recommend) == { + _item("top", shoulder=("垫肩", "自然肩"), sleeve=("泡泡袖",)), + _item("bottom", silhouette=("阔腿", "直筒")), + _item("top", collar=("衬衫领",)), + _item("bottom", silhouette=("直筒",)), + _item("outer", thickness=("薄款",)), + } + assert _norm(verdict.avoid) == { + _item("top", shoulder=("落肩",), silhouette=("修身",)), + _item("bottom", silhouette=("紧身",)), + _item("top", neckline=("深V",)), + _item("outer", thickness=("厚款",)), + } + assert verdict.blocked is False + + +def test_same_priority_sorted_by_confidence_and_dedupe_collapsed() -> None: + # A-002 不命中（缺 features/goals），A-010 与面试/通勤规则叠加； + # 场景(80)与风格(70)推荐同一件衬衫领 → 去重为一条 + verdict = _evaluate(body_shape="A型", occasion="面试", style="通勤") + assert _hit_keys(verdict) == [ + "R-BODY-A-010", + "R-OCCASION-INTERVIEW-001", + "R-STYLE-OL-001", + ] + assert _norm(verdict.recommend) == { + _item("bottom", silhouette=("阔腿", "直筒")), + _item("bottom", silhouette=("直筒",)), + _item("top", collar=("衬衫领",)), + } + assert verdict.explanations == ( + "A型下半身宜直线轮廓，避免紧身放大胯部。", + "面试场合需要干练可信的轮廓。", + "通勤风格以衬衫领立住精神气。", + ) + + +def test_temperature_closed_interval_boundaries() -> None: + assert _hit_keys(_evaluate(temperature=18.0)) == ["R-TEMP-COMFORT-18-26"] + assert _hit_keys(_evaluate(temperature=26.0)) == ["R-TEMP-COMFORT-18-26"] + assert _hit_keys(_evaluate(temperature=17.9)) == [] + assert _hit_keys(_evaluate(temperature=26.1)) == [] + + +# ── 3. 用户禁忌压过专业推荐（硬性场景）────────────────────────────────────── + + +def test_user_taboo_overrides_trend_recommendation() -> None: + # 趋势(50)推荐短款上衣，用户禁忌(95)回避短款/露脐 → 推荐被剔除 + verdict = _evaluate(style="甜酷", user_taboos=("露脐",)) + assert _hit_keys(verdict) == ["R-TABOO-NAVEL-001", "R-TREND-SWEET-001"] + assert verdict.recommend == () + assert _norm(verdict.avoid) == {_item("top", length=("露脐", "短款"))} + # 禁忌不是 is_hard：不构成硬拦截，但推荐侧已经干净 + assert verdict.hard_blocks == () + assert verdict.explanation.startswith("用户不接受露脐") + + +def test_user_taboo_keeps_unrelated_slot_recommendations() -> None: + # 禁忌只作用于 top.length：体型对 bottom 的推荐不受影响 + verdict = _evaluate(body_shape="A型", user_taboos=("露脐",)) + assert _hit_keys(verdict) == ["R-TABOO-NAVEL-001", "R-BODY-A-010"] + assert _norm(verdict.recommend) == {_item("bottom", silhouette=("阔腿", "直筒"))} + + +# ── 4. 硬约束拦截（is_hard）───────────────────────────────────────────────── + + +def test_hard_compliance_rule_blocks_gala_occasion() -> None: + verdict = _evaluate(occasion="红毯") + assert _hit_keys(verdict) == ["R-HARD-EXOTIC-001"] + assert verdict.blocked is True + assert [b.rule_key for b in verdict.hard_blocks] == ["R-HARD-EXOTIC-001"] + # 调用方用 hard_blocks.avoid 对候选做硬过滤：约束内容完整带出 + assert _norm(verdict.hard_blocks[0].avoid) == { + _item("accessory", material=("真兽皮",)), + _item("shoes", material=("真皮",)), + } + assert verdict.hard_blocks[0].explanation == "合规红线：不推荐真兽皮材质。" + + +def test_hard_rule_not_fired_on_other_occasions() -> None: + assert _evaluate(occasion="通勤").hard_blocks == () + assert _evaluate(occasion="通勤").blocked is False + + +# ── 5. 综合叠加：优先级链 + 禁忌压制 + 硬拦截同时发生 ──────────────────────── + + +def test_composite_profile_full_priority_chain() -> None: + verdict = _evaluate( + body_shape="A型", + features=("溜肩",), + goals=("修饰溜肩",), + style="甜酷", + user_taboos=("露脐",), + occasion="红毯", + ) + # 命中链严格按 priority 降序、同优先级按 confidence 降序 + assert _hit_keys(verdict) == [ + "R-HARD-EXOTIC-001", # 100 + "R-TABOO-NAVEL-001", # 95 + "R-BODY-A-002", # 90 / 0.95 + "R-BODY-A-010", # 90 / 0.90 + "R-TREND-SWEET-001", # 50 + ] + # 体型推荐保留（与禁忌不同维度）；趋势推荐短款被禁忌压掉 + assert _norm(verdict.recommend) == { + _item("top", shoulder=("垫肩", "自然肩"), sleeve=("泡泡袖",)), + _item("bottom", silhouette=("阔腿", "直筒")), + } + assert verdict.blocked is True + assert [b.rule_key for b in verdict.hard_blocks] == ["R-HARD-EXOTIC-001"] diff --git a/tests/unit/test_rules_service.py b/tests/unit/test_rules_service.py new file mode 100644 index 0000000..56e25f4 --- /dev/null +++ b/tests/unit/test_rules_service.py @@ -0,0 +1,122 @@ +"""service 层纯逻辑：状态机矩阵、DSL 写入校验、ORM 行 → 域规则转换。 + +不连库——DB 路径由 tests/integration/test_rules_engine.py 覆盖。 +""" + +from __future__ import annotations + +from decimal import Decimal + +import pytest + +from packages.rules import service +from packages.shared.errors import AppError, ErrorCode +from packages.shared.models import KbRule + +VALID_PAYLOAD = { + "rule_key": "R-TEST-001", + "category": "body_shape", + "priority": 90, + "conditions": {"body_shape": "A型"}, + "effects": { + "recommend": [{"slot": "top", "attrs": {"shoulder": ["垫肩"]}}], + "avoid": [{"slot": "top", "attrs": {"shoulder": ["落肩"]}}], + "explanation": "测试规则", + }, + "is_hard": False, + "confidence": 0.95, +} + +ALL_STATUSES = ("draft", "published", "archived") + + +# ── 状态机全跃迁矩阵 ─────────────────────────────────────────────────────── + + +def test_legal_transitions_accepted() -> None: + for current, target in ( + ("draft", "published"), # 发布 + ("published", "archived"), # 下线 + ("draft", "archived"), # 撤稿 + ): + service.ensure_transition(current, target) # 不抛即通过 + + +@pytest.mark.parametrize("current", ALL_STATUSES) +@pytest.mark.parametrize("target", ALL_STATUSES) +def test_illegal_transitions_rejected(current: str, target: str) -> None: + if (current, target) in service.LEGAL_TRANSITIONS: + pytest.skip("合法跃迁另测") + with pytest.raises(AppError) as excinfo: + service.ensure_transition(current, target) + assert excinfo.value.code == ErrorCode.STATE_CONFLICT + assert excinfo.value.data == {"current": current, "target": target} + + +# ── DSL 写入校验 ─────────────────────────────────────────────────────────── + + +def test_validate_rule_payload_accepts_valid_dsl() -> None: + rule = service.validate_rule_payload(VALID_PAYLOAD) + assert rule.rule_key == "R-TEST-001" + + +def test_validate_rule_payload_rejects_unknown_keys_with_details() -> None: + with pytest.raises(AppError) as excinfo: + service.validate_rule_payload({**VALID_PAYLOAD, "conclusions": []}) + assert excinfo.value.code == ErrorCode.VALIDATION_FAILED + assert excinfo.value.data["errors"] # 拒绝原因必须可读，不能只说"不合法" + assert any("conclusions" in err for err in excinfo.value.data["errors"]) + + +def test_validate_rule_payload_rejects_unknown_condition_field() -> None: + with pytest.raises(AppError) as excinfo: + service.validate_rule_payload({**VALID_PAYLOAD, "conditions": {"body_shap": "A型"}}) + assert excinfo.value.code == ErrorCode.VALIDATION_FAILED + + +# ── ORM 行 → 域规则 ──────────────────────────────────────────────────────── + + +def _row(**overrides: object) -> KbRule: + values: dict[str, object] = { + "rule_id": 1, + "rule_key": "R-ROW-001", + "category": "body_shape", + "version": 1, + "status": "published", + "priority": 90, + "conditions": {"body_shape": "A型"}, + "effects": { + "recommend": [{"slot": "top", "attrs": {"shoulder": ["垫肩"]}}], + "explanation": "行转换", + }, + "is_hard": False, + "confidence": Decimal("0.950"), + } + values.update(overrides) + return KbRule(**values) # type: ignore[arg-type] + + +def test_row_to_rule_parses_stored_jsonb() -> None: + rule = service.row_to_rule(_row()) + assert rule.rule_key == "R-ROW-001" + assert rule.confidence == pytest.approx(0.95) # Numeric(4,3) → float + assert rule.conditions.clauses["body_shape"] == "A型" + assert rule.effects.recommend[0].attrs["shoulder"] == ("垫肩",) + + +def test_row_to_rule_rejects_corrupted_stored_dsl() -> None: + # 库里数据坏了是 5xxxx 知识库数据错误，必须响亮失败而不是静默跳过 + with pytest.raises(AppError) as excinfo: + service.row_to_rule(_row(conditions={"mood": "开心"})) + assert excinfo.value.code == ErrorCode.CONFIG_ERROR + assert excinfo.value.data == {"rule_key": "R-ROW-001"} + + +def test_published_rules_cache_invalidate_resets() -> None: + cache = service.PublishedRules() + assert cache._rules is None + cache._rules = () + cache.invalidate() + assert cache._rules is None