cleanup: remove obsolete sage_datamart/init_data.py (moved to scripts/init_data.py)

sage_datamart/init_data.py

@@ -1,124 +0,0 @@
-"""
-数据集市初始化 — 从 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,
-    }