475 lines
15 KiB
Python
475 lines
15 KiB
Python
"""模块：统计编排器
|
||
SAAS 级每日数据统计引擎核心入口——子服务通过此管理器配置和查询统计数据。
|
||
|
||
职责：
|
||
1. 配置统计指标和维度
|
||
2. 原子增量记录（支持并发安全）
|
||
3. 多维度聚合查询
|
||
4. 排行榜生成
|
||
5. 周期性重置（日清零/月清零）
|
||
6. 数据清理"""
|
||
|
||
import uuid
|
||
import datetime
|
||
import logging
|
||
from decimal import Decimal
|
||
from typing import List, Optional, Dict, Any
|
||
from gvsds.apps.stats.models import (
|
||
StatMetric,
|
||
StatDimension,
|
||
StatRecord,
|
||
StatRanking,
|
||
StatConfig,
|
||
)
|
||
from gvsds.apps.stats import engine as _eng
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
|
||
class StatEngine:
|
||
"""每日数据统计编排器——SAAS 主服务器统计核心入口
|
||
|
||
用法:
|
||
engine = StatEngine(tenant_uuid='xxx')
|
||
|
||
# 定义统计指标
|
||
metric = engine.DefineMetric('dashou_jiedan', '打手接单数', 'count')
|
||
|
||
# 原子增量记录
|
||
engine.Increment(metric_uuid=metric.MetricUUID,
|
||
dimension_values={'role': 'dashou', 'user_id': '123'},
|
||
delta=1)
|
||
|
||
# 查询今日统计
|
||
today = datetime.date.today()
|
||
records = engine.QueryToday(metric_uuid=metric.MetricUUID,
|
||
dimension_filter={'role': 'dashou'})
|
||
|
||
# 聚合排行
|
||
top = engine.AggregateTopN(metric_uuid=metric.MetricUUID,
|
||
period_type='daily',
|
||
start_key=today.isoformat(),
|
||
end_key=today.isoformat(),
|
||
top_n=50)
|
||
|
||
# 生成日榜快照
|
||
engine.GenerateDailyRanking(metric_uuid=metric.MetricUUID)
|
||
"""
|
||
|
||
def __init__(self, tenant_uuid):
|
||
self.TenantUUID = tenant_uuid
|
||
|
||
def DefineMetric(self, metric_code, metric_name, metric_type='count',
|
||
unit='次', default_value=None):
|
||
"""定义统计指标
|
||
|
||
Args:
|
||
metric_code: 指标编码（如 'dashou_jiedan'）
|
||
metric_name: 指标名称（如 '打手接单数'）
|
||
metric_type: 指标类型 count/amount/duration/ratio/custom
|
||
unit: 单位
|
||
default_value: 默认值
|
||
|
||
Returns:
|
||
StatMetric 实例
|
||
"""
|
||
metric, created = StatMetric.objects.update_or_create(
|
||
TenantUUID=self.TenantUUID,
|
||
MetricCode=metric_code,
|
||
defaults={
|
||
'MetricName': metric_name,
|
||
'MetricType': metric_type,
|
||
'Unit': unit,
|
||
'DefaultValue': default_value or 0,
|
||
'MetricStatus': 1,
|
||
},
|
||
)
|
||
logger.info(f'统计指标{"创建" if created else "更新"}: {metric_code}@tenant={self.TenantUUID}')
|
||
return metric
|
||
|
||
def DefineDimension(self, dimension_code, dimension_name):
|
||
"""定义统计维度
|
||
|
||
Args:
|
||
dimension_code: 维度编码（如 'role', 'platform'）
|
||
dimension_name: 维度名称（如 '角色', '平台'）
|
||
|
||
Returns:
|
||
StatDimension 实例
|
||
"""
|
||
dim, created = StatDimension.objects.update_or_create(
|
||
TenantUUID=self.TenantUUID,
|
||
DimensionCode=dimension_code,
|
||
defaults={
|
||
'DimensionName': dimension_name,
|
||
'DimensionStatus': 1,
|
||
},
|
||
)
|
||
return dim
|
||
|
||
def ConfigureAutoReset(self, metric_uuid, auto_reset=True,
|
||
reset_cron='0 0 * * *', enable_ranking=False,
|
||
ranking_top_n=50, retention_days=365):
|
||
"""配置指标的自动重置和排行榜
|
||
|
||
Args:
|
||
metric_uuid: 指标UUID
|
||
auto_reset: 是否每日自动重置
|
||
reset_cron: 重置时间cron表达式
|
||
enable_ranking: 是否启用排行榜
|
||
ranking_top_n: 排行榜取前N名
|
||
retention_days: 数据保留天数
|
||
|
||
Returns:
|
||
StatConfig 实例
|
||
"""
|
||
config, _ = StatConfig.objects.update_or_create(
|
||
TenantUUID=self.TenantUUID,
|
||
Metric_id=metric_uuid,
|
||
defaults={
|
||
'AutoReset': auto_reset,
|
||
'ResetCron': reset_cron,
|
||
'EnableRanking': enable_ranking,
|
||
'RankingTopN': ranking_top_n,
|
||
'RetentionDays': retention_days,
|
||
'ConfigStatus': 1,
|
||
},
|
||
)
|
||
return config
|
||
|
||
def Increment(self, metric_uuid, dimension_values, delta=1,
|
||
date=None, period_type='daily'):
|
||
"""原子增量——每日统计计数+1或金额+N
|
||
|
||
Args:
|
||
metric_uuid: 指标UUID (bytes)
|
||
dimension_values: 维度值字典 {'role': 'dashou', 'user_id': '123'}
|
||
delta: 增量（正数为加，负数为减）
|
||
date: 日期（默认今天）
|
||
period_type: 周期类型
|
||
|
||
Returns:
|
||
StatRecord 实例
|
||
"""
|
||
if date is None:
|
||
date = datetime.date.today()
|
||
|
||
if period_type == 'daily':
|
||
period_key = date.strftime('%Y-%m-%d')
|
||
elif period_type == 'monthly':
|
||
period_key = date.strftime('%Y-%m')
|
||
elif period_type == 'hourly':
|
||
period_key = date.strftime('%Y-%m-%dT%H')
|
||
else:
|
||
period_key = str(date)
|
||
|
||
record, created = _eng.AtomicIncrement(
|
||
tenant_uuid=self.TenantUUID,
|
||
period_type=period_type,
|
||
period_key=period_key,
|
||
metric_uuid=metric_uuid,
|
||
dimension_values=dimension_values,
|
||
delta=delta,
|
||
)
|
||
return record
|
||
|
||
def IncrementMulti(self, increments, date=None, period_type='daily'):
|
||
"""批量原子增量
|
||
|
||
Args:
|
||
increments: [(metric_uuid, dimension_values, delta), ...]
|
||
date: 日期
|
||
period_type: 周期类型
|
||
|
||
Returns:
|
||
List[StatRecord]
|
||
"""
|
||
if date is None:
|
||
date = datetime.date.today()
|
||
|
||
if period_type == 'daily':
|
||
period_key = date.strftime('%Y-%m-%d')
|
||
elif period_type == 'monthly':
|
||
period_key = date.strftime('%Y-%m')
|
||
else:
|
||
period_key = str(date)
|
||
|
||
return _eng.AtomicIncrementMulti(
|
||
tenant_uuid=self.TenantUUID,
|
||
period_type=period_type,
|
||
period_key=period_key,
|
||
increments=increments,
|
||
)
|
||
|
||
def Decrement(self, metric_uuid, dimension_values, delta=1, date=None):
|
||
"""原子减量——用于退款/取消场景的回滚
|
||
|
||
Args:
|
||
metric_uuid: 指标UUID
|
||
dimension_values: 维度值
|
||
delta: 减量
|
||
date: 日期
|
||
|
||
Returns:
|
||
StatRecord 实例
|
||
"""
|
||
return self.Increment(
|
||
metric_uuid=metric_uuid,
|
||
dimension_values=dimension_values,
|
||
delta=-delta,
|
||
date=date,
|
||
)
|
||
|
||
def QueryToday(self, metric_uuid, dimension_filter=None):
|
||
"""查询今日统计
|
||
|
||
Args:
|
||
metric_uuid: 指标UUID
|
||
dimension_filter: 维度过滤
|
||
|
||
Returns:
|
||
QuerySet[StatRecord]
|
||
"""
|
||
today = datetime.date.today().strftime('%Y-%m-%d')
|
||
return _eng.QueryPeriod(
|
||
self.TenantUUID, metric_uuid, 'daily', today, dimension_filter,
|
||
)
|
||
|
||
def QueryDate(self, metric_uuid, date, period_type='daily',
|
||
dimension_filter=None):
|
||
"""查询指定日期的统计
|
||
|
||
Args:
|
||
metric_uuid: 指标UUID
|
||
date: 日期
|
||
period_type: 周期类型
|
||
dimension_filter: 维度过滤
|
||
|
||
Returns:
|
||
QuerySet[StatRecord]
|
||
"""
|
||
if period_type == 'daily':
|
||
period_key = date.strftime('%Y-%m-%d') if hasattr(date, 'strftime') else str(date)
|
||
elif period_type == 'monthly':
|
||
period_key = date.strftime('%Y-%m') if hasattr(date, 'strftime') else str(date)
|
||
else:
|
||
period_key = str(date)
|
||
return _eng.QueryPeriod(
|
||
self.TenantUUID, metric_uuid, period_type, period_key, dimension_filter,
|
||
)
|
||
|
||
def QueryRange(self, metric_uuid, start_date, end_date,
|
||
period_type='daily', dimension_filter=None):
|
||
"""查询时间范围内的统计"""
|
||
start_key = start_date.strftime('%Y-%m-%d') if hasattr(start_date, 'strftime') else str(start_date)
|
||
end_key = end_date.strftime('%Y-%m-%d') if hasattr(end_date, 'strftime') else str(end_date)
|
||
return _eng.QueryRange(
|
||
self.TenantUUID, metric_uuid, period_type,
|
||
start_key, end_key, dimension_filter,
|
||
)
|
||
|
||
def AggregateTopN(self, metric_uuid, start_date, end_date,
|
||
dimension_key='user_id', period_type='daily', top_n=50):
|
||
"""按维度聚合TopN排名
|
||
|
||
Args:
|
||
metric_uuid: 指标UUID
|
||
start_date: 起始日期
|
||
end_date: 截止日期
|
||
dimension_key: 聚合的维度键
|
||
period_type: 周期类型
|
||
top_n: 取前N名
|
||
|
||
Returns:
|
||
list[{dim_value, total}]，前N名
|
||
"""
|
||
start_key = start_date.strftime('%Y-%m-%d') if hasattr(start_date, 'strftime') else str(start_date)
|
||
end_key = end_date.strftime('%Y-%m-%d') if hasattr(end_date, 'strftime') else str(end_date)
|
||
results = _eng.AggregateByDimension(
|
||
self.TenantUUID, metric_uuid, period_type,
|
||
start_key, end_key, dimension_key,
|
||
)
|
||
return results[:top_n]
|
||
|
||
def GenerateDailyRanking(self, metric_uuid, ranking_date=None,
|
||
dimension_filter=None, top_n=50):
|
||
"""生成日排行榜快照
|
||
|
||
Args:
|
||
metric_uuid: 指标UUID
|
||
ranking_date: 排行日期（默认今天）
|
||
dimension_filter: 过滤条件
|
||
top_n: 取前N名
|
||
|
||
Returns:
|
||
List[StatRanking]
|
||
"""
|
||
if ranking_date is None:
|
||
ranking_date = datetime.date.today()
|
||
|
||
StatRanking.objects.filter(
|
||
TenantUUID=self.TenantUUID,
|
||
RankingPeriod='daily',
|
||
RankingDate=ranking_date,
|
||
Metric_id=metric_uuid,
|
||
).delete()
|
||
|
||
return _eng.GenerateRanking(
|
||
tenant_uuid=self.TenantUUID,
|
||
metric_uuid=metric_uuid,
|
||
ranking_period='daily',
|
||
ranking_date=ranking_date,
|
||
dimension_filter=dimension_filter,
|
||
top_n=top_n,
|
||
)
|
||
|
||
def GenerateMonthlyRanking(self, metric_uuid, ranking_date=None,
|
||
dimension_filter=None, top_n=50):
|
||
"""生成月排行榜快照"""
|
||
if ranking_date is None:
|
||
today = datetime.date.today()
|
||
ranking_date = today.replace(day=1)
|
||
|
||
StatRanking.objects.filter(
|
||
TenantUUID=self.TenantUUID,
|
||
RankingPeriod='monthly',
|
||
RankingDate=ranking_date,
|
||
Metric_id=metric_uuid,
|
||
).delete()
|
||
|
||
return _eng.GenerateRanking(
|
||
tenant_uuid=self.TenantUUID,
|
||
metric_uuid=metric_uuid,
|
||
ranking_period='monthly',
|
||
ranking_date=ranking_date,
|
||
dimension_filter=dimension_filter,
|
||
top_n=top_n,
|
||
)
|
||
|
||
def QueryRanking(self, ranking_period, ranking_date=None,
|
||
metric_uuid=None, top_n=None):
|
||
"""查询排行榜快照
|
||
|
||
Args:
|
||
ranking_period: 'daily' | 'monthly'
|
||
ranking_date: 排行日期
|
||
metric_uuid: 指标UUID（None=全部）
|
||
top_n: 取前N名
|
||
|
||
Returns:
|
||
QuerySet[StatRanking]
|
||
"""
|
||
qs = StatRanking.objects.filter(
|
||
TenantUUID=self.TenantUUID,
|
||
RankingPeriod=ranking_period,
|
||
)
|
||
if ranking_date:
|
||
qs = qs.filter(RankingDate=ranking_date)
|
||
if metric_uuid:
|
||
qs = qs.filter(Metric_id=metric_uuid)
|
||
|
||
qs = qs.order_by('Metric', 'Rank')
|
||
|
||
if top_n:
|
||
metric_ids = qs.values_list('Metric_id', flat=True).distinct()
|
||
filtered = StatRanking.objects.none()
|
||
for mid in metric_ids:
|
||
filtered |= qs.filter(Metric_id=mid)[:top_n]
|
||
return filtered
|
||
|
||
return qs
|
||
|
||
def DailyReset(self, metric_uuid, target_date=None):
|
||
"""手动执行日清零
|
||
|
||
Args:
|
||
metric_uuid: 指标UUID
|
||
target_date: 目标日期（默认昨天）
|
||
|
||
Returns:
|
||
int 删除记录数
|
||
"""
|
||
if target_date is None:
|
||
target_date = datetime.date.today() - datetime.timedelta(days=1)
|
||
|
||
period_key = target_date.strftime('%Y-%m-%d')
|
||
return _eng.ResetPeriod(
|
||
self.TenantUUID, metric_uuid, 'daily', period_key,
|
||
)
|
||
|
||
def MonthlyReset(self, metric_uuid, target_date=None):
|
||
"""手动执行月清零"""
|
||
if target_date is None:
|
||
today = datetime.date.today()
|
||
last_month = today.month - 1 if today.month > 1 else 12
|
||
last_year = today.year if today.month > 1 else today.year - 1
|
||
target_date = datetime.date(last_year, last_month, 1)
|
||
|
||
period_key = target_date.strftime('%Y-%m')
|
||
return _eng.ResetPeriod(
|
||
self.TenantUUID, metric_uuid, 'monthly', period_key,
|
||
)
|
||
|
||
def PurgeOld(self, before_date, period_type='daily'):
|
||
"""清理过期数据"""
|
||
return _eng.PurgeOld(self.TenantUUID, before_date, period_type)
|
||
|
||
def RunScheduledResets(self, target_date=None):
|
||
"""执行所有配置了自动重置的指标的周期清零
|
||
|
||
由 Celery Beat 定时任务调用。
|
||
"""
|
||
configs = StatConfig.objects.filter(
|
||
TenantUUID=self.TenantUUID,
|
||
AutoReset=True,
|
||
ConfigStatus=1,
|
||
)
|
||
if target_date is None:
|
||
target_date = datetime.date.today() - datetime.timedelta(days=1)
|
||
|
||
total_deleted = 0
|
||
for config in configs:
|
||
deleted = self.DailyReset(
|
||
metric_uuid=config.Metric_id,
|
||
target_date=target_date,
|
||
)
|
||
total_deleted += deleted
|
||
|
||
return total_deleted
|
||
|
||
def BuildRankings(self, target_date=None):
|
||
"""执行所有配置了排行榜的指标的排名生成
|
||
|
||
由 Celery Beat 定时任务调用。
|
||
"""
|
||
configs = StatConfig.objects.filter(
|
||
TenantUUID=self.TenantUUID,
|
||
EnableRanking=True,
|
||
ConfigStatus=1,
|
||
)
|
||
if target_date is None:
|
||
target_date = datetime.date.today()
|
||
|
||
total_rankings = 0
|
||
for config in configs:
|
||
rankings = self.GenerateDailyRanking(
|
||
metric_uuid=config.Metric_id,
|
||
ranking_date=target_date,
|
||
dimension_filter=config.RankingDimensionFilter or None,
|
||
top_n=config.RankingTopN,
|
||
)
|
||
total_rankings += len(rankings)
|
||
|
||
return total_rankings
|
||
|
||
def GetMetrics(self):
|
||
"""获取所有指标"""
|
||
return StatMetric.objects.filter(
|
||
TenantUUID=self.TenantUUID, MetricStatus=1,
|
||
)
|
||
|
||
def GetDimensions(self):
|
||
"""获取所有维度"""
|
||
return StatDimension.objects.filter(
|
||
TenantUUID=self.TenantUUID, DimensionStatus=1,
|
||
) |