教你一招 | 用Python實現簡易可拓展的規則引擎


做這個規則引擎的初衷是用來實現一個可序列號為json,容易拓展的條件執行引擎,用在類似工作流的場景中,最終實現的效果希望是這樣的:

img

簡單整理下需求

  • 執行結果最終返回=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值,當然這里也可以拓展改成允許返回任何值,然后根據返回值來決定后續走向,這便可以成為一個工作流中的條件節點了。

結束語

東西簡單粗陋,希望能給大家帶來一些幫助或者一些啟發~


免責聲明!

本站轉載的文章為個人學習借鑒使用,本站對版權不負任何法律責任。如果侵犯了您的隱私權益,請聯系本站郵箱yoyou2525@163.com刪除。



 
粵ICP備18138465號   © 2018-2025 CODEPRJ.COM