284 lines
11 KiB
Python
284 lines
11 KiB
Python
"""
|
||
数据集市 ETL — 增量同步 llmusage → 集市表
|
||
建表由 models/*.json + json2ddl + build.sh 部署时完成，ETL 只负责数据读写。
|
||
"""
|
||
import json
|
||
from datetime import datetime, timedelta
|
||
from appPublic.uniqueID import getID
|
||
from appPublic.dictObject import DictObject
|
||
from appPublic.log import info
|
||
|
||
MODULE_NAME = "sage_datamart"
|
||
|
||
|
||
def parse_usages(usages_str):
|
||
"""从 llmusage.usages JSON 中提取 token 数"""
|
||
try:
|
||
u = json.loads(usages_str) if isinstance(usages_str, str) else usages_str
|
||
if isinstance(u, dict):
|
||
return (
|
||
u.get('prompt_tokens', 0) or 0,
|
||
u.get('completion_tokens', 0) or 0,
|
||
u.get('total_tokens', 0) or 0
|
||
)
|
||
except:
|
||
pass
|
||
return 0, 0, 0
|
||
|
||
|
||
async def sync_call_fact(sor, last_sync=None):
|
||
"""增量同步 llmusage → dm_model_call_fact（含币种折算）
|
||
|
||
金额优先从 product_usage_log.sell_price 取实际计费金额，
|
||
取不到时 fallback 到 llmusage.amount。
|
||
币种从 llmusage.amount_currency 取，按 exchange_rate 折算 CNY。
|
||
"""
|
||
if last_sync is None:
|
||
last_sync = (datetime.now() - timedelta(minutes=10)).strftime('%Y-%m-%d %H:%M:%S')
|
||
|
||
# 预取汇率
|
||
stat_date = datetime.now().strftime('%Y-%m-%d')
|
||
rates = await _get_exchange_rates(sor, stat_date)
|
||
|
||
# 预取 product_usage_log 实际计费金额（按 source_ref_table='llmusage' 关联）
|
||
usage_log_map = await _load_usage_log_bills(sor, last_sync)
|
||
|
||
sql = """
|
||
SELECT lu.*, l.model as llm_model, l.providerid,
|
||
(SELECT llmcatelogid FROM llm_api_map
|
||
WHERE llmid = lu.llmid AND isdefaultcatelog = '1' LIMIT 1) as catelogid
|
||
FROM llmusage lu
|
||
JOIN llm l ON l.id = lu.llmid
|
||
WHERE lu.use_time >= ${last_sync}$
|
||
ORDER BY lu.use_time
|
||
"""
|
||
rows = await sor.sqlExe(sql, {'last_sync': last_sync})
|
||
info('sage_datamart ETL: syncing ' + str(len(rows)) + ' rows from ' + last_sync)
|
||
|
||
count = 0
|
||
for r in rows:
|
||
r = DictObject(r)
|
||
# Skip existing
|
||
existing = await sor.sqlExe(
|
||
'SELECT id FROM dm_model_call_fact WHERE luid=${luid}$',
|
||
{'luid': r.id})
|
||
if existing:
|
||
continue
|
||
|
||
prompt_tokens, completion_tokens, _ = parse_usages(r.usages)
|
||
call_time = datetime.strptime(r.use_time[:19], '%Y-%m-%d %H:%M:%S')
|
||
currency = getattr(r, 'amount_currency', None) or 'CNY'
|
||
|
||
# 优先取 product_usage_log.sell_price（实际计费金额），fallback 到 llmusage.amount
|
||
bill_data = usage_log_map.get(r.id)
|
||
if bill_data:
|
||
amount = bill_data.get('sell_price', 0) or 0
|
||
# product_usage_log 的币种来自 product_subscription.purchase_currency
|
||
bill_currency = bill_data.get('purchase_currency')
|
||
if bill_currency:
|
||
currency = bill_currency
|
||
else:
|
||
amount = r.amount or 0
|
||
|
||
rate = rates.get(currency, 1.0)
|
||
amount_cny = round(float(amount) * rate, 4)
|
||
|
||
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': amount,
|
||
'currency': currency,
|
||
'amount_cny': amount_cny,
|
||
}
|
||
await sor.C('dm_model_call_fact', ns)
|
||
count += 1
|
||
|
||
info('sage_datamart ETL: inserted ' + str(count) + ' new rows')
|
||
return count
|
||
|
||
|
||
async def _load_usage_log_bills(sor, last_sync):
|
||
"""从 product_usage_log 预取实际计费金额"""
|
||
usage_log_map = {}
|
||
try:
|
||
sql = """
|
||
SELECT pul.source_ref_id, pul.sell_price, ps.purchase_currency
|
||
FROM product_usage_log pul
|
||
LEFT JOIN product_subscription ps ON ps.id = pul.subscription_id
|
||
WHERE pul.source_ref_table = 'llmusage'
|
||
AND pul.use_time >= ${last_sync}$
|
||
"""
|
||
rows = await sor.sqlExe(sql, {'last_sync': last_sync})
|
||
for r in rows:
|
||
ref_id = r.source_ref_id if hasattr(r, 'source_ref_id') else r['source_ref_id']
|
||
sell_price = r.sell_price if hasattr(r, 'sell_price') else r['sell_price']
|
||
purchase_currency = r.purchase_currency if hasattr(r, 'purchase_currency') else r.get('purchase_currency')
|
||
if ref_id:
|
||
usage_log_map[ref_id] = {
|
||
'sell_price': float(sell_price) if sell_price else 0.0,
|
||
'purchase_currency': purchase_currency,
|
||
}
|
||
except Exception as e:
|
||
info('sage_datamart: product_usage_log lookup failed: ' + str(e))
|
||
return usage_log_map
|
||
|
||
|
||
async def _get_exchange_rates(sor, stat_date):
|
||
"""查询汇率表"""
|
||
rates = {'CNY': 1.0}
|
||
try:
|
||
sql = """SELECT from_currency, mid_rate FROM exchange_rate
|
||
WHERE to_currency = 'CNY' AND effective_date <= ${d}$
|
||
ORDER BY effective_date DESC"""
|
||
rows = await sor.sqlExe(sql, {'d': stat_date})
|
||
for r in rows:
|
||
cur = r.from_currency if hasattr(r, 'from_currency') else r['from_currency']
|
||
rate = r.mid_rate if hasattr(r, 'mid_rate') else r['mid_rate']
|
||
if cur not in rates and rate:
|
||
rates[cur] = float(rate)
|
||
except Exception as e:
|
||
info('sage_datamart: exchange_rate lookup failed: ' + str(e))
|
||
return rates
|
||
|
||
|
||
async def aggregate_daily_perf(sor, stat_date=None):
|
||
"""聚合当天 dm_model_call_fact → dm_model_perf_daily（按币种分组）"""
|
||
if stat_date is None:
|
||
stat_date = datetime.now().strftime('%Y-%m-%d')
|
||
|
||
rates = await _get_exchange_rates(sor, stat_date)
|
||
|
||
sql = """
|
||
SELECT
|
||
call_date, llmid, model, catelogid, providerid, ownerid,
|
||
COALESCE(userorgid, '') as userorgid,
|
||
COALESCE(currency, 'CNY') as currency,
|
||
COUNT(*) as total_calls,
|
||
SUM(CASE WHEN status='SUCCEEDED' THEN 1 ELSE 0 END) as success_calls,
|
||
SUM(CASE WHEN status='FAILED' THEN 1 ELSE 0 END) as fail_calls,
|
||
AVG(ttft_ms) as avg_ttft,
|
||
AVG(ttot_ms) as avg_ttot,
|
||
SUM(prompt_tokens) as total_prompt,
|
||
SUM(completion_tokens) as total_completion,
|
||
SUM(amount) as total_amount
|
||
FROM dm_model_call_fact
|
||
WHERE call_date = ${stat_date}$
|
||
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)
|
||
|
||
for r in rows:
|
||
r = DictObject(r)
|
||
cur = r.currency or 'CNY'
|
||
rate = rates.get(cur, 1.0)
|
||
amount_cny = round(float(r.total_amount or 0) * rate, 4)
|
||
|
||
await sor.sqlExe(
|
||
'DELETE FROM dm_model_perf_daily WHERE stat_date=${d}$ AND llmid=${l}$ AND COALESCE(currency,"CNY")=${c}$ AND COALESCE(userorgid,"")=${u}$',
|
||
{'d': stat_date, 'l': r.llmid, 'c': cur, 'u': r.userorgid or ''})
|
||
|
||
ns = {
|
||
'id': getID(),
|
||
'stat_date': stat_date,
|
||
'llmid': r.llmid,
|
||
'model': r.model,
|
||
'catelogid': r.catelogid,
|
||
'providerid': r.providerid,
|
||
'ownerid': r.ownerid,
|
||
'total_calls': r.total_calls,
|
||
'success_calls': r.success_calls,
|
||
'fail_calls': r.fail_calls,
|
||
'success_rate': round(r.success_calls * 100.0 / r.total_calls, 2) if r.total_calls else 0,
|
||
'avg_ttft_ms': round(r.avg_ttft, 1) if r.avg_ttft else None,
|
||
'avg_ttot_ms': round(r.avg_ttot, 1) if r.avg_ttot else None,
|
||
'total_prompt_tokens': r.total_prompt or 0,
|
||
'total_completion_tokens': r.total_completion or 0,
|
||
'total_amount': r.total_amount or 0,
|
||
'currency': cur,
|
||
'total_amount_cny': amount_cny,
|
||
'userorgid': r.userorgid or None,
|
||
}
|
||
await sor.C('dm_model_perf_daily', ns)
|
||
|
||
return len(rows)
|
||
|
||
|
||
async def aggregate_provider_cost(sor, stat_date=None):
|
||
"""聚合同模型供应商性价比 — dm_provider_cost_daily（按币种分组）"""
|
||
if stat_date is None:
|
||
stat_date = (datetime.now() - timedelta(days=1)).strftime('%Y-%m-%d')
|
||
|
||
rates = await _get_exchange_rates(sor, stat_date)
|
||
|
||
sql = """
|
||
SELECT
|
||
model, providerid,
|
||
COALESCE(currency, 'CNY') as currency,
|
||
COUNT(*) as total_calls,
|
||
SUM(prompt_tokens + completion_tokens) as total_tokens,
|
||
SUM(amount) as total_amount,
|
||
AVG(ttft_ms) as avg_ttft
|
||
FROM dm_model_call_fact
|
||
WHERE call_date = ${stat_date}$
|
||
GROUP BY model, providerid, dm_model_call_fact.currency
|
||
"""
|
||
rows = await sor.sqlExe(sql, {'stat_date': stat_date})
|
||
|
||
for r in rows:
|
||
r = DictObject(r)
|
||
cur = r.currency or 'CNY'
|
||
rate = rates.get(cur, 1.0)
|
||
amount_cny = round(float(r.total_amount or 0) * rate, 4)
|
||
|
||
await sor.sqlExe(
|
||
'DELETE FROM dm_provider_cost_daily WHERE stat_date=${d}$ AND model=${m}$ AND providerid=${p}$ AND COALESCE(currency,"CNY")=${c}$',
|
||
{'d': stat_date, 'm': r.model, 'p': r.providerid, 'c': cur})
|
||
|
||
ns = {
|
||
'id': getID(),
|
||
'stat_date': stat_date,
|
||
'model': r.model,
|
||
'providerid': r.providerid,
|
||
'total_calls': r.total_calls,
|
||
'total_tokens': r.total_tokens or 0,
|
||
'total_amount': r.total_amount or 0,
|
||
'currency': cur,
|
||
'total_amount_cny': amount_cny,
|
||
'unit_price': round(float(r.total_amount) / r.total_tokens, 8) if r.total_tokens else None,
|
||
'avg_ttft_ms': round(r.avg_ttft, 1) if r.avg_ttft else None,
|
||
}
|
||
await sor.C('dm_provider_cost_daily', ns)
|
||
|
||
return len(rows)
|
||
|
||
|
||
async def run_etl_sync(sor):
|
||
"""每5分钟运行: 增量同步"""
|
||
await sync_call_fact(sor)
|
||
|
||
|
||
async def run_etl_aggregate(sor):
|
||
"""每小时运行: 聚合天级性能 + 并发统计"""
|
||
await aggregate_daily_perf(sor)
|
||
|
||
|
||
async def run_etl_provider_cost(sor):
|
||
"""每天运行: 供应商性价比"""
|
||
await aggregate_provider_cost(sor)
|