fix: dm_model_perf_daily.catelogid 10→32 + GROUP BY userorgid加表前缀 + 独立聚合脚本aggregate_all.py

models/dm_model_perf_daily.json

@@ -38,7 +38,7 @@
             "name": "catelogid",
             "title": "目录ID",
             "type": "str",
-            "length": 10
+            "length": 32
         },
         {
             "name": "providerid",

sage_datamart/etl.py

@@ -178,7 +178,7 @@ async def aggregate_daily_perf(sor, stat_date=None):
             SUM(amount) as total_amount
         FROM dm_model_call_fact
         WHERE call_date = ${stat_date}$
-        GROUP BY call_date, llmid, model, catelogid, providerid, ownerid, userorgid, dm_model_call_fact.currency
+        GROUP BY call_date, llmid, model, catelogid, providerid, ownerid, dm_model_call_fact.userorgid, dm_model_call_fact.currency
     """
     rows = await sor.sqlExe(sql, {'stat_date': stat_date})
     info('sage_datamart AGG: aggregating ' + str(len(rows)) + ' groups for ' + stat_date)

scripts/aggregate_all.py

@@ -0,0 +1,61 @@
+#!/usr/bin/env python3
+"""
+数据集市聚合脚本 — 从 dm_model_call_fact 重算所有历史日期的聚合表
+独立于 init_data.py，可单独执行。
+
+用法: cd /path/to/sage && python /path/to/sage_datamart/scripts/aggregate_all.py
+"""
+import asyncio
+from datetime import datetime
+from appPublic.jsonConfig import getConfig
+from appPublic.uniqueID import getID
+from appPublic.dictObject import DictObject
+from sqlor.dbpools import DBPools
+
+DB_NAME = 'sage'
+
+
+async def batch_aggregate_history(sor):
+    """按 dm_model_call_fact 中所有历史日期重算聚合表"""
+    from sage_datamart.etl import aggregate_daily_perf, aggregate_provider_cost
+
+    rows = await sor.sqlExe(
+        'SELECT DISTINCT call_date FROM dm_model_call_fact ORDER BY call_date', {}
+    )
+    dates = []
+    for r in rows:
+        if hasattr(r, 'call_date'):
+            d = r.call_date
+        else:
+            d = r.get('call_date', r[0])
+        dates.append(d.strftime('%Y-%m-%d') if hasattr(d, 'strftime') else str(d))
+
+    print('Aggregating ' + str(len(dates)) + ' distinct dates...')
+    ok = fail = 0
+    for i, d in enumerate(dates):
+        try:
+            await aggregate_daily_perf(sor, stat_date=d)
+            await aggregate_provider_cost(sor, stat_date=d)
+            ok += 1
+        except Exception as e:
+            fail += 1
+            print('  [' + d + '] FAILED: ' + str(e)[:120])
+        if (i + 1) % 10 == 0:
+            print('  progress: ' + str(i + 1) + '/' + str(len(dates)))
+
+    print('Done: ' + str(ok) + ' ok, ' + str(fail) + ' failed')
+    return ok, fail
+
+
+async def main():
+    config = getConfig('.')
+    db = DBPools()
+    db.databases = config.databases
+    async with db.sqlorContext(DB_NAME) as sor:
+        ok, fail = await batch_aggregate_history(sor)
+    print('\n=== Aggregation Complete ===')
+    print('ok=' + str(ok) + ' fail=' + str(fail))
+
+
+if __name__ == '__main__':
+    asyncio.run(main())