diff --git a/product_management/__init__.py b/product_management/__init__.py index 9aa9403..f5b9b0b 100644 --- a/product_management/__init__.py +++ b/product_management/__init__.py @@ -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 diff --git a/product_management/core.py b/product_management/core.py index 193659d..3b32974 100644 --- a/product_management/core.py +++ b/product_management/core.py @@ -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_balance（llmage 专属，非关键） + 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_items（PAY/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): + """资源引用 id（spec.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."""