feat(accounting): 计费落账通用化,账号/存储接入出账

- 抽 product_accounting_generic(product_id, usage_data, ...):算金额+分销链+consume_accounting,llm/账号/存储通用
- product_accounting(llmusage) 变薄封装:转换 llmusage→通用参数→generic+finalize+标记
- 新增 account_resource_accounting / storage_resource_accounting:acctres/storres usage → generic → 标记 accounted
- _resource_ref_to_product_id: spec.id → product.id(按 product_type 区分 account/workspace_storage)
This commit is contained in:
ymq 2026-08-26 17:20:23 +08:00
parent 622dec8495
commit 29f1c1fc92
2 changed files with 114 additions and 101 deletions

View File

@ -117,6 +117,18 @@ async def product_accounting(llmusage):
return await manager.product_accounting(llmusage)
async def account_resource_accounting(usage_record):
"""账号用量计费落账acctres_usage → 通用计费落账)。"""
manager = get_manager()
return await manager.account_resource_accounting(usage_record)
async def storage_resource_accounting(usage_record):
"""存储用量计费落账storres_usage → 通用计费落账)。"""
manager = get_manager()
return await manager.storage_resource_accounting(usage_record)
async def backend_accounting():
"""Background accounting loop."""
manager = get_manager()
@ -145,5 +157,7 @@ def load_product_management():
env.sync_llm_product = sync_llm_product
# Product accounting
env.product_accounting = product_accounting
env.account_resource_accounting = account_resource_accounting
env.storage_resource_accounting = storage_resource_accounting
env.backend_accounting = backend_accounting
return True

View File

