From abc457d547c9be35e1363e12a3ffa89849c93bd9 Mon Sep 17 00:00:00 2001
From: yumoqing <yumoqing@gmail.com>
Date: Tue, 21 Jul 2026 17:54:48 +0800
Subject: [PATCH] =?UTF-8?q?fix:=20=E5=8E=BB=E6=8E=89get=5Fsor=5Fcontext?=
 =?UTF-8?q?=E5=8C=85=E8=A3=85,sor=E5=B7=B2=E4=BC=A0=E5=85=A5=E7=9B=B4?=
 =?UTF-8?q?=E6=8E=A5=E7=94=A8?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

---
 sage_datamart/etl.py | 64 +++++++++++++++++++-------------------------
 1 file changed, 28 insertions(+), 36 deletions(-)

diff --git a/sage_datamart/etl.py b/sage_datamart/etl.py
index 6788ce6..cc31984 100644
--- a/sage_datamart/etl.py
+++ b/sage_datamart/etl.py
@@ -7,8 +7,6 @@ from datetime import datetime, timedelta
 from appPublic.uniqueID import getID
 from appPublic.dictObject import DictObject
 from appPublic.log import info
-from ahserver.serverenv import ServerEnv
-from sqlor.dbpools import get_sor_context
 
 MODULE_NAME = "sage_datamart"
 
@@ -115,50 +113,44 @@ async def sync_call_fact(sor, last_sync=None):
 
 
 async def _load_usage_log_bills(sor, last_sync):
-    """从 product_usage_log 预取与 llmusage 关联的实际计费记录。"""
-    env = sor.env if hasattr(sor, 'env') and sor.env is not None else ServerEnv()
+    """从 product_usage_log 预取实际计费金额"""
     usage_log_map = {}
     try:
-        async with get_sor_context(env, '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,
-                    }
+        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):
-    """查询汇率表，返回 {currency: mid_rate} 字典"""
-    env = sor.env if hasattr(sor, 'env') and sor.env is not None else ServerEnv()
+    """查询汇率表"""
     rates = {'CNY': 1.0}
     try:
-        async with get_sor_context(env, 'accounting') as acc_sor:
-            sql = """SELECT from_currency, mid_rate FROM exchange_rate
-                     WHERE to_currency = 'CNY' AND effective_date <= ${d}$
-                     ORDER BY effective_date DESC"""
-            rows = await acc_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)
+        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