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

fix: 去掉get_sor_context包装,sor已传入直接用

Browse Source
This commit is contained in:
yumoqing 2026-07-21 17:54:48 +08:00
parent 33447fbbe5
commit abc457d547
1 changed files with 28 additions and 36 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

18
sage_datamart/etl.py
View File

@ -7,8 +7,6 @@ from datetime import datetime, timedelta
from appPublic.uniqueID import getID from appPublic.uniqueID import getID
from appPublic.dictObject import DictObject from appPublic.dictObject import DictObject
from appPublic.log import info from appPublic.log import info
from ahserver.serverenv import ServerEnv
from sqlor.dbpools import get_sor_context
MODULE_NAME = "sage_datamart" MODULE_NAME = "sage_datamart"
@ -115,21 +113,17 @@ async def sync_call_fact(sor, last_sync=None):
async def _load_usage_log_bills(sor, last_sync): async def _load_usage_log_bills(sor, last_sync):
"""从 product_usage_log 预取与 llmusage 关联的实际计费记录。""" """从 product_usage_log 预取实际计费金额"""
env = sor.env if hasattr(sor, 'env') and sor.env is not None else ServerEnv()
usage_log_map = {} usage_log_map = {}
try: try:
async with get_sor_context(env, 'product_management') as pm_sor:
sql = """ sql = """
SELECT pul.source_ref_id, SELECT pul.source_ref_id, pul.sell_price, ps.purchase_currency
pul.sell_price,
ps.purchase_currency
FROM product_usage_log pul FROM product_usage_log pul
LEFT JOIN product_subscription ps ON ps.id = pul.subscription_id LEFT JOIN product_subscription ps ON ps.id = pul.subscription_id
WHERE pul.source_ref_table = 'llmusage' WHERE pul.source_ref_table = 'llmusage'
AND pul.use_time >= ${last_sync}$ AND pul.use_time >= ${last_sync}$
""" """
rows = await pm_sor.sqlExe(sql, {'last_sync': last_sync}) rows = await sor.sqlExe(sql, {'last_sync': last_sync})
for r in rows: for r in rows:
ref_id = r.source_ref_id if hasattr(r, 'source_ref_id') else r['source_ref_id'] 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'] sell_price = r.sell_price if hasattr(r, 'sell_price') else r['sell_price']
@ -145,15 +139,13 @@ async def _load_usage_log_bills(sor, last_sync):
async def _get_exchange_rates(sor, stat_date): 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} rates = {'CNY': 1.0}
try: try:
async with get_sor_context(env, 'accounting') as acc_sor:
sql = """SELECT from_currency, mid_rate FROM exchange_rate sql = """SELECT from_currency, mid_rate FROM exchange_rate
WHERE to_currency = 'CNY' AND effective_date <= ${d}$ WHERE to_currency = 'CNY' AND effective_date <= ${d}$
ORDER BY effective_date DESC""" ORDER BY effective_date DESC"""
rows = await acc_sor.sqlExe(sql, {'d': stat_date}) rows = await sor.sqlExe(sql, {'d': stat_date})
for r in rows: for r in rows:
cur = r.from_currency if hasattr(r, 'from_currency') else r['from_currency'] 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'] rate = r.mid_rate if hasattr(r, 'mid_rate') else r['mid_rate']