yumoqing/sage_datamart
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix: _get_exchange_rates/_load_usage_log_bills env取sor.env,无则ServerEnv()兜底

Browse Source
This commit is contained in:
yumoqing 2026-07-21 17:52:18 +08:00
parent 5030e0640c
commit 33447fbbe5
1 changed files with 35 additions and 33 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

68
sage_datamart/etl.py
View File

@ -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