From 9f37ec4717fd2369af5ffef0d3c1f2d4e2d31fec Mon Sep 17 00:00:00 2001 From: yumoqing Date: Tue, 4 Aug 2026 12:55:35 +0800 Subject: [PATCH] fix(accounting): auto-sync published llm to product on publish; skip PAY leg for owner self-use --- product_management/__init__.py | 6 ++ product_management/core.py | 119 +++++++++++++++++++++++++++++---- 2 files changed, 113 insertions(+), 12 deletions(-) diff --git a/product_management/__init__.py b/product_management/__init__.py index 58312f1..7fcb8aa 100644 --- a/product_management/__init__.py +++ b/product_management/__init__.py @@ -105,6 +105,12 @@ async def llm_id_to_product_id(llm_id): return await manager.llm_id_to_product_id(llm_id) +async def sync_llm_product(llm_id): + """Auto-sync a published llm model as product (idempotent).""" + manager = get_manager() + return await manager.sync_llm_product(llm_id) + + async def product_accounting(llmusage): """Full product accounting for one llmusage record.""" manager = get_manager() diff --git a/product_management/core.py b/product_management/core.py index af9b5d2..52a371a 100644 --- a/product_management/core.py +++ b/product_management/core.py @@ -1114,6 +1114,91 @@ class ProductManager: f'跳过 {skipped_cats} 个已有类别, {skipped_prods} 个已有产品' } + # ─── Auto-sync a published llm model as product ─── + + async def sync_llm_product(self, llm_id): + """llmage 上架模型后自动同步为产品（幂等）。 + + 复用 import_categories_and_products 的增量导入（已存在的产品跳过）， + 顺带补齐其他已上架但缺产品的模型。 + """ + if not llm_id: + return {'success': False, 'error': '缺少 llm_id'} + + dbname = self._get_dbname() + lmage_dbname = ServerEnv().get_module_dbname('llmage') + + # 1. llm 记录必须存在且已上架 + async with DBPools().sqlorContext(lmage_dbname) as sor: + llm_rows = await sor.sqlExe( + "SELECT id, status, ownerid FROM llm WHERE id = ${lid}$", + {'lid': llm_id}) + if not llm_rows: + return {'success': False, 'error': '模型不存在: %s' % llm_id} + llm_rec = llm_rows[0] + if getattr(llm_rec, 'status', '') != 'published': + return {'success': False, 'error': '模型未上架，无需同步'} + ownerid = getattr(llm_rec, 'ownerid', None) or '0' + + # 2. 幂等：产品已存在则跳过 + async with DBPools().sqlorContext(dbname) as sor: + existing = await sor.sqlExe( + """SELECT id FROM product + WHERE resource_ref_id = ${ref_id}$ AND org_id = ${org_id}$""", + {'ref_id': llm_id, 'org_id': ownerid}) + if existing: + return {'success': True, 'skipped': True, 'product_id': existing[0].id} + + # 3. 确定导入父类别：沿用已有 llmage 类别的父级，避免类别树分裂； + # 完全没有 llmage 类别时才新建根类别 + async with DBPools().sqlorContext(dbname) as sor: + cat_rows = await sor.sqlExe( + """SELECT parent_id FROM product_category + WHERE resource_module = 'llmage' AND org_id = ${org_id}$ + LIMIT 1""", + {'org_id': ownerid}) + parent_category_id = None + if cat_rows: + parent_category_id = getattr(cat_rows[0], 'parent_id', None) + if not parent_category_id: + parent_category_id = getID() + now = time.strftime('%Y-%m-%d %H:%M:%S') + await sor.C('product_category', { + 'id': parent_category_id, + 'parent_id': '0', + 'name': 'AI模型', + 'description': 'llmage 模型产品类别', + 'has_product': '1', + 'product_type': 'llm_model', + 'product_type_title': '', + 'sort_order': '0', + 'icon': '', + 'status': '1', + 'resource_module': 'llmage', + 'org_id': ownerid, + 'created_by': 'system', + 'created_at': now, + 'updated_at': now + }) + + # 4. 增量导入（跳过已有类别/产品） + result = await self.import_categories_and_products( + resource_module='llmage', org_id=ownerid, + parent_category_id=parent_category_id, user_id='system') + if not result or not result.get('success'): + err = result.get('error', '导入失败') if result else '导入无返回' + return {'success': False, 'error': err} + + # 5. 确认产品已落库 + async with DBPools().sqlorContext(dbname) as sor: + new_prod = await sor.sqlExe( + """SELECT id FROM product + WHERE resource_ref_id = ${ref_id}$ AND org_id = ${org_id}$""", + {'ref_id': llm_id, 'org_id': ownerid}) + if new_prod: + return {'success': True, 'product_id': new_prod[0].id} + return {'success': False, 'error': '导入后仍未找到产品（模型可能不在已上架列表）'} + # ─── Resource Module Interface Dispatcher ─── async def _get_product_interface(self, product_id=None, product_code=None): @@ -1367,6 +1452,12 @@ 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): @@ -1446,12 +1537,14 @@ WHERE resource_ref_id = ${llm_id}$ AND product_type = 'llm_model' LIMIT 1""" 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, - currency=user_currency, base_amount=round(customer_amount, 2), - variable={"交易金额": customer_amount, "交易币种": user_currency, "交易手续费": 0})) + # (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, + providerid=providerid, biz_date=biz_date, timestamp=timestamp, + 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 = raw_cost * supplier_discount supplier_amount = raw_cost * last_link_discount accounting_items.append(DictObject( @@ -1466,12 +1559,14 @@ WHERE resource_ref_id = ${llm_id}$ AND product_type = 'llm_model' LIMIT 1""" is_last = (i == len(chain) - 1) if is_last: # Last link (closest to customer): PAY - accounting_items.append(DictObject( - action='PAY', customerid=userorgid, resellerid=orgid, - providerid=providerid, biz_date=biz_date, timestamp=timestamp, - productid=product_id, transamt=customer_amount, - currency=user_currency, base_amount=round(customer_amount, 2), - variable={"交易金额": customer_amount, "交易币种": user_currency, "交易手续费": 0})) + # (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, + providerid=providerid, biz_date=biz_date, timestamp=timestamp, + productid=product_id, transamt=customer_amount, + currency=user_currency, base_amount=round(customer_amount, 2), + variable={"交易金额": customer_amount, "交易币种": user_currency, "交易手续费": 0})) else: # Middle link: PAY* next_orgid = chain[i + 1]