73 lines
2.2 KiB
Python
73 lines
2.2 KiB
Python
from dataclasses import dataclass
|
||
from typing import Callable, Dict, List, Optional
|
||
|
||
|
||
@dataclass
|
||
class MetricConfig:
|
||
"""
|
||
指标字段配置（完全去业务语义）
|
||
"""
|
||
field: str # 原始字段名，如 amount
|
||
name: str # 展示名称，如 实收金额 / 时长 / 评分
|
||
agg: str = "sum" # 聚合方式：sum / mean / max / min / count
|
||
format: str = "auto" # 展示格式：auto / money / number:2 / int / duration
|
||
|
||
|
||
class DomainConfig:
|
||
def __init__(
|
||
self,
|
||
name: str,
|
||
api_url: str,
|
||
entity_name: str,
|
||
field_name_map: Dict[str, str],
|
||
detail_fields: List[str] | None = None,
|
||
search_fields: List[str] | None = None,
|
||
metrics: Optional[List[MetricConfig]] = None,
|
||
token_getter: Callable[[], str] | None = None,
|
||
auth_header: str = "Authorization",
|
||
date_field: str = "date",
|
||
primary_key: str = "id",
|
||
):
|
||
self.name = name
|
||
self.api_url = api_url
|
||
self.entity_name = entity_name
|
||
self.field_name_map = field_name_map
|
||
self.detail_fields = detail_fields or []
|
||
self.search_fields = search_fields or []
|
||
self.metrics: List[MetricConfig] = metrics or []
|
||
|
||
self.token_getter = token_getter
|
||
self.auth_header = auth_header
|
||
self.date_field = date_field
|
||
self.primary_key = primary_key
|
||
|
||
self._metric_map: Dict[str, MetricConfig] = {
|
||
m.field: m for m in self.metrics
|
||
}
|
||
|
||
def get_metric(self, field: str) -> Optional[MetricConfig]:
|
||
return self._metric_map.get(field)
|
||
|
||
def metric_fields(self) -> List[str]:
|
||
return [m.field for m in self.metrics]
|
||
|
||
def has_metric(self, field: str) -> bool:
|
||
return field in self._metric_map
|
||
|
||
def get_chinese_name(self, field: str) -> str:
|
||
return self.field_name_map.get(field, field)
|
||
|
||
|
||
_DOMAINS: Dict[str, DomainConfig] = {}
|
||
|
||
|
||
def register_domain(config: DomainConfig):
|
||
print("注册业务域：", config.name)
|
||
_DOMAINS[config.name] = config
|
||
|
||
|
||
def get_domain(name: str) -> DomainConfig:
|
||
if name not in _DOMAINS:
|
||
raise ValueError(f"未注册的业务域：{name}")
|
||
return _DOMAINS[name]
|