feat: product_accounting + backend_accounting — product-based accounting with distribution chain
- product_accounting: llm_id→product_id, calculate_product_cost, distribution chain discounts, consume_accounting - backend_accounting: accounting loop migrated from llmage - Both registered in load_product_management()
This commit is contained in:
parent
3132a837a6
commit
89847538f7
@ -55,7 +55,6 @@ async def get_products_by_category(category_id, status='1'):
|
||||
return await manager.get_products_by_category(category_id, status)
|
||||
|
||||
|
||||
|
||||
async def import_categories_and_products(resource_module, org_id, parent_category_id, user_id):
|
||||
"""Import sub-categories and products from a resource module."""
|
||||
manager = get_manager()
|
||||
@ -101,14 +100,23 @@ async def calculate_product_cost(product_id=None, product_code=None, usage_data=
|
||||
|
||||
|
||||
async def llm_id_to_product_id(llm_id):
|
||||
"""Convert LLM model ID (llmage.llm.id) to product ID (product.id).
|
||||
|
||||
Uses the product.resource_ref_id → llm.id mapping established during import.
|
||||
"""
|
||||
"""Convert LLM model ID (llmage.llm.id) to product ID (product.id)."""
|
||||
manager = get_manager()
|
||||
return await manager.llm_id_to_product_id(llm_id)
|
||||
|
||||
|
||||
async def product_accounting(llmusage):
|
||||
"""Full product accounting for one llmusage record."""
|
||||
manager = get_manager()
|
||||
return await manager.product_accounting(llmusage)
|
||||
|
||||
|
||||
async def backend_accounting():
|
||||
"""Background accounting loop."""
|
||||
manager = get_manager()
|
||||
await manager.backend_accounting()
|
||||
|
||||
|
||||
def load_product_management():
|
||||
"""Register all functions with ServerEnv so they can be called from .ui/.dspy files."""
|
||||
env = ServerEnv()
|
||||
@ -128,4 +136,7 @@ def load_product_management():
|
||||
env.execute_product_stream = execute_product_stream
|
||||
env.calculate_product_cost = calculate_product_cost
|
||||
env.llm_id_to_product_id = llm_id_to_product_id
|
||||
# Product accounting
|
||||
env.product_accounting = product_accounting
|
||||
env.backend_accounting = backend_accounting
|
||||
return True
|
||||
|
||||
@ -3,7 +3,9 @@ import json
|
||||
import time
|
||||
import datetime
|
||||
from appPublic.uniqueID import getID
|
||||
from appPublic.log import info, error, exception
|
||||
from appPublic.dictObject import DictObject
|
||||
from appPublic.log import info, error, exception, debug
|
||||
from appPublic.jsonConfig import getConfig
|
||||
from sqlor.dbpools import DBPools
|
||||
from ahserver.serverenv import ServerEnv
|
||||
|
||||
@ -1342,3 +1344,226 @@ WHERE resource_ref_id = ${llm_id}$ AND product_type = 'llm_model' LIMIT 1"""
|
||||
if recs:
|
||||
return recs[0].id
|
||||
return None
|
||||
|
||||
# ─── 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
|
||||
"""
|
||||
from appPublic.uniqueID import getID as _getID
|
||||
from accounting.consume import consume_accounting
|
||||
import json, time
|
||||
|
||||
env = ServerEnv()
|
||||
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:
|
||||
raise Exception(f'llm({llmid}) not found')
|
||||
ownerid = getattr(llm, 'ownerid', '0') or '0'
|
||||
providerid = getattr(llm, 'providerid', '0') or '0'
|
||||
|
||||
# 3. Parse usage data
|
||||
usages = llmusage.usages
|
||||
if isinstance(usages, str):
|
||||
usages = json.loads(usages)
|
||||
|
||||
# 4. Calculate raw cost
|
||||
cost_result = await self.calculate_product_cost(
|
||||
product_id=product_id, usage_data=usages, 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)
|
||||
|
||||
# 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
|
||||
try:
|
||||
from supplychain.init import get_distribution_chain as _get_chain
|
||||
chain = await _get_chain(ownerid, seller_orgid)
|
||||
except Exception:
|
||||
chain = [ownerid]
|
||||
|
||||
# 7. Get customer discount (seller gives to customer)
|
||||
cust_discount = await env.get_min_product_discount(
|
||||
product_id, seller_orgid, userorgid)
|
||||
customer_amount = raw_cost * cust_discount
|
||||
|
||||
# 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
|
||||
timestamp = time.strftime('%Y-%m-%d %H:%M:%S')
|
||||
orderid = _getID()
|
||||
|
||||
# Determine supplier discount (for last link to provider)
|
||||
last_link_discount = 1.0
|
||||
try:
|
||||
last_link_discount = await env.get_min_supplier_discount(
|
||||
product_id, ownerid, providerid) # supplier discount from product owner
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
accounting_items = []
|
||||
|
||||
if len(chain) == 1:
|
||||
# Direct: customer → product_owner → supplier
|
||||
# PAY: customer → owner, amount = customer_amount
|
||||
accounting_items.append(DictObject(
|
||||
action='PAY', customerid=userorgid, resellerid=ownerid,
|
||||
providerid=providerid, biz_date=biz_date, timestamp=timestamp,
|
||||
productid=product_id, transamt=customer_amount,
|
||||
variable={"交易金额": customer_amount, "交易手续费": 0}))
|
||||
# PAY*: owner → supplier, amount = raw_cost * supplier_discount
|
||||
supplier_amount = raw_cost * last_link_discount
|
||||
accounting_items.append(DictObject(
|
||||
action='PAY*', customerid=ownerid, resellerid=providerid,
|
||||
providerid=providerid, biz_date=biz_date, timestamp=timestamp,
|
||||
productid=product_id, transamt=supplier_amount,
|
||||
variable={"采购成本": supplier_amount}))
|
||||
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
|
||||
# Get this distributor's discount towards their downstream
|
||||
accounting_items.append(DictObject(
|
||||
action='PAY', customerid=userorgid, resellerid=orgid,
|
||||
providerid=providerid, biz_date=biz_date, timestamp=timestamp,
|
||||
productid=product_id, transamt=customer_amount,
|
||||
variable={"交易金额": customer_amount, "交易手续费": 0}))
|
||||
else:
|
||||
# Middle link: PAY*
|
||||
next_orgid = chain[i + 1]
|
||||
# Get discount from this level to next level
|
||||
up_discount = 1.0
|
||||
try:
|
||||
up_discount = await env.get_min_distributor_discount(
|
||||
product_id, next_orgid, orgid)
|
||||
except Exception:
|
||||
pass
|
||||
link_amount = raw_cost * up_discount
|
||||
accounting_items.append(DictObject(
|
||||
action='PAY*', customerid=orgid, resellerid=next_orgid,
|
||||
providerid=providerid, biz_date=biz_date, timestamp=timestamp,
|
||||
productid=product_id, transamt=link_amount,
|
||||
variable={"分销结算": link_amount}))
|
||||
# Final PAY*: top of chain → supplier
|
||||
supplier_amount = raw_cost * last_link_discount
|
||||
accounting_items.append(DictObject(
|
||||
action='PAY*', customerid=chain[0], resellerid=providerid,
|
||||
providerid=providerid, biz_date=biz_date, timestamp=timestamp,
|
||||
productid=product_id, transamt=supplier_amount,
|
||||
variable={"采购成本": supplier_amount}))
|
||||
|
||||
# 9. Write biz_order and call 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,
|
||||
"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
|
||||
}
|
||||
await sor.C('biz_orderdetail', orderdetail)
|
||||
await consume_accounting(sor, orderid, accounting_items)
|
||||
|
||||
# Mark llmusage as accounted
|
||||
async with db.sqlorContext('llmage') as sor:
|
||||
await sor.U('llmusage', {
|
||||
'id': llmusage.id,
|
||||
'accounting_status': 'accounted'
|
||||
})
|
||||
|
||||
return {'success': True, 'orderid': orderid}
|
||||
|
||||
async def backend_accounting(self):
|
||||
"""Background accounting loop — replaces llmage.backend_accounting."""
|
||||
import asyncio
|
||||
from datetime import datetime, timedelta
|
||||
from llmage.accounting import get_accounting_llmusages, llm_accoung_failed
|
||||
from llmage.utils import get_user_tpac
|
||||
|
||||
info(f"product backend accounting started ...")
|
||||
last_backup_date = None
|
||||
while True:
|
||||
try:
|
||||
lus = await get_accounting_llmusages()
|
||||
info(f"accounting loop: got {len(lus)} records")
|
||||
except Exception as e:
|
||||
exception(f"get_accounting_llmusages failed: {e}")
|
||||
lus = []
|
||||
for lu in lus:
|
||||
try:
|
||||
tpac = await get_user_tpac(lu.userid)
|
||||
if tpac:
|
||||
from llmage.accounting import tpac_accounting
|
||||
debug(f'{lu.id=},{lu.userid=}, {tpac=}, go tpac')
|
||||
await tpac_accounting(tpac, lu.userid, lu.llmid, lu.amount, lu.usages, lu.id, lu.model)
|
||||
else:
|
||||
debug(f'{lu.id=},{lu.userid=}, go product accounting')
|
||||
await self.product_accounting(lu)
|
||||
# Clean up failed records
|
||||
try:
|
||||
async with DBPools().sqlorContext('llmage') as sor:
|
||||
await sor.execute(
|
||||
"DELETE FROM llmusage_accounting_failed WHERE llmusageid=${luid}$",
|
||||
{'luid': lu.id})
|
||||
except Exception as e2:
|
||||
debug(f'清理失败记录异常: {e2}')
|
||||
except Exception as e:
|
||||
exception(f'{e}, {lu.id=}')
|
||||
await llm_accoung_failed(lu.id, reason=str(e))
|
||||
|
||||
# Daily backup
|
||||
today = datetime.now().strftime('%Y-%m-%d')
|
||||
if today != last_backup_date:
|
||||
yesterday = (datetime.now() - timedelta(days=1)).strftime('%Y-%m-%d')
|
||||
last_backup_date = today
|
||||
try:
|
||||
from llmage.accounting import backup_accounted_llmusage
|
||||
debug(f'date changed to {today}, triggering backup')
|
||||
await backup_accounted_llmusage(yesterday)
|
||||
except Exception as e:
|
||||
exception(f'backup failed: {e}')
|
||||
|
||||
await asyncio.sleep(10)
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user