教你一招 | 用Python实现简易可拓展的规则引擎
教你一招 | 用Python实现简易可拓展的规则引擎
做这个规则引擎的初衷是用来实现一个可序列号为json，容易拓展的条件执行引擎，用在类似工作流的场景中，最终实现的效果希望是这样的：
简单整理下需求
- 执行结果最终返回=true= or false
- 支持四则运算，逻辑运算以及自定义函数等
- 支持多级规则组合，级别理论上无限（Python递归调用深度限制）
- 序列化成json
实现
json没有条件判断和流程控制，且不可引用对象，是不好序列化规则的，除非用树来保存，但这样又过于臃肿不好阅读。
在苦苦思索的时候，突然灵光一闪~曾经我用过一个自动装机系统—razor，
它使用一种tag语法来匹配机器并打标签，他的语法是这样的：
["or"，
["="， ["fact"， "macaddress"]， "de：ea：db：ee：f0：00"]
["="， ["fact"， "macaddress"]， "de：ea：db：ee：f0：01"]]这表示匹配目标机器的Mac地址等于=de：ea：db：ee：f0：00=或=de：ea：db：ee：f0：00=，这种表达既简洁，又足够灵活这种灵活体现在理论上可以无限嵌套，也可以随意自定义操作函数（这里的=、fact）
这灵感来自于古老的=Lisp=，完全可以实现我们的想法~并且简单、好用，还非常非常灵活！就它了！
因此我就使用这种基于=Json Array=的语法来实现我们的规则引擎。
最后实现的语法规则是这样的：
规则语法 基本语法： [“操作符”， “参数1”， “参数2”， …]
多条判断语句可组合，如：
["操作符"，
["操作符1"， "参数1"， "参数2"， ...]，["操作符2"， "参数1"， "参数2"， ...]
]
["and"，
[">"， 0 ， 0.05]，
[">"， 3， 2]
]支持的操作符： 比较运算符：
=， ！=， >， <， >=， <=逻辑运算符：
and， or， not， in四则运算：
+， -， *， /数据转换：
int， str， upper， lower其他特殊操作符：
可自定义操作符，例如get，从某http服务获取数据代码
class RuleParser(object)：
def __init__(self， rule)：
if isinstance(rule， basestring)：
self.rule = json.loads(rule)
else：
self.rule = rule
self.validate(self.rule)
class Functions(object)：
ALIAS = {
'='： 'eq'，
'！='： 'neq'，
'>'： 'gt'，
'>='： 'gte'，
'<'： 'lt'，
'<='： 'lte'，
'and'： 'and_'，
'in'： 'in_'，
'or'： 'or_'，
'not'： 'not_'，
'str'： 'str_'，
'int'： 'int_'，
'+'： 'plus'，
'-'： 'minus'，
'*'： 'multiply'，
'/'： 'divide'
}
def eq(self， *args)：
return args[0] == args[1]
def neq(self， *args)：
return args[0] ！= args[1]
def in_(self， *args)：
return args[0] in args[1：]
def gt(self， *args)：
return args[0] > args[1]
def gte(self， *args)：
return args[0] >= args[1]
def lt(self， *args)：
return args[0] < args[1]
def lte(self， *args)：
return args[0] <= args[1]
def not_(self， *args)：
return not args[0]
def or_(self， *args)：
return any(args)
def and_(self， *args)：
return all(args)
def int_(self， *args)：
return int(args[0])
def str_(self， *args)：
return unicode(args[0])
def upper(self， *args)：
return args[0].upper()
def lower(self， *args)：
return args[0].lower()
def plus(self， *args)：
return sum(args)
def minus(self， *args)：
return args[0] - args[1]
def multiply(self， *args)：
return args[0] * args[1]
def divide(self， *args)：
return float(args[0]) / float(args[1])
def abs(self， *args)：
return abs(args[0])
@staticmethod
def validate(rule)：
if not isinstance(rule， list)：
raise RuleEvaluationError('Rule must be a list， got {}'.format(type(rule)))
if len(rule) < 2：
raise RuleEvaluationError('Must have at least one argument.')
def _evaluate(self， rule， fns)：
"""
递归执行list内容
"""
def _recurse_eval(arg)：
if isinstance(arg， list)：
return self._evaluate(arg， fns)
else：
return arg
r = map(_recurse_eval， rule)
r[0] = self.Functions.ALIAS.get(r[0]) or r[0]
func = getattr(fns， r[0])
return func(*r[1：])
def evaluate(self)：
fns = self.Functions()
ret = self._evaluate(self.rule， fns)
if not isinstance(ret， bool)：
logger.warn('In common usage， a rule must return a bool value，'
'but get {}， please check the rule to ensure it is true' )
return ret解析
这里Functions这个类，就是用来存放操作符方法的，由于有些操作符不是合法的Python变量名，所以需要用ALIAS做一次转换。
当需要添加新的操作，只需在Functions中添加方法即可。由于始终使用array来存储，所以方法接收的参数始终可以用args[n]来访问到，这里没有做异常处理，如果想要更健壮的话可以拓展validate方法，以及在每次调用前检查参数。
整个规则引擎的核心代码其实就是=~evaluate~=这个10行不到的方法，在这里会递归遍历列表，从最里层的列表开始执行，然后层层往外执行，最后执行完毕返回一个Boolean值，当然这里也可以拓展改成允许返回任何值，然后根据返回值来决定后续走向，这便可以成为一个工作流中的条件节点了。
结束语
东西简单粗陋，希望能给大家带来一些帮助或者一些启发~
原文链接：https://yq.aliyun.com/articles/2631?utm_campaign=wenzhang&utm_medium=article&utm_source=QQ-qun&utm_content=m_8867