From 7f5c1ed372f4f484f40698087b4661c646f26464 Mon Sep 17 00:00:00 2001 From: yumoqing Date: Mon, 20 Jul 2026 13:08:04 +0800 Subject: [PATCH] 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 | 13 ++++ sage_datamart/init_data.py | 124 ++++++++++++++++++++++++++++++++++++ scripts/load_path.py | 1 + wwwroot/cron/init_data.dspy | 3 + 4 files changed, 141 insertions(+) create mode 100644 sage_datamart/init_data.py create mode 100644 wwwroot/cron/init_data.dspy diff --git a/sage_datamart/init.py b/sage_datamart/init.py index a3825e7..84186ee 100644 --- a/sage_datamart/init.py +++ b/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") diff --git a/sage_datamart/init_data.py b/sage_datamart/init_data.py new file mode 100644 index 0000000..7301126 --- /dev/null +++ b/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, + } diff --git a/scripts/load_path.py b/scripts/load_path.py index 368bfe5..72076e8 100644 --- a/scripts/load_path.py +++ b/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 = [ diff --git a/wwwroot/cron/init_data.dspy b/wwwroot/cron/init_data.dspy new file mode 100644 index 0000000..51f965f --- /dev/null +++ b/wwwroot/cron/init_data.dspy @@ -0,0 +1,3 @@ +# 数据集市初始化：从 llmusage + llmusage_history 导入全量历史数据 +result = await request._run_ns.init_datamart(request) +return str(result)