文章目录
  1. 1. 摘要
  2. 2. 用Python实现简易可拓展的规则引擎
  3. 3. 简单整理下需求
  4. 4. 实现
  5. 5. 代码
  6. 6. 解析
  7. 7. 结束语

摘要

用Python实现简易可拓展的规则引擎 做这个规则引擎的初衷是用来实现一个可序列号为json,容易拓展的条件执行引擎,用在类似工作流的场景中,最终实现的效果希望是这样的: ![] 简单整理下需求 执行结果最终返回=true= or false 支持四则运算,逻辑运算以及自定义函数等 支持多…

用Python实现简易可拓展的规则引擎

做这个规则引擎的初衷是用来实现一个可序列号为json,容易拓展的条件执行引擎,用在类似工作流的场景中,最终实现的效果希望是这样的:

1

![]

简单整理下需求

执行结果最终返回=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值,当然这里也可以拓展改成允许返回任何值,然后根据返回值来决定后续走向,这便可以成为一个工作流中的条件节点了。

结束语

东西简单粗陋,希望能给大家带来一些帮助或者一些启发~

作者:癫行
地址:http://yq.aliyun.com/articles/2631?spm=5176.100240.searchblog.24


本文出处程序员头条:http://www.iswifting.com/2016/01/18/Python-archieve-engine/
转载请在开头注明本文出处。

欢迎关注本站微信公众号:为程序员提供最优质的博文、最精彩的讨论、最实用的开发资源;提供最新最全的编程学习资料:PHP、Objective-C、Java、Swift、C/C++函数库、.NET Framework类库、J2SE API等等.并不定期奉送各种福利.
微信公众号猿圈:CodePush

文章目录
  1. 1. 摘要
  2. 2. 用Python实现简易可拓展的规则引擎
  3. 3. 简单整理下需求
  4. 4. 实现
  5. 5. 代码
  6. 6. 解析
  7. 7. 结束语