From 43a115a14db1bd30fe55670f9e9405b7f3632ef7 Mon Sep 17 00:00:00 2001 From: yumoqing Date: Mon, 20 Jul 2026 16:10:17 +0800 Subject: [PATCH] =?UTF-8?q?ETL:=20=E9=87=91=E9=A2=9D=E5=8F=96product=5Fusa?= =?UTF-8?q?ge=5Flog.sell=5Fprice=E5=AE=9E=E9=99=85=E8=AE=A1=E8=B4=B9?= =?UTF-8?q?=E9=87=91=E9=A2=9D,=E5=B8=81=E7=A7=8D=E5=8F=96purchase=5Fcurren?= =?UTF-8?q?cy,=E5=A4=9A=E5=B8=81=E7=A7=8D=E5=88=86=E5=88=AB=E7=BB=9F?= =?UTF-8?q?=E8=AE=A1=E5=B9=B6=E5=B9=B6=E8=B4=A6=E5=88=B0CNY?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- sage_datamart/etl.py | 58 ++++++++++++++++++++++++++++++++++++++++++-- 1 file changed, 56 insertions(+), 2 deletions(-) diff --git a/sage_datamart/etl.py b/sage_datamart/etl.py index 29ec3e5..e206491 100644 --- a/sage_datamart/etl.py +++ b/sage_datamart/etl.py @@ -27,7 +27,12 @@ def parse_usages(usages_str): async def sync_call_fact(sor, last_sync=None): - """增量同步 llmusage → dm_model_call_fact（含币种折算）""" + """增量同步 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') @@ -35,6 +40,9 @@ async def sync_call_fact(sor, last_sync=None): 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 @@ -60,7 +68,18 @@ async def sync_call_fact(sor, last_sync=None): 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' - amount = r.amount or 0 + + # 优先取 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) @@ -93,6 +112,41 @@ async def sync_call_fact(sor, last_sync=None): return count +async def _load_usage_log_bills(sor, last_sync): + """从 product_usage_log 预取与 llmusage 关联的实际计费记录。 + + product_usage_log 通过 source_ref_table='llmusage' + source_ref_id= 关联 llmusage。 + 返回 {llmusage.id: {'sell_price': float, 'purchase_currency': str|None}, ...} + """ + from sqlor.dbpools import get_sor_context + usage_log_map = {} + try: + async with get_sor_context(sor.env if hasattr(sor, 'env') else None, + 'product_management') as pm_sor: + 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 pm_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): """从 sage 库获取汇率表，返回 {currency: mid_rate} 字典""" from sqlor.dbpools import get_sor_context