ETL: 金额取product_usage_log.sell_price实际计费金额,币种取purchase_currency,多币种分别统计并并账到CNY

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=<luid> 关联 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