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直接用传入sor,去掉get_sor_context(None)导致的NoneType错误

Browse Source
This commit is contained in:
yumoqing 2026-07-21 17:45:12 +08:00
parent ed0107347e
commit 5030e0640c
1 changed files with 29 additions and 34 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

63
sage_datamart/etl.py
View File

@ -118,50 +118,45 @@ async def _load_usage_log_bills(sor, last_sync):
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,
}
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):
"""从 sage 库获取汇率表，返回 {currency: mid_rate} 字典"""
from sqlor.dbpools import get_sor_context
"""查询汇率表，返回 {currency: mid_rate} 字典"""
rates = {'CNY': 1.0}
try:
async with get_sor_context(sor.env if hasattr(sor, 'env') else None, 'accounting') as sage_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 sage_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