做這個規則引擎的初衷是用來實現一個可序列號為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值,當然這里也可以拓展改成允許返回任何值,然后根據返回值來決定后續走向,這便可以成為一個工作流中的條件節點了。
結束語
東西簡單粗陋,希望能給大家帶來一些幫助或者一些啟發~