2. 规则引擎基础：多规则顺序执行

任务说明

本案例实现一个最小可用的规则引擎：定义规则列表，按顺序逐条执行，收集所有命中规则的结果。每条规则是一个函数，接收数据字典，返回命中结果或 None（未命中）。

核心实现

from typing import Callable, Optional

# 规则类型：接收数据，命中返回结果字典，未命中返回 None
Rule = Callable[[dict], Optional[dict]]

def rule_high_amount(data: dict) -> Optional[dict]:
    """规则1：金额超过 10000 命中"""
    if data.get("amount", 0) > 10000:
        return {"rule": "high_amount", "reason": "金额超过1万"}
    return None

def rule_new_device(data: dict) -> Optional[dict]:
    """规则2：使用新设备命中"""
    if data.get("new_device"):
        return {"rule": "new_device", "reason": "新设备登录"}
    return None

def rule_blacklist(data: dict) -> Optional[dict]:
    """规则3：命中黑名单"""
    if data.get("user_id") in data.get("blacklist", []):
        return {"rule": "blacklist", "reason": "用户在黑名单中"}
    return None

def run_rules(rules: list[Rule], data: dict) -> list[dict]:
    """按顺序执行所有规则，返回所有命中结果"""
    hits = []
    for rule in rules:
        result = rule(data)
        if result is not None:
            hits.append(result)
    return hits

if __name__ == "__main__":
    rules = [rule_high_amount, rule_new_device, rule_blacklist]

    sample = {
        "user_id": "u1001",
        "amount": 25000,
        "new_device": True,
        "blacklist": ["u9999"],
    }

    hits = run_rules(rules, sample)
    for h in hits:
        print(h)
    print(f"共命中 {len(hits)} 条规则")

运行方式

python rule_engine.py

验证方法与通过条件

• 运行后应输出 2 条命中结果：high_amount 与 new_device，最后一行显示 共命中 2 条规则。
• 把 amount 改为 5000、new_device 改为 False，再运行应输出 共命中 0 条规则。
• 把 user_id 改为 "u9999"，应额外命中 blacklist。
• 命中顺序应与 rules 列表顺序一致，这是顺序执行的关键验证点。

关键点

• 规则统一签名 (dict) -> Optional[dict]，便于顺序遍历执行。
• 用 None 表示未命中，避免用异常或特殊值判断，逻辑清晰。
• 新增规则只需写一个同签名函数并加入 rules 列表，无需改动执行逻辑。