feat: add init_data script to import historical data from llmusage + llmusage_history into data mart

- init_data.py: batch import with dedup, aggregate all historical dates
- wwwroot/cron/init_data.dspy: trigger endpoint
- init.py: register init_datamart function with IP whitelist
- load_path.py: add RBAC permission for init_data.dspy

sage_datamart/init.py

@@ -3,6 +3,7 @@ from ahserver.serverenv import ServerEnv
 from appPublic.log import debug
 from sqlor.dbpools import get_sor_context
 from .etl import sync_call_fact, aggregate_daily_perf, aggregate_provider_cost, ensure_tables
+from .init_data import run_init as _run_init
 from .dashboards import (
     get_today_usage, get_today_amount, get_success_rate, get_fail_count,
     get_active_users, get_top_models, get_top_providers, get_top_users,
@@ -185,6 +186,17 @@ async def cron_etl_provider_cost(request):
         return {'status': 'ok', 'aggregated': count}
 
 
+async def init_datamart(request):
+    """初始化数据集市：从 llmusage + llmusage_history 导入全量历史"""
+    ip = request['client_ip']
+    if ip not in ['127.0.0.1']:
+        return {'error': f'IP {ip} not allowed'}
+    env = request._run_ns
+    async with get_sor_context(env, MODULE_NAME) as sor:
+        result = await _run_init(sor)
+        return result
+
+
 def load_sage_datamart():
     env = ServerEnv()
     env.api_model_perf = api_model_perf
@@ -205,4 +217,5 @@ def load_sage_datamart():
     env.cron_etl_sync = cron_etl_sync
     env.cron_etl_aggregate = cron_etl_aggregate
     env.cron_etl_provider_cost = cron_etl_provider_cost
+    env.init_datamart = init_datamart
     debug("[sage_datamart] registered all API endpoints and cron functions")

sage_datamart/init_data.py

@@ -0,0 +1,124 @@
+"""
+数据集市初始化 — 从 llmusage + llmusage_history 导入全量历史数据到 dm_*
+"""
+import json
+from datetime import datetime
+from appPublic.uniqueID import getID
+from appPublic.dictObject import DictObject
+from appPublic.log import info
+from sage_datamart.etl import (
+    MODULE_NAME, ensure_tables, parse_usages,
+    aggregate_daily_perf, aggregate_provider_cost
+)
+
+
+async def batch_sync_call_fact(sor, source_table, batch_size=1000):
+    """
+    批量从 llmusage / llmusage_history 导入 dm_model_call_fact
+    自动去重（已存在 luid 跳过）
+    """
+    offset = 0
+    total_inserted = 0
+    total_skipped = 0
+
+    while True:
+        sql = (
+            'SELECT lu.*, l.model as llm_model, l.catelogid, l.providerid, l.ownerid '
+            'FROM ' + source_table + ' lu '
+            'JOIN llm l ON l.id = lu.llmid '
+            'ORDER BY lu.use_time '
+            'LIMIT ${limit}$ OFFSET ${offset}$'
+        )
+        rows = await sor.sqlExe(sql, {'limit': batch_size, 'offset': offset})
+        if not rows:
+            break
+
+        info('init_data: ' + source_table + ' batch offset=' + str(offset) + ' rows=' + str(len(rows)))
+
+        for r in rows:
+            r = DictObject(r)
+            existing = await sor.sqlExe(
+                'SELECT id FROM dm_model_call_fact WHERE luid=${luid}$',
+                {'luid': r.id}
+            )
+            if existing:
+                total_skipped += 1
+                continue
+
+            prompt_tokens, completion_tokens, _ = parse_usages(r.usages)
+            try:
+                call_time = datetime.strptime(str(r.use_time)[:19], '%Y-%m-%d %H:%M:%S')
+            except Exception:
+                total_skipped += 1
+                continue
+
+            ns = {
+                'id': getID(),
+                'luid': r.id,
+                'llmid': r.llmid,
+                'model': r.llm_model or r.model,
+                'catelogid': r.catelogid,
+                'userid': r.userid,
+                'userorgid': r.userorgid,
+                'ownerid': r.ownerid,
+                'providerid': r.providerid,
+                'call_date': r.use_date or call_time.strftime('%Y-%m-%d'),
+                'call_hour': call_time.hour,
+                'call_time': r.use_time,
+                'ttft_ms': int(r.responsed_seconds * 1000) if r.responsed_seconds else None,
+                'ttot_ms': int(r.finish_seconds * 1000) if r.finish_seconds else None,
+                'prompt_tokens': prompt_tokens,
+                'completion_tokens': completion_tokens,
+                'status': r.status or 'SUCCEEDED',
+                'amount': r.amount or 0,
+                'currency': getattr(r, 'currency', 'CNY') or 'CNY',
+            }
+            await sor.C('dm_model_call_fact', ns)
+            total_inserted += 1
+
+        offset += batch_size
+
+    info('init_data: ' + source_table + ' done inserted=' + str(total_inserted) + ' skipped=' + str(total_skipped))
+    return total_inserted, total_skipped
+
+
+async def batch_aggregate_history(sor):
+    """按 dm_model_call_fact 中所有历史日期重算 dm_model_perf_daily / dm_provider_cost_daily"""
+    rows = await sor.sqlExe(
+        'SELECT DISTINCT call_date FROM dm_model_call_fact ORDER BY call_date',
+        {}
+    )
+    dates = [r['call_date'].strftime('%Y-%m-%d') if hasattr(r['call_date'], 'strftime') else str(r['call_date']) for r in rows]
+    info('init_data: aggregating ' + str(len(dates)) + ' distinct dates')
+
+    for d in dates:
+        await aggregate_daily_perf(sor, stat_date=d)
+        await aggregate_provider_cost(sor, stat_date=d)
+
+    return len(dates)
+
+
+async def run_init(sor):
+    """
+    初始化入口：
+      1. 建表（如不存在）
+      2. 从 llmusage_history 批量导入历史
+      3. 从 llmusage 导入当前数据
+      4. 重算所有日期的聚合表
+    """
+    await ensure_tables(sor)
+
+    info('init_data: step 1/4 importing llmusage_history')
+    ins_hist, skip_hist = await batch_sync_call_fact(sor, 'llmusage_history')
+
+    info('init_data: step 2/4 importing llmusage')
+    ins_curr, skip_curr = await batch_sync_call_fact(sor, 'llmusage')
+
+    info('init_data: step 3/4 aggregating historical dates')
+    dates = await batch_aggregate_history(sor)
+
+    return {
+        'llmusage_history': {'inserted': ins_hist, 'skipped': skip_hist},
+        'llmusage':         {'inserted': ins_curr, 'skipped': skip_curr},
+        'aggregated_dates': dates,
+    }

scripts/load_path.py

@@ -12,6 +12,7 @@ PATHS_ANY = [
     f"/{MOD}/cron/etl_sync.dspy",
     f"/{MOD}/cron/etl_aggregate.dspy",
     f"/{MOD}/cron/etl_provider_cost.dspy",
+    f"/{MOD}/cron/init_data.dspy",
 ]
 
 PATHS_LOGINED = [

wwwroot/cron/init_data.dspy

@@ -0,0 +1,3 @@
+# 数据集市初始化：从 llmusage + llmusage_history 导入全量历史数据
+result = await request._run_ns.init_datamart(request)
+return str(result)