@ -1585,25 +1585,15 @@ WHERE resource_ref_id = ${llm_id}$ AND product_type = 'llm_model' LIMIT 1"""
# ─── Product Accounting ───
async def product_accounting(self, llmusage):
"""Full product accounting for one llmusage record.
1. Convert llm_id product_id
2. Calculate raw cost via resource module
3. Determine distribution chain (product_owner ... seller)
4. Compute final amounts with discounts for each party
5. Build accounting items and call consume_accounting
"""
env = ServerEnv()
"""llmage 用量计费落账(薄封装:转换 llmusage → 通用参数 → generic"""
llmid = llmusage.llmid
userorgid = llmusage.userorgid
userid = llmusage.userid
# 1. Get product_id
product_id = await self.llm_id_to_product_id(llmid)
if not product_id:
raise Exception(f'llm({llmid}) has no product mapping')
# 2. Get llm info for ownerid
from llmage.utils import get_llmage_llm
llm = await get_llmage_llm(llmid)
if not llm:
@ -1611,27 +1601,55 @@ WHERE resource_ref_id = ${llm_id}$ AND product_type = 'llm_model' LIMIT 1"""
ownerid = getattr(llm, 'ownerid', '0') or '0'
providerid = getattr(llm, 'providerid', '0') or '0'
# 2b. Self-use detection: customer org == owner org (e.g. owner-side users
# or platform users with NULL/'0' userorgid using owner '0' models).
# Self-use skips the PAY leg — merchant never opens a customer account
# for itself, so PAY would fail account lookup. PAY* legs stay balanced.
is_self_use = (userorgid or '0') == (ownerid or '0')
# 3. Parse usage data
usages = llmusage.usages
if isinstance(usages, str):
usages = json.loads(usages)
# 4. Calculate raw cost
result = await self.product_accounting_generic(
product_id, usages, userorgid, userid,
providerid=providerid, ownerid=ownerid,
tenantid=getattr(llmusage, 'tenantid', None))
# finalize_balancellmage 专属,非关键)
try:
luid = getattr(llmusage, 'id', None)
if luid:
from llmage.balance import finalize_balance
await finalize_balance(ServerEnv(), luid, result.get('customer_amount', 0),
llmid=llmid)
except Exception as e:
exception(f'finalize_balance failed (non-critical): {e}')
# 标记 llmusage accounted
lmage_dbname = ServerEnv().get_module_dbname('llmage')
async with DBPools().sqlorContext(lmage_dbname) as sor:
await sor.U('llmusage', {'id': llmusage.id, 'accounting_status': 'accounted'})
return result
async def product_accounting_generic(self, product_id, usage_data, userorgid, userid,
providerid=None, ownerid=None, tenantid=None):
"""通用商品计费落账llm/账号/存储通用)。
核心算金额(calculate_product_cost含客户折扣+供应商成本)
分销链各环节金额 consume_accounting 落账
返回 {'success', 'orderid', 'customer_amount', 'supplier_cost', 'original_amount'}
"""
env = ServerEnv()
ownerid = ownerid or '0'
providerid = providerid or '0'
is_self_use = (userorgid or '0') == (ownerid or '0')
# 1. 算金额(客户售价 amount + 供应商成本 cost
cost_result = await self.calculate_product_cost(
product_id=product_id, usage_data=usages, user_org_id=userorgid)
product_id=product_id, usage_data=usage_data, user_org_id=userorgid)
if not cost_result or not cost_result.get('success'):
raise Exception(f'calculate_product_cost failed: {cost_result}')
raw_cost = cost_result.get('amount', 0)
original_amount = cost_result.get('original_amount', 0) or raw_cost
supplier_cost = cost_result.get('cost', 0) or raw_cost
# 4b. Get product currency
# 2. 币种转换
prod_currency = 'CNY'
try:
dbname = self._get_dbname()
@ -1643,12 +1661,10 @@ WHERE resource_ref_id = ${llm_id}$ AND product_type = 'llm_model' LIMIT 1"""
except Exception:
pass
# 4c. Get user billing currency and exchange rates
user_currency = 'CNY'
cost_in_user = raw_cost
cost_base = raw_cost
try:
env = ServerEnv()
user_currency = await env.get_user_currency(userorgid)
if prod_currency != user_currency:
rate = await env.get_exchange_rate(prod_currency, user_currency, 'sell_rate')
@ -1657,42 +1673,22 @@ WHERE resource_ref_id = ${llm_id}$ AND product_type = 'llm_model' LIMIT 1"""
except Exception:
pass
# 5. Determine seller (who sold to customer)
dbname = self._get_dbname()
seller_orgid = userorgid # default: customer bought directly from owner
# Try tenantid from llmusage
tenantid = getattr(llmusage, 'tenantid', None)
if tenantid:
seller_orgid = tenantid
# 6. Get distribution chain
# 3. 分销链
seller_orgid = tenantid or userorgid
try:
from supplychain.init import get_distribution_chain as _get_chain
chain = await _get_chain(ownerid, seller_orgid)
except Exception:
chain = [ownerid]
# 7. 客户售价calculate_product_cost 的 amount 已含客户折扣discount 模块),
# 这里不再重复乘,避免双重折扣。
customer_amount = cost_in_user
# 8. Build accounting items for each link in chain
# Chain: product_owner → ... → seller
# Each link: PAY* (reseller→provider) except last link: PAY (customer→reseller)
biz_date = await env.get_business_date(None) # will use default
biz_date = await env.get_business_date(None)
timestamp = time.strftime('%Y-%m-%d %H:%M:%S')
orderid = getID()
# 供应商成本calculate_product_cost 的 cost 已含 supplychain 供应商折扣(基于原价)。
# 不用 discount 模块的 get_min_supplier_discount供应商折扣归 supplychain 管)。
# 4. 构建 accounting_itemsPAY/PAY*
accounting_items = []
if len(chain) == 1:
# Direct: customer → product_owner → supplier
# PAY: customer → owner, amount = customer_amount
# (skipped for self-use: no customer account exists for owner itself)
if not is_self_use:
accounting_items.append(DictObject(
action='PAY', customerid=userorgid, resellerid=ownerid,
@ -1700,21 +1696,16 @@ WHERE resource_ref_id = ${llm_id}$ AND product_type = 'llm_model' LIMIT 1"""
productid=product_id, transamt=customer_amount,
currency=user_currency, base_amount=round(customer_amount, 2),
variable={"交易金额": customer_amount, "交易币种": user_currency, "交易手续费": 0}))
# PAY*: owner → supplier, amount = supplier_cost含 supplychain 供应商折扣)
supplier_amount = supplier_cost
accounting_items.append(DictObject(
action='PAY*', customerid=ownerid, resellerid=ownerid,
providerid=providerid, biz_date=biz_date, timestamp=timestamp,
productid=product_id, transamt=supplier_amount,
productid=product_id, transamt=supplier_cost,
currency=prod_currency, base_amount=cost_base,
variable={"采购成本": supplier_amount, "采购币种": prod_currency}))
variable={"采购成本": supplier_cost, "采购币种": prod_currency}))
else:
# Multi-level distribution chain
for i, orgid in enumerate(chain):
is_last = (i == len(chain) - 1)
if is_last:
# Last link (closest to customer): PAY
# (skipped for self-use: no customer account for owner itself)
if not is_self_use:
accounting_items.append(DictObject(
action='PAY', customerid=userorgid, resellerid=orgid,
@ -1723,9 +1714,6 @@ WHERE resource_ref_id = ${llm_id}$ AND product_type = 'llm_model' LIMIT 1"""
currency=user_currency, base_amount=round(customer_amount, 2),
variable={"交易金额": customer_amount, "交易币种": user_currency, "交易手续费": 0}))
else:
# Middle link: PAY*
# 分销折扣从 supplychain 查distribution_agreement_items
# 不用 discount 模块的 get_min_distributor_discount。
next_orgid = chain[i + 1]
link_amount = raw_cost
try:
@ -1749,64 +1737,75 @@ WHERE resource_ref_id = ${llm_id}$ AND product_type = 'llm_model' LIMIT 1"""
productid=product_id, transamt=link_amount,
currency=prod_currency, base_amount=round(link_amount, 2),
variable={"分销结算": link_amount, "结算币种": prod_currency}))
# Final PAY*: top of chain → supplier
supplier_amount = supplier_cost
accounting_items.append(DictObject(
action='PAY*', customerid=chain[0], resellerid=chain[0],
providerid=providerid, biz_date=biz_date, timestamp=timestamp,
productid=product_id, transamt=supplier_amount,
productid=product_id, transamt=supplier_cost,
currency=prod_currency, base_amount=cost_base,
variable={"采购成本": supplier_amount, "采购币种": prod_currency}))
variable={"采购成本": supplier_cost, "采购币种": prod_currency}))
# 9. Write biz_order and call consume_accounting
# 5. 写 biz_order + consume_accounting
dbname = self._get_dbname()
db = DBPools()
config = getConfig()
db.databases = config.databases
async with db.sqlorContext(dbname) as sor:
order = {
"id": orderid,
"customerid": userorgid,
"resellerid": ownerid,
"order_date": biz_date,
"order_status": "1",
"business_op": "PAY",
"amount": customer_amount,
"currency": user_currency,
"userid": userid,
"productid": product_id
}
await sor.C('biz_order', order)
orderdetail = {
"id": getID(),
"orderid": orderid,
"productid": product_id,
"product_cnt": 1,
"trans_amount": customer_amount,
"currency": user_currency,
}
await sor.C('biz_orderdetail', orderdetail)
await sor.C('biz_order', {
"id": orderid, "customerid": userorgid, "resellerid": ownerid,
"order_date": biz_date, "order_status": "1", "business_op": "PAY",
"amount": customer_amount, "currency": user_currency,
"userid": userid, "productid": product_id})
await sor.C('biz_orderdetail', {
"id": getID(), "orderid": orderid, "productid": product_id,
"product_cnt": 1, "trans_amount": customer_amount, "currency": user_currency})
await consume_accounting(sor, orderid, accounting_items)
# Finalize Redis balance reservation (non-critical)
try:
luid = getattr(llmusage, 'id', None)
if luid:
from llmage.balance import finalize_balance
await finalize_balance(ServerEnv(), luid, customer_amount,
llmid=getattr(llmusage, 'llmid', None))
except Exception as e:
exception(f'finalize_balance failed (non-critical): {e}')
return {'success': True, 'orderid': orderid, 'customer_amount': customer_amount,
'supplier_cost': supplier_cost, 'original_amount': original_amount}
# Mark llmusage as accounted
lmage_dbname = ServerEnv().get_module_dbname('llmage')
async with db.sqlorContext(lmage_dbname) as sor:
await sor.U('llmusage', {
'id': llmusage.id,
'accounting_status': 'accounted'
})
async def _resource_ref_to_product_id(self, ref_id, product_type):
"""资源引用 idspec.id→ 产品 id按 product_type 区分资源模块)。"""
dbname = self._get_dbname()
async with DBPools().sqlorContext(dbname) as sor:
recs = await sor.sqlExe(
"SELECT id FROM product WHERE resource_ref_id=${rid}$ "
"AND product_type=${pt}$ LIMIT 1",
{'rid': ref_id, 'pt': product_type})
if recs:
return recs[0].id
return None
return {'success': True, 'orderid': orderid}
async def account_resource_accounting(self, usage_record):
"""账号用量计费落账acctres_usage → generic"""
spec_id = getattr(usage_record, 'spec_id', '')
userorgid = getattr(usage_record, 'userorgid', '') or '0'
userid = getattr(usage_record, 'userid', '') or ''
product_id = await self._resource_ref_to_product_id(spec_id, 'account')
if not product_id:
raise Exception(f'acctres spec({spec_id}) has no product mapping')
usages = getattr(usage_record, 'usages', '{}')
if isinstance(usages, str):
usages = json.loads(usages)
result = await self.product_accounting_generic(product_id, usages, userorgid, userid)
from account_resource.init import mark_acctres_accounted
await mark_acctres_accounted(getattr(usage_record, 'id', ''))
return result
async def storage_resource_accounting(self, usage_record):
"""存储用量计费落账storres_usage → generic"""
spec_id = getattr(usage_record, 'spec_id', '')
userorgid = getattr(usage_record, 'userorgid', '') or '0'
userid = getattr(usage_record, 'userid', '') or ''
product_id = await self._resource_ref_to_product_id(spec_id, 'workspace_storage')
if not product_id:
raise Exception(f'storres spec({spec_id}) has no product mapping')
usages = getattr(usage_record, 'usages', '{}')
if isinstance(usages, str):
usages = json.loads(usages)
result = await self.product_accounting_generic(product_id, usages, userorgid, userid)
from storage_resource.init import mark_storres_accounted
await mark_storres_accounted(getattr(usage_record, 'id', ''))
return result
async def backend_accounting(self):
"""Background accounting loop — replaces llmage.backend_accounting."""