diff --git a/SAAS_ARCHITECTURE.md b/SAAS_ARCHITECTURE.md index 44ba2124..3eb0a521 100644 --- a/SAAS_ARCHITECTURE.md +++ b/SAAS_ARCHITECTURE.md @@ -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 个月） diff --git a/docs/stats_engine.md b/docs/stats_engine.md index ef58a7f4..d1934540 100644 --- a/docs/stats_engine.md +++ b/docs/stats_engine.md @@ -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 不变 | \ No newline at end of file +| `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 访问 | \ No newline at end of file diff --git a/gvsds/apps/stats/manager.py b/gvsds/apps/stats/manager.py deleted file mode 100644 index 2c2aa1f8..00000000 --- a/gvsds/apps/stats/manager.py +++ /dev/null @@ -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, - ) \ No newline at end of file diff --git a/gvsds/apps/stats/tasks.py b/gvsds/apps/stats/tasks.py index a4163939..5e35d4ac 100644 --- a/gvsds/apps/stats/tasks.py +++ b/gvsds/apps/stats/tasks.py @@ -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(): diff --git a/gvsds/apps/stats/views.py b/gvsds/apps/stats/views.py index 4c990e2f..da0c258f 100644 --- a/gvsds/apps/stats/views.py +++ b/gvsds/apps/stats/views.py @@ -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 * * *'),