82 lines
2.4 KiB
Python
82 lines
2.4 KiB
Python
# -*- coding: utf-8 -*-
|
||
|
||
"""
|
||
封装ZenEngine
|
||
"""
|
||
|
||
# 导入模块
|
||
|
||
from datetime import datetime
|
||
from decimal import Decimal
|
||
from pathlib import Path
|
||
from typing import Any, Dict
|
||
|
||
from zen import ZenDecision, ZenEngine
|
||
|
||
|
||
class RuleEngine:
|
||
"""
|
||
规则引擎，实现打开并读取规则，根据规则和输入评估并输出
|
||
"""
|
||
|
||
def __init__(self, rules_path: Path):
|
||
"""
|
||
初始化规则引擎
|
||
:param rules_path: 规则文件夹路径（path对象）
|
||
"""
|
||
rules_path = Path(rules_path)
|
||
rules_path.mkdir(parents=True, exist_ok=True) # 若规则文件夹不存在则创建
|
||
|
||
# 初始化规则缓存
|
||
self.decisions = {}
|
||
for decision_path in rules_path.glob("*.json"):
|
||
# 打开并读取规则文件并实例化规则
|
||
self.decisions[decision_path.stem] = self._get_decision(decision_path)
|
||
|
||
@staticmethod
|
||
def _get_decision(decision_path: Path) -> ZenDecision:
|
||
"""
|
||
打开并读取规则文件并实例化规则
|
||
:param decision_path: 规则文件路径（path对象）
|
||
:return: 实例化规则
|
||
"""
|
||
|
||
def loader(path):
|
||
with open(path, "r", encoding="utf-8") as file:
|
||
return file.read()
|
||
|
||
return ZenEngine({"loader": loader}).get_decision(decision_path.as_posix())
|
||
|
||
def evaluate(self, decision: str, inputs: Dict[str, Any]) -> Dict[str, Any]:
|
||
"""
|
||
调用规则并评估
|
||
:param decision: 规则
|
||
:param inputs: 输入
|
||
:return: 输出
|
||
"""
|
||
if decision not in self.decisions:
|
||
raise ValueError(f"规则不存在：{decision}")
|
||
|
||
return self.decisions[decision].evaluate(self._format_value(inputs))["result"]
|
||
|
||
def _format_value(self, value: Any) -> Any:
|
||
"""
|
||
格式化值为字符串
|
||
:param value: 值
|
||
:return: 格式化后值
|
||
"""
|
||
match value:
|
||
case int():
|
||
return str(value)
|
||
case Decimal():
|
||
# noinspection PyTypeChecker
|
||
return format(value, f".{abs(value.as_tuple().exponent)}f")
|
||
case datetime():
|
||
return value.strftime("%Y-%m-%d %H:%M:%S")
|
||
case list():
|
||
return [self._format_value(e) for e in value]
|
||
case dict():
|
||
return {k: self._format_value(v) for k, v in value.items()}
|
||
case _:
|
||
return value
|