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

fix: init_data.py补全缺失字段+userorgid为None兜底(空串),支持重复执行

Browse Source
This commit is contained in:
yumoqing 2026-07-21 15:05:02 +08:00
parent 52acdbbfab
commit 237c09262f
1 changed files with 18 additions and 10 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

28
scripts/init_data.py
View File

@ -54,7 +54,7 @@ async def batch_sync_call_fact(sor, source_table, batch_size=1000):
if not rows: if not rows:
break break
print(f' [{source_table}] batch offset={offset} rows={len(rows)}') print(' [' + source_table + '] batch offset=' + str(offset) + ' rows=' + str(len(rows)))
for r in rows: for r in rows:
r = DictObject(r) r = DictObject(r)
@ -73,16 +73,21 @@ async def batch_sync_call_fact(sor, source_table, batch_size=1000):
total_skipped += 1 total_skipped += 1
continue continue
currency = getattr(r, 'amount_currency', None) or 'CNY'
amount = getattr(r, 'amount', 0) or 0
userorgid = getattr(r, 'userorgid', None)
distributor_orgid = getattr(r, 'distributor_orgid', None)
ns = { ns = {
'id': getID(), 'id': getID(),
'luid': r.id, 'luid': r.id,
'llmid': r.llmid, 'llmid': r.llmid,
'model': r.llm_model, 'model': r.llm_model or '',
'catelogid': r.catelogid or '', 'catelogid': r.catelogid or '',
'userid': r.userid, 'userid': r.userid or '',
'userorgid': r.userorgid, 'userorgid': userorgid or '',
'ownerid': r.ownerid, 'ownerid': r.ownerid or '',
'providerid': r.providerid, 'providerid': r.providerid or '',
'call_date': r.use_date or call_time.strftime('%Y-%m-%d'), 'call_date': r.use_date or call_time.strftime('%Y-%m-%d'),
'call_hour': call_time.hour, 'call_hour': call_time.hour,
'call_time': r.use_time, 'call_time': r.use_time,
@ -91,15 +96,18 @@ async def batch_sync_call_fact(sor, source_table, batch_size=1000):
'prompt_tokens': prompt_tokens, 'prompt_tokens': prompt_tokens,
'completion_tokens': completion_tokens, 'completion_tokens': completion_tokens,
'status': r.status or 'SUCCEEDED', 'status': r.status or 'SUCCEEDED',
'amount': r.amount or 0, 'amount': amount,
'currency': getattr(r, 'amount_currency', None) or 'CNY', 'currency': currency,
'amount_cny': amount,
'distributor_orgid': distributor_orgid or '',
'sale_userid': '',
} }
await sor.C('dm_model_call_fact', ns) await sor.C('dm_model_call_fact', ns)
total_inserted += 1 total_inserted += 1
offset += batch_size offset += batch_size
print(f' [{source_table}] done: inserted={total_inserted}, skipped={total_skipped}') print(' [' + source_table + '] done: inserted=' + str(total_inserted) + ', skipped=' + str(total_skipped))
return total_inserted, total_skipped return total_inserted, total_skipped
@ -112,7 +120,7 @@ async def batch_aggregate_history(sor):
{} {}
) )
dates = [r['call_date'].strftime('%Y-%m-%d') if hasattr(r['call_date'], 'strftime') else str(r['call_date']) for r in rows] dates = [r['call_date'].strftime('%Y-%m-%d') if hasattr(r['call_date'], 'strftime') else str(r['call_date']) for r in rows]
print(f' aggregating {len(dates)} distinct dates...') print(' aggregating ' + str(len(dates)) + ' distinct dates...')
for d in dates: for d in dates:
await aggregate_daily_perf(sor, stat_date=d) await aggregate_daily_perf(sor, stat_date=d)