from typing import TypeVar, Any, Generic, Dict, Optional
from olc.bean.olc_config_rule import BasePolicy
from olc.bean.tag_group import TagGroup
from olc.utils.num_utils import NumUtils
T = TypeVar("T")
class AlgContext(Generic[T]):
def __init__(
self,
group: TagGroup = None,
policy: BasePolicy = None,
parameter: Dict[str, Any] = None,
previous: T = None,
init_result: T = None,
max_result: T = None,
request_size: Optional[int] = None,
):
"""
动态算法上下文
:param group: 组
:param policy: 规则
:param parameter: 参数
:param previous: 上次结果
:param init_result: 初始结果
:param max_result: 最大结果
:param request_size: 请求体大小(字节);-1 表示头缺失/提取失败,None 表示定时刷新路径
"""
self.group: TagGroup = group
self.policy: BasePolicy = policy
self.parameter: Dict[str, Any] = parameter
self.previous: T = previous
self.init_result: T = init_result
self.max_result: T = max_result
self.request_size: Optional[int] = request_size
def get_float_value_by_dynamic(self, param: str) -> Optional[float]:
param = self.parameter.get(param)
return NumUtils.cast_to_float(param)
def get_float_value_by_param(self, param: str, default: float) -> Optional[float]:
param = self.parameter.get(param)
if param is None:
return default
else:
return NumUtils.cast_to_float(param)
class AlgContextBuilder(Generic[T]):
def __init__(self):
self.alg_context: AlgContext = AlgContext()
@classmethod
def new_builder(cls) -> "AlgContextBuilder":
return AlgContextBuilder()
def set_group(self, group: TagGroup) -> "AlgContextBuilder":
self.alg_context.group = group
return self
def set_policy(self, policy: BasePolicy) -> "AlgContextBuilder":
self.alg_context.policy = policy
return self
def set_parameter(self, parameter: Dict[str, Any]) -> "AlgContextBuilder":
self.alg_context.parameter = parameter
return self
def set_previous(self, previous: T) -> "AlgContextBuilder":
self.alg_context.previous = previous
return self
def set_init_result(self, init_result: T) -> "AlgContextBuilder":
self.alg_context.init_result = init_result
return self
def set_max_result(self, max_result: T) -> "AlgContextBuilder":
self.alg_context.max_result = max_result
return self
def set_request_size(self, request_size: int) -> "AlgContextBuilder":
"""设置请求体大小"""
self.alg_context.request_size = request_size
return self
def build(self) -> AlgContext:
return self.alg_context