diff --git a/sage_datamart/init_data.py b/sage_datamart/init_data.py deleted file mode 100644 index 7301126..0000000 --- a/sage_datamart/init_data.py +++ /dev/null @@ -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, - }