GVSDS/SMANAGE
0
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

refactor(stats): 删除 StatEngine 统一走 gvsdsdk StatsManager，修复定时任务错误字段查询

Browse Source
This commit is contained in:
TermiNexus committed 2026-08-07 21:44:20 +08:00
1 parent 222fae6fbe
commit 68bcd5d26c
5 files changed
+167 -606
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

SAAS_ARCHITECTURE.md
+2 -2
View File
@@ -218,7 +218,7 @@ HTTP Request
| 中 | ABAC 实现本质是 RBAC（字符串集合） | 名实不符，扩展受限 |
| 中 | 数据权限硬编码 `if table_name == 'X'` | 违反开闭原则 |
| 中 | 缺乏软删除机制 | 误删不可恢复 |
| 中 | stats 双轨：`StatEngine`（gvsds/apps/stats/manager.py，22 方法）与 gvsdsdk `StatsManager` 并存 | 需统一收敛 |
| ~~中~~ | ~~stats 双轨~~：✅ 已统一（2026-08-07）——`StatEngine`（gvsds/apps/stats/manager.py）已删除，编排方法全部并入 gvsdsdk `StatsManager`，视图/Celery 任务统一走 SDK | — |
| 低 | UUID 主键无序导致索引膨胀 | 大数据量写入性能下降 |
---
@@ -231,7 +231,7 @@ HTTP Request
2. **补全 TenantUUID 字段**：所有租户语义表直接打 `TenantUUID`，简化跨租户查询与租户删除
3. **数据权限策略化**：用策略模式替代 `if table_name == 'X'`，每个模型注册自己的 `ScopeFilter`
4. ~~更新根 README.md~~：✅ 已反映当前 Django 架构
5. **统一 stats 双轨实现**：`StatEngine` 方法收敛到 gvsdsdk `StatsManager`
5. ~~统一 stats 双轨实现~~：✅ 已收敛——`StatEngine` 删除，`StatsManager` 补齐编排方法（DefineMetric/GenerateDailyRanking/QueryRanking/RunScheduledResets/BuildRankings 等），视图与 Celery 任务统一经 gvsdsdk 访问
### 7.2 中期（3-6 个月）
docs/stats_engine.md
+90 -66
View File
@@ -41,7 +41,7 @@
│ 充值到账 → stats.IncrementMulti([...])
│ 会员购买 → stats.Increment('vip_sale', {'level':'vip1'}, 1)
│
└──→ StatEngine
└──→ StatsManager（gvsdsdk）
│
├── 写: AtomicIncrement() → StatRecord（并发安全）
├── 读: QueryToday() / QueryRange() / AggregateTopN()
@@ -57,8 +57,8 @@
| 层级 | 文件 | 职责 |
|------|------|------|
| **模型** | `models.py` | 5 张表：Metric / Dimension / Record / Ranking / Config |
| **引擎** | `engine.py` | 7 个原子函数：AtomicIncrement / QueryPeriod / AggregateByDimension / GenerateRanking |
| **编排器** | `manager.py` | StatEngine — 15 个方法覆盖增删查改排清配 |
| **原子引擎** | `gvsdsdk/stats/manager.py` | StatsManager 底层：AtomicIncrement / QueryPeriod / AggregateByDimension / GenerateRanking 等 8 个原子方法 |
| **编排器** | `gvsdsdk/stats/manager.py` | StatsManager 编排：DefineMetric / ConfigureAutoReset / QueryRanking / RunScheduledResets / BuildRankings 等（原 StatEngine 已并入） |
| **API** | `views.py` | 10 个 REST 端点 |
| **序列化** | `serializers.py` | 14 个 Serializer |
| **路由** | `urls.py` | 10 个 URL 路由 |
@@ -169,42 +169,41 @@ def AtomicIncrement(tenant_uuid, period_type, period_key, metric_uuid,
---
## 七、StatEngine 编排器
## 七、StatsManager 编排器（gvsdsdk）
> 原 `StatEngine`（gvsds/apps/stats/manager.py）已于 2026-08-07 删除，编排能力全部并入 gvsdsdk `StatsManager` 静态方法，`tenant_uuid` 为首参。
```python
class StatEngine:
def __init__(self, tenant_uuid)
class StatsManager: # 全部为静态方法，tenant_uuid 作首参
# 指标管理
def DefineMetric(metric_code, metric_name, metric_type, unit) # 定义统计指标
def DefineDimension(dimension_code, dimension_name) # 定义统计维度
def GetMetrics() / GetDimensions() # 查询
def DefineMetric(tenant_uuid, metric_code, metric_name, metric_type, unit) # 定义统计指标
def DefineDimension(tenant_uuid, dimension_code, dimension_name) # 定义统计维度
def ListMetrics(tenant_uuid, status) / ListDimensions(tenant_uuid, status) # 查询
# 配置
def ConfigureAutoReset(metric_uuid, auto_reset, enable_ranking, ...) # 配置重置/排行
def ConfigureAutoReset(tenant_uuid, metric_uuid, auto_reset, enable_ranking, ...) # 配置重置/排行
# 数据写入
def Increment(metric_uuid, dimension_values, delta, date) # 原子增量
def IncrementMulti(increments, date) # 批量原子增量
def Decrement(metric_uuid, dimension_values, delta, date) # 原子减量
def AtomicIncrement(tenant_uuid, period_type, period_key, metric_uuid, dimension_values, delta) # 原子增量
def AtomicIncrementMulti(tenant_uuid, period_type, period_key, increments) # 批量原子增量
# 数据查询
def QueryToday(metric_uuid, dimension_filter) # 今日统计
def QueryDate(metric_uuid, date, period_type) # 指定日期
def QueryRange(metric_uuid, start_date, end_date) # 时间范围
def AggregateTopN(metric_uuid, start_date, end_date, dimension_key) # 聚合排名
def QueryPeriod(tenant_uuid, metric_uuid, period_type, period_key) # 特定周期
def QueryRange(tenant_uuid, metric_uuid, period_type, start_key, end_key) # 时间范围
def AggregateByDimension(tenant_uuid, metric_uuid, period_type, start_key, end_key, dimension_key) # 聚合排名
# 排行榜
def GenerateDailyRanking(metric_uuid, ranking_date, top_n) # 生成日榜
def GenerateMonthlyRanking(metric_uuid, ranking_date, top_n) # 生成月榜
def QueryRanking(ranking_period, ranking_date, metric_uuid) # 查询排行榜
def GenerateDailyRanking(tenant_uuid, metric_uuid, ranking_date, top_n) # 生成日榜
def GenerateMonthlyRanking(tenant_uuid, metric_uuid, ranking_date, top_n) # 生成月榜
def QueryRanking(tenant_uuid, ranking_period, ranking_date, metric_uuid) # 查询排行榜
# 维护
def DailyReset(metric_uuid, target_date) # 手动日清
def MonthlyReset(metric_uuid, target_date) # 手动月清
def RunScheduledResets(target_date) # 定时清零（Celery调用）
def BuildRankings(target_date) # 定时生成排行（Celery调用）
def PurgeOld(before_date, period_type) # 清理旧数据
def DailyReset(tenant_uuid, metric_uuid, target_date) # 手动日清
def MonthlyReset(tenant_uuid, metric_uuid, target_date) # 手动月清
def RunScheduledResets(tenant_uuid, target_date) # 定时清零（Celery调用）
def BuildRankings(tenant_uuid, target_date) # 定时生成排行（Celery调用）
def PurgeOld(tenant_uuid, before_date, period_type) # 清理旧数据
```
---
@@ -247,24 +246,25 @@ POST /api/stats/purge/ # 清理过期数据
### 9.1 初始化统计体系
```python
from gvsds.apps.stats.manager import StatEngine
from gvsdsdk.stats.manager import StatsManager
engine = StatEngine(tenant_uuid='al_long_001')
tenant = 'al_long_001'
# 定义指标
jiedan = engine.DefineMetric('dashou_jiedan', '打手接单数', 'count', '单')
shouru = engine.DefineMetric('dashou_shouru', '打手收入', 'amount', '元')
chongzhi = engine.DefineMetric('guanshi_chongzhi', '管事充值人数', 'count', '人')
vip = engine.DefineMetric('vip_sale', '会员销量', 'count', '份')
jiedan = StatsManager.DefineMetric(tenant, 'dashou_jiedan', '打手接单数', 'count', '单')
shouru = StatsManager.DefineMetric(tenant, 'dashou_shouru', '打手收入', 'amount', '元')
chongzhi = StatsManager.DefineMetric(tenant, 'guanshi_chongzhi', '管事充值人数', 'count', '人')
vip = StatsManager.DefineMetric(tenant, 'vip_sale', '会员销量', 'count', '份')
# 定义维度
engine.DefineDimension('role', '角色')
engine.DefineDimension('user_id', '用户ID')
engine.DefineDimension('platform', '平台')
engine.DefineDimension('order_type', '订单类型')
StatsManager.DefineDimension(tenant, 'role', '角色')
StatsManager.DefineDimension(tenant, 'user_id', '用户ID')
StatsManager.DefineDimension(tenant, 'platform', '平台')
StatsManager.DefineDimension(tenant, 'order_type', '订单类型')
# 配置每日清零 + 排行榜
engine.ConfigureAutoReset(
StatsManager.ConfigureAutoReset(
tenant_uuid=tenant,
metric_uuid=jiedan.MetricUUID,
auto_reset=True,
enable_ranking=True,
@@ -275,49 +275,66 @@ engine.ConfigureAutoReset(
### 9.2 订单完成时写统计
```python
import datetime
# 订单完成 → 写多条统计（原子批量）
engine.IncrementMulti(increments=[
(jiedan.MetricUUID, {'role': 'dashou', 'user_id': 'd_123'}, 1),
(jiedan.MetricUUID, {'role': 'dashou', 'user_id': 'd_123', 'order_type': 'platform'}, 1),
(jiedan.MetricUUID, {'role': 'guanshi', 'user_id': 'g_456'}, 1),
(shouru.MetricUUID, {'role': 'dashou', 'user_id': 'd_123'}, 80.00),
(shouru.MetricUUID, {'role': 'platform'}, 20.00),
])
StatsManager.AtomicIncrementMulti(
tenant_uuid=tenant,
period_type='daily',
period_key=datetime.date.today().strftime('%Y-%m-%d'),
increments=[
(jiedan.MetricUUID, {'role': 'dashou', 'user_id': 'd_123'}, 1),
(jiedan.MetricUUID, {'role': 'dashou', 'user_id': 'd_123', 'order_type': 'platform'}, 1),
(jiedan.MetricUUID, {'role': 'guanshi', 'user_id': 'g_456'}, 1),
(shouru.MetricUUID, {'role': 'dashou', 'user_id': 'd_123'}, 80.00),
(shouru.MetricUUID, {'role': 'platform'}, 20.00),
],
)
```
### 9.3 查询今日统计
```python
import datetime
# 查询打手123今日统计
records = engine.QueryToday(
records = StatsManager.QueryPeriod(
tenant_uuid=tenant,
metric_uuid=jiedan.MetricUUID,
period_type='daily',
period_key=datetime.date.today().strftime('%Y-%m-%d'),
dimension_filter={'user_id': 'd_123'},
)
# → [{period_key:'2026-06-11', dimension_values:{...}, value: 5}]
# 查询今日所有打手排名（实时聚合）
top = engine.AggregateTopN(
top = StatsManager.AggregateByDimension(
tenant_uuid=tenant,
metric_uuid=jiedan.MetricUUID,
start_date=datetime.date.today(),
end_date=datetime.date.today(),
period_type='daily',
start_key=datetime.date.today().strftime('%Y-%m-%d'),
end_key=datetime.date.today().strftime('%Y-%m-%d'),
dimension_key='user_id',
top_n=10,
)
)[:10]
# → [{dim_value:'d_123', total: 5}, {dim_value:'d_456', total: 3}, ...]
```
### 9.4 排行榜
```python
import datetime
# 生成今日排行榜快照（定时任务调用）
engine.GenerateDailyRanking(
StatsManager.GenerateDailyRanking(
tenant_uuid=tenant,
metric_uuid=jiedan.MetricUUID,
dimension_filter={'role': 'dashou'},
top_n=50,
)
# 前端查询日榜
rankings = engine.QueryRanking(
rankings = StatsManager.QueryRanking(
tenant_uuid=tenant,
ranking_period='daily',
ranking_date=datetime.date.today(),
metric_uuid=jiedan.MetricUUID,
@@ -328,33 +345,39 @@ rankings = engine.QueryRanking(
### 9.5 Celery 定时任务
```python
# celery.py 中注册每日任务
@app.task
# celery.py 中注册每日任务（实际实现见 gvsds/apps/stats/tasks.py）
@shared_task
def stats_daily_reset():
"""每日统计清零 + 排行榜生成"""
from gvsds.apps.stats.models import StatConfig
from gvsds.apps.stats.manager import StatEngine
from gvsds.apps.stats.tasks import _GetActiveTenants
from gvsdsdk.stats.manager import StatsManager
tenants = StatConfig.objects.filter(AutoReset=True).values_list('TenantUUID', flat=True).distinct()
for tenant in tenants:
engine = StatEngine(tenant)
engine.RunScheduledResets()
engine.BuildRankings()
for tenant in _GetActiveTenants():
StatsManager.RunScheduledResets(tenant)
StatsManager.BuildRankings(tenant)
```
### 9.6 退款回滚统计
```python
# 退款 → 回滚相关统计
engine.Decrement(
import datetime
# 退款 → 回滚相关统计（负增量）
StatsManager.AtomicIncrement(
tenant_uuid=tenant,
period_type='daily',
period_key=datetime.date.today().strftime('%Y-%m-%d'),
metric_uuid=jiedan.MetricUUID,
dimension_values={'role': 'dashou', 'user_id': 'd_123'},
delta=1,
delta=-1,
)
engine.Decrement(
StatsManager.AtomicIncrement(
tenant_uuid=tenant,
period_type='daily',
period_key=datetime.date.today().strftime('%Y-%m-%d'),
metric_uuid=shouru.MetricUUID,
dimension_values={'role': 'dashou', 'user_id': 'd_123'},
delta=80.00,
delta=-80.00,
)
```
@@ -381,4 +404,5 @@ engine.Decrement(
| `settings.py` | 新增 `'gvsds.apps.stats'` 到 `INSTALLED_APPS` |
| `db_router.py` | 新增 `'stats': 'default'` 到 `app_db_map` |
| `urls.py` | 新增 `path('api/stats/', include('gvsds.apps.stats.urls'))` |
| `gvsds/apps/stats/engine.py` | **已删除（2026-08-07）**：8 个模块级函数并入 gvsdsdk `StatsManager`；`StatEngine`（gvsds/apps/stats/manager.py）与 Celery 任务底层统一走 `gvsdsdk.stats.manager.StatsManager`，签名 1:1 不变 |
| `gvsds/apps/stats/engine.py` | **已删除（2026-08-07）**：8 个模块级函数并入 gvsdsdk `StatsManager` |
| `gvsds/apps/stats/manager.py` | **已删除（2026-08-07）**：`StatEngine` 编排方法（DefineMetric / ConfigureAutoReset / QueryRanking / RunScheduledResets / BuildRankings 等）并入 gvsdsdk `StatsManager`，视图与 Celery 任务统一经 gvsdsdk 访问 |
gvsds/apps/stats/manager.py
-475
View File
@@ -1,475 +0,0 @@
"""模块：统计编排器
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 gvsdsdk.stats.manager import StatsManager 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,
)
gvsds/apps/stats/tasks.py
+23 -28
View File
@@ -4,6 +4,8 @@
通过 gvsds/celery.py 的 beat_schedule 配置定时触发。
无 Celery 时可通过 management 命令或外部 cron 手动调用。
统一走 gvsdsdk StatsManager 编排方法，避免双轨实现。
"""
import logging
from datetime import date, timedelta
@@ -15,44 +17,41 @@ logger = logging.getLogger(__name__)
@shared_task(bind=True, name='stats.ResetDailyStats')
def ResetDailyStats(self):
"""每日统计清零任务——重置所有租户的 daily 周期数据。
"""每日统计清零任务——重置所有租户的 daily 周期数据（默认清零昨天）。
建议 beat 调度：每日 00:00 执行。
"""
from gvsds.apps.stats.models import StatMetric
from gvsdsdk.stats.manager import StatsManager
today = date.today().strftime('%Y-%m-%d')
total = 0
for metric in StatMetric.objects.filter(PeriodType='daily', Status=1):
for tenant_uuid in _GetActiveTenants():
try:
total += StatsManager.ResetPeriod(tenant_uuid, metric.MetricUUID, 'daily', today)
except Exception:
logger.warning('[StatsTasks] ResetDaily 失败: tenant=%s metric=%s', tenant_uuid, metric.MetricUUID, exc_info=True)
for tenant_uuid in _GetActiveTenants():
try:
total += StatsManager.RunScheduledResets(tenant_uuid)
except Exception:
logger.warning('[StatsTasks] ResetDaily 失败: tenant=%s', tenant_uuid, exc_info=True)
logger.info('[StatsTasks] ResetDailyStats 完成，重置 %d 条', total)
return {'reset_count': total, 'period_key': today}
return {'reset_count': total}
@shared_task(bind=True, name='stats.ResetMonthlyStats')
def ResetMonthlyStats(self):
"""每月统计清零任务——重置所有租户的 monthly 周期数据。
"""每月统计清零任务——重置所有租户的 monthly 周期数据（默认清零上月）。
建议 beat 调度：每月 1 日 00:00 执行。
"""
from gvsds.apps.stats.models import StatMetric
from gvsdsdk.stats.manager import StatsManager
from gvsdsdk.stats.models import StatConfig
month_key = date.today().strftime('%Y-%m')
total = 0
for metric in StatMetric.objects.filter(PeriodType='monthly', Status=1):
for tenant_uuid in _GetActiveTenants():
for tenant_uuid in _GetActiveTenants():
for config in StatConfig.objects.filter(
TenantUUID=tenant_uuid, AutoReset=True, ConfigStatus=1):
try:
total += StatsManager.ResetPeriod(tenant_uuid, metric.MetricUUID, 'monthly', month_key)
total += StatsManager.MonthlyReset(tenant_uuid, config.Metric_id)
except Exception:
logger.warning('[StatsTasks] ResetMonthly 失败: tenant=%s metric=%s', tenant_uuid, metric.MetricUUID, exc_info=True)
logger.warning('[StatsTasks] ResetMonthly 失败: tenant=%s metric=%s', tenant_uuid, config.Metric_id, exc_info=True)
logger.info('[StatsTasks] ResetMonthlyStats 完成，重置 %d 条', total)
return {'reset_count': total, 'period_key': month_key}
return {'reset_count': total}
@shared_task(bind=True, name='stats.PurgeOldStats')
@@ -80,20 +79,16 @@ def BuildDailyRankings(self):
建议 beat 调度：每日 23:50 执行（当日数据基本定型）。
"""
from gvsds.apps.stats.models import StatMetric
from gvsdsdk.stats.manager import StatsManager
today = date.today()
total = 0
for metric in StatMetric.objects.filter(EnableRanking=True, Status=1):
for tenant_uuid in _GetActiveTenants():
try:
StatsManager.GenerateRanking(tenant_uuid, metric.MetricUUID, 'daily', today)
total += 1
except Exception:
logger.warning('[StatsTasks] BuildRanking 失败: tenant=%s metric=%s', tenant_uuid, metric.MetricUUID, exc_info=True)
for tenant_uuid in _GetActiveTenants():
try:
total += StatsManager.BuildRankings(tenant_uuid)
except Exception:
logger.warning('[StatsTasks] BuildRanking 失败: tenant=%s', tenant_uuid, exc_info=True)
logger.info('[StatsTasks] BuildDailyRankings 完成，生成 %d 个排行', total)
return {'ranking_count': total, 'date': today.isoformat()}
return {'ranking_count': total}
def _GetActiveTenants():
gvsds/apps/stats/views.py
+52 -35
View File
@@ -1,8 +1,6 @@
import logging
from rest_framework.views import APIView
from rest_framework.response import Response
from gvsds.apps.stats.manager import StatEngine
from gvsds.apps.stats.models import StatMetric, StatDimension
from gvsdsdk import StatsManager
from gvsds.apps.stats.serializers import (
StatMetricSerializer,
@@ -43,7 +41,6 @@ class MetricConfigView(APIView):
tenant = _get_tenant(request)
if not tenant:
return Response(_err('缺少租户标识', 401), status=401)
from gvsdsdk.stats.manager import StatsManager
metrics = StatsManager.ListMetrics(tenant, status=1)
return Response(_ok(StatMetricSerializer(metrics, many=True).data))
@@ -54,8 +51,8 @@ class MetricConfigView(APIView):
ser = MetricDefineSerializer(data=request.data)
if not ser.IsValid():
return Response(_err(ser.errors, 400), status=400)
engine = StatEngine(tenant)
metric = engine.DefineMetric(
metric = StatsManager.DefineMetric(
tenant_uuid=tenant,
metric_code=ser.ValidatedData['MetricCode'],
metric_name=ser.ValidatedData['MetricName'],
metric_type=ser.ValidatedData.get('MetricType', 'count'),
@@ -71,7 +68,6 @@ class DimensionView(APIView):
tenant = _get_tenant(request)
if not tenant:
return Response(_err('缺少租户标识', 401), status=401)
from gvsdsdk.stats.manager import StatsManager
dims = StatsManager.ListDimensions(tenant, status=1)
return Response(_ok(StatDimensionSerializer(dims, many=True).data))
@@ -82,8 +78,8 @@ class DimensionView(APIView):
ser = DimensionDefineSerializer(data=request.data)
if not ser.IsValid():
return Response(_err(ser.errors, 400), status=400)
engine = StatEngine(tenant)
dim = engine.DefineDimension(
dim = StatsManager.DefineDimension(
tenant_uuid=tenant,
dimension_code=ser.ValidatedData['DimensionCode'],
dimension_name=ser.ValidatedData['DimensionName'],
)
@@ -99,13 +95,24 @@ class IncrementView(APIView):
ser = IncrementSerializer(data=request.data)
if not ser.IsValid():
return Response(_err(ser.errors, 400), status=400)
engine = StatEngine(tenant)
record = engine.Increment(
import datetime as _dt
_date = ser.ValidatedData.get('Date') or _dt.date.today()
_period_type = ser.ValidatedData.get('PeriodType', 'daily')
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 = StatsManager.AtomicIncrement(
tenant_uuid=tenant,
period_type=_period_type,
period_key=_period_key,
metric_uuid=ser.ValidatedData['MetricUUID'],
dimension_values=ser.ValidatedData['DimensionValues'],
delta=ser.ValidatedData.get('Delta', 1),
date=ser.ValidatedData.get('Date'),
period_type=ser.ValidatedData.get('PeriodType', 'daily'),
)
return Response(_ok(StatRecordSerializer(record).data), status=201)
@@ -119,10 +126,13 @@ class IncrementMultiView(APIView):
ser = IncrementMultiSerializer(data=request.data)
if not ser.IsValid():
return Response(_err(ser.errors, 400), status=400)
engine = StatEngine(tenant)
records = engine.IncrementMulti(
import datetime as _dt
_date = ser.ValidatedData.get('Date') or _dt.date.today()
records = StatsManager.AtomicIncrementMulti(
tenant_uuid=tenant,
period_type='daily',
period_key=_date.strftime('%Y-%m-%d'),
increments=ser.ValidatedData['Increments'],
date=ser.ValidatedData.get('Date'),
)
return Response(_ok(StatRecordSerializer(records, many=True).data), status=201)
@@ -139,19 +149,25 @@ class QueryView(APIView):
EndDate = request.query_params.get('EndDate')
PeriodType = request.query_params.get('PeriodType', 'daily')
engine = StatEngine(tenant)
if StartDate and EndDate:
import datetime
sd = datetime.date.fromisoformat(StartDate)
ed = datetime.date.fromisoformat(EndDate)
records = engine.QueryRange(
records = StatsManager.QueryRange(
tenant_uuid=tenant,
metric_uuid=MetricUUID,
start_date=sd,
end_date=ed,
period_type=PeriodType,
start_key=sd.strftime('%Y-%m-%d'),
end_key=ed.strftime('%Y-%m-%d'),
)
elif MetricUUID:
records = engine.QueryToday(metric_uuid=MetricUUID)
import datetime
records = StatsManager.QueryPeriod(
tenant_uuid=tenant,
metric_uuid=MetricUUID,
period_type='daily',
period_key=datetime.date.today().strftime('%Y-%m-%d'),
)
else:
return Response(_err('缺少查询参数', 400), status=400)
@@ -175,15 +191,17 @@ class AggregateView(APIView):
return Response(_err('缺少必要参数', 400), status=400)
import datetime
engine = StatEngine(tenant)
results = engine.AggregateTopN(
sd = datetime.date.fromisoformat(StartDate)
ed = datetime.date.fromisoformat(EndDate)
results = StatsManager.AggregateByDimension(
tenant_uuid=tenant,
metric_uuid=MetricUUID,
start_date=datetime.date.fromisoformat(StartDate),
end_date=datetime.date.fromisoformat(EndDate),
period_type='daily',
start_key=sd.strftime('%Y-%m-%d'),
end_key=ed.strftime('%Y-%m-%d'),
dimension_key=DimensionKey,
top_n=TopN,
)
return Response(_ok(results))
return Response(_ok(results[:TopN]))
class RankingView(APIView):
@@ -199,9 +217,9 @@ class RankingView(APIView):
TopN = request.query_params.get('TopN')
import datetime
engine = StatEngine(tenant)
rd = datetime.date.fromisoformat(RankingDate) if RankingDate else None
rankings = engine.QueryRanking(
rankings = StatsManager.QueryRanking(
tenant_uuid=tenant,
ranking_period=RankingPeriod,
ranking_date=rd,
metric_uuid=MetricUUID or None,
@@ -216,8 +234,8 @@ class RankingView(APIView):
ser = RankingGenerateSerializer(data=request.data)
if not ser.IsValid():
return Response(_err(ser.errors, 400), status=400)
engine = StatEngine(tenant)
rankings = engine.GenerateDailyRanking(
rankings = StatsManager.GenerateDailyRanking(
tenant_uuid=tenant,
metric_uuid=ser.ValidatedData['MetricUUID'],
ranking_date=ser.ValidatedData.get('RankingDate'),
dimension_filter=ser.ValidatedData.get('DimensionFilter'),
@@ -235,8 +253,8 @@ class ResetView(APIView):
ser = ResetSerializer(data=request.data)
if not ser.IsValid():
return Response(_err(ser.errors, 400), status=400)
engine = StatEngine(tenant)
deleted = engine.DailyReset(
deleted = StatsManager.DailyReset(
tenant_uuid=tenant,
metric_uuid=ser.ValidatedData['MetricUUID'],
target_date=ser.ValidatedData.get('TargetDate'),
)
@@ -266,7 +284,6 @@ class StatConfigView(APIView):
tenant = _get_tenant(request)
if not tenant:
return Response(_err('缺少租户标识', 401), status=401)
from gvsdsdk.stats.manager import StatsManager
configs = StatsManager.ListConfigs(tenant, status=1)
return Response(_ok(StatConfigSerializer(configs, many=True).data))
@@ -277,8 +294,8 @@ class StatConfigView(APIView):
ser = ConfigSerializer(data=request.data)
if not ser.IsValid():
return Response(_err(ser.errors, 400), status=400)
engine = StatEngine(tenant)
config = engine.ConfigureAutoReset(
config = StatsManager.ConfigureAutoReset(
tenant_uuid=tenant,
metric_uuid=ser.ValidatedData['MetricUUID'],
auto_reset=ser.ValidatedData.get('AutoReset', True),
reset_cron=ser.ValidatedData.get('ResetCron', '0 0 * * *'),