From 89847538f7618337f6a5bd80525f000a8eac0ae9 Mon Sep 17 00:00:00 2001 From: yumoqing Date: Tue, 7 Jul 2026 12:27:40 +0800 Subject: [PATCH] =?UTF-8?q?feat:=20product=5Faccounting=20+=20backend=5Fac?= =?UTF-8?q?counting=20=E2=80=94=20product-based=20accounting=20with=20dist?= =?UTF-8?q?ribution=20chain?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 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() --- product_management/__init__.py | 21 ++- product_management/core.py | 227 ++++++++++++++++++++++++++++++++- 2 files changed, 242 insertions(+), 6 deletions(-) diff --git a/product_management/__init__.py b/product_management/__init__.py index ef42251..58312f1 100644 --- a/product_management/__init__.py +++ b/product_management/__init__.py @@ -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 diff --git a/product_management/core.py b/product_management/core.py index 8610ca4..ffb3f7a 100644 --- a/product_management/core.py +++ b/product_management/core.py @@ -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)