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

2021-07-28 19:33:47 阅读：202 来源： 互联网

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

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

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

screenshot ![]

简单整理下需求

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

结束语

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

标签：Python,self,args,rule,简易,引擎,操作符,规则
来源： https://blog.csdn.net/yangyin007/article/details/119188052

本站声明： 1. iCode9 技术分享网（下文简称本站）提供的所有内容，仅供技术学习、探讨和分享；
2. 关于本站的所有留言、评论、转载及引用，纯属内容发起人的个人观点，与本站观点和立场无关；
3. 关于本站的所有言论和文字，纯属内容发起人的个人观点，与本站观点和立场无关；
4. 本站文章均是网友提供，不完全保证技术分享内容的完整性、准确性、时效性、风险性和版权归属；如您发现该文章侵犯了您的权益，可联系我们第一时间进行删除；
5. 本站为非盈利性的个人网站，所有内容不会用来进行牟利，也不会利用任何形式的广告来间接获益，纯粹是为了广大技术爱好者提供技术内容和技术思想的分享性交流网站。

ICode9

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

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

简单整理下需求

实现

代码

解析

结束语