diff --git a/sage_datamart/etl.py b/sage_datamart/etl.py
index fa7f65a..6788ce6 100644
--- a/sage_datamart/etl.py
+++ b/sage_datamart/etl.py
@@ -7,6 +7,8 @@ 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"
 
@@ -113,32 +115,30 @@ async def sync_call_fact(sor, last_sync=None):
 
 
 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}, ...}
-    """
+    """从 product_usage_log 预取与 llmusage 关联的实际计费记录。"""
+    env = sor.env if hasattr(sor, 'env') and sor.env is not None else ServerEnv()
     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,
-                }
+        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,
+                    }
     except Exception as e:
         info('sage_datamart: product_usage_log lookup failed: ' + str(e))
     return usage_log_map
@@ -146,17 +146,19 @@ async def _load_usage_log_bills(sor, last_sync):
 
 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:
-        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)
+        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)
     except Exception as e:
         info('sage_datamart: exchange_rate lookup failed: ' + str(e))
     return rates