From cbbdac9adc91b400fc7fa3932eb6a1be167835c1 Mon Sep 17 00:00:00 2001 From: ping <1017253325@qq.com> Date: Wed, 22 Jul 2026 15:37:50 +0800 Subject: [PATCH] update --- b/cntoai/sync_model_to_llm.dspy | 344 +++++++++++++++++++++++--------- 1 file changed, 248 insertions(+), 96 deletions(-) diff --git a/b/cntoai/sync_model_to_llm.dspy b/b/cntoai/sync_model_to_llm.dspy index c119057..ee87c1d 100644 --- a/b/cntoai/sync_model_to_llm.dspy +++ b/b/cntoai/sync_model_to_llm.dspy @@ -6,6 +6,177 @@ async def sync_model_to_llm(ns={}): return ns.get('orgid') user_li = await sor.R('users', {'id': await get_user()}) return user_li[0].get('orgid') if user_li else '' + + def build_llm_data(item): + return { + 'id': item.get('id') or uuid(), + 'name': item.get('name'), + 'model': item.get('model'), + 'description': item.get('description'), + 'iconid': item.get('iconid'), + 'upappid': item.get('upappid'), + 'providerid': item.get('providerid'), + 'ownerid': item.get('ownerid') or '0', + 'enabled_date': item.get('enabled_date'), + 'expired_date': item.get('expired_date'), + 'min_balance': item.get('min_balance'), + 'status': item.get('status') or 'published', + 'catelogname': item.get('catelogname'), + 'catelog_id': item.get('catelog_id') or item.get('catelogid'), + 'llmcatelogid': item.get('catelog_id') or item.get('catelogid'), + 'apiname': item.get('apiname'), + 'query_apiname': item.get('query_apiname'), + 'query_period': item.get('query_period'), + 'ppid': item.get('ppid'), + 'provider_name': item.get('provider_name'), + 'pricing_display': item.get('pricing_display'), + 'provider_iconurl': item.get('provider_iconurl') + } + + async def ensure_provider(sor, provider_orgcode, provider_name, sync_orgid, item): + exist_org_li = await sor.R('organization', {'orgcode': provider_orgcode, 'del_flg': '0'}) + if exist_org_li: + provider_orgid = exist_org_li[0].get('id') + await sor.U('organization', {'id': provider_orgid}, { + 'id': provider_orgid, + 'orgname': provider_name, + 'orgcode': provider_orgcode, + 'parentid': sync_orgid, + 'org_type': '4' + }) + else: + provider_orgid = uuid() + await sor.C('organization', { + 'id': provider_orgid, + 'orgname': provider_name, + 'contactor': '连战', + 'contactor_phone': '13800138000', + 'address': '北京市海淀区', + 'main_business': provider_name, + 'orgcode': provider_orgcode, + 'emailaddress': 'lianzhan@cntoai.com', + 'org_type': '4', + 'parentid': sync_orgid + }) + await openProviderAccounts(sor, sync_orgid, provider_orgid) + + provider_li = await sor.R('provider', {'orgid': provider_orgid, 'del_flg': '0'}) + provider_data = { + 'name': provider_name, + 'description': provider_name, + 'appid': item.get('upappid') or '', + 'providerkey': provider_orgcode, + 'state': '1', + 'start_date': '2026-07-01', + 'end_date': '9999-12-31', + 'settle_mode': '3', + 'settle_flowid': '0', + 'settle_datep': 'M00', + 'discount_mode': 1, + 'orgid': provider_orgid + } + if provider_li: + provider_id = provider_li[0].get('id') + provider_data['id'] = provider_id + await sor.U('provider', {'id': provider_id}, provider_data) + else: + provider_id = uuid() + provider_data['id'] = provider_id + await sor.C('provider', provider_data) + + protocol_li = await sor.R('saleprotocol', {'offer_orgid': provider_orgid, 'bid_orgid': sync_orgid, 'del_flg': '0'}) + if protocol_li: + protocolid = protocol_li[0].get('id') + await sor.U('saleprotocol', {'id': protocolid}, { + 'id': protocolid, + 'offer_orgid': provider_orgid, + 'bid_orgid': sync_orgid, + 'salemode': '0', + 'settle_mode': '3', + 'settle_dp': 'M00', + 'start_date': '2026-07-01', + 'end_date': '9999-12-31' + }) + else: + protocolid = uuid() + await sor.C('saleprotocol', { + 'id': protocolid, + 'offer_orgid': provider_orgid, + 'bid_orgid': sync_orgid, + 'salemode': '0', + 'settle_mode': '3', + 'settle_dp': 'M00', + 'start_date': '2026-07-01', + 'end_date': '9999-12-31' + }) + + return { + 'provider_orgid': provider_orgid, + 'provider_id': provider_id, + 'protocolid': protocolid + } + + async def sync_product(sor, item, category_list, provider_orgid, protocolid, sync_orgid): + product_data = { + 'providerid': provider_orgid, + 'providerpid': item.get('id'), + 'name': item.get('name'), + 'description': item.get('description') or '', + 'ptype': '21', + 'state': '1', + 'salemode': '0', + 'product_code': item.get('model'), + 'reseller_orgid': sync_orgid, + 'classify': 'A', + 'label': item.get('catelogname') or category_list.get('catelogname'), + 'productgroup': item.get('catelog_id') or item.get('catelogid') or category_list.get('catelogid'), + 'product_area': '北京', + 'effect_date': '2026-07-01', + 'expire_date': '9999-12-31' + } + product_li = await sor.R('product', {'providerid': provider_orgid, 'product_code': item.get('model'), 'del_flg': '0'}) + if product_li: + product_id = product_li[0].get('id') + product_data['id'] = product_id + await sor.U('product', {'id': product_id}, product_data) + product_action = 'update' + else: + product_id = uuid() + product_data['id'] = product_id + await sor.C('product', product_data) + product_action = 'insert' + + salemode_data = { + 'protocolid': protocolid, + 'providerid': provider_orgid, + 'productid': product_id, + 'discount': 1, + 'unit': 'hour' + } + product_salemode_li = await sor.R('product_salemode', { + 'protocolid': protocolid, + 'providerid': provider_orgid, + 'productid': product_id, + 'del_flg': '0' + }) + if product_salemode_li: + salemode_data['id'] = product_salemode_li[0].get('id') + await sor.U('product_salemode', {'id': salemode_data['id']}, salemode_data) + else: + salemode_data['id'] = uuid() + await sor.C('product_salemode', salemode_data) + + return product_action + + async def upsert_llm(sor, item): + llm_data = build_llm_data(item) + exist_llm = await sor.R('llm', {'model': item.get('model')}) + if exist_llm: + llm_data['id'] = exist_llm[0].get('id') + await sor.U('llm', {'id': llm_data['id']}, llm_data) + return 'update' + await sor.C('llm', llm_data) + return 'insert' # 从数据库读取domain和Bearer token db = DBPools() @@ -45,112 +216,93 @@ async def sync_model_to_llm(ns={}): # 插入数据库 db = DBPools() async with db.sqlorContext('kboss') as sor: - new_llms_count = 0 - sync_products_count = 0 - new_llms_list = [] + provider_groups = {} + current_model_set = set() for category_list in result: for item in category_list.get('llms', []): provider_orgcode = item.get('providerid') - provider_name = item.get('provider_name') or item.get('providerid') or '模型供应商' - if not provider_orgcode: + model = item.get('model') + if not provider_orgcode or not model: continue - - provider_ns = { - 'orgid': sync_orgid, - 'orgcode': provider_orgcode, - 'orgname': provider_name, - 'name': provider_name, - 'description': provider_name, - 'appid': item.get('upappid') or '', - 'providerkey': provider_orgcode, - 'state': '1', - 'start_date': '2026-07-01', - 'end_date': '9999-12-31', - 'settle_flowid': '0', - 'salemode': '0', - 'settle_mode': '3', - 'settle_datep': 'M00', - 'contactor': '连战', - 'contactor_phone': '13800138000', - 'emailaddress': 'lianzhan@cntoai.com', - 'address': '北京市海淀区', - 'org_type': '4' - } - provider_res = await path_call('../provider/providerAdd.dspy', provider_ns) - if not provider_res.get('status'): - return { - 'status': False, - 'msg': '供应商同步失败, %s, %s' % (provider_name, provider_res.get('msg')) - } - provider_data = provider_res.get('data') or {} - provider_orgid = provider_data.get('provider_orgid') or provider_data.get('orgid') - protocolid = provider_data.get('protocolid') - if not provider_orgid or not protocolid: - return { - 'status': False, - 'msg': '供应商同步失败, 未返回provider_orgid或protocolid, %s' % provider_name + item['catelogname'] = item.get('catelogname') or category_list.get('catelogname') + item['catelog_id'] = item.get('catelog_id') or item.get('catelogid') or category_list.get('catelogid') + current_model_set.add(model) + if provider_orgcode not in provider_groups: + provider_groups[provider_orgcode] = { + 'provider_name': item.get('provider_name') or provider_orgcode or '模型供应商', + 'items': [] } + elif item.get('provider_name'): + provider_groups[provider_orgcode]['provider_name'] = item.get('provider_name') + provider_groups[provider_orgcode]['items'].append({ + 'item': item, + 'category': category_list + }) - product_ns = { - 'providerid': provider_orgid, - 'providerpid': item.get('id'), - 'protocolid': protocolid, - 'name': item.get('name'), - 'product_code': item.get('model'), - 'reseller_orgid': sync_orgid, - 'label': item.get('catelogname') or category_list.get('catelogname'), - 'classify': 'A', - 'ptype': '21', - 'unit': 'hour', - 'discount': '1', - 'product_area': '北京' - } - product_res = await path_call('../product/productAdd.dspy', product_ns) - if not product_res.get('status'): - return { - 'status': False, - 'msg': '产品同步失败, %s, %s' % (item.get('name'), product_res.get('msg')) - } - sync_products_count += 1 + provider_count = 0 + product_insert_count = 0 + product_update_count = 0 + llm_insert_count = 0 + llm_update_count = 0 + synced_llms_list = [] - # 查找数据库中是否已经存在，不存在就插入 - exist_llm = await sor.R('llm', {'model': item.get('model')}) - if exist_llm: - # 更新 - exits_id = exist_llm[0].get('id') - await sor.U('llm', {'id': exits_id}, item) - continue - - new_llms = { - 'id': item.get('id'), - 'name': item.get('name'), - 'model': item.get('model'), - 'description': item.get('description'), - 'iconid': item.get('iconid'), - 'upappid': item.get('upappid'), - 'providerid': item.get('providerid'), - 'ownerid': item.get('ownerid'), - 'enabled_date': item.get('enabled_date'), - 'expired_date': item.get('expired_date'), - 'min_balance': item.get('min_balance'), - 'status': item.get('status'), - 'catelogname': item.get('catelogname'), - 'catelog_id': item.get('catelog_id'), - 'apiname': item.get('apiname'), - 'query_apiname': item.get('query_apiname'), - 'query_period': item.get('query_period'), - 'ppid': item.get('ppid'), - 'provider_name': item.get('provider_name'), - 'pricing_display': item.get('pricing_display'), - 'provider_iconurl': item.get('provider_iconurl') - } - new_llms_count += 1 - new_llms_list.append(new_llms.get('model')) - await sor.C('llm', new_llms) + for provider_orgcode, provider_group in provider_groups.items(): + provider_name = provider_group.get('provider_name') or provider_orgcode or '模型供应商' + provider_data = await ensure_provider( + sor, + provider_orgcode, + provider_name, + sync_orgid, + provider_group['items'][0]['item'] + ) + provider_count += 1 + + for model_info in provider_group['items']: + item = model_info['item'] + category_list = model_info['category'] + product_action = await sync_product( + sor, + item, + category_list, + provider_data['provider_orgid'], + provider_data['protocolid'], + sync_orgid + ) + if product_action == 'insert': + product_insert_count += 1 + else: + product_update_count += 1 + + llm_action = await upsert_llm(sor, item) + if llm_action == 'insert': + llm_insert_count += 1 + else: + llm_update_count += 1 + synced_llms_list.append(item.get('model')) + + llm_unpublished_count = 0 + exist_llm_li = await sor.R('llm', {}) + for exist_llm in exist_llm_li: + exist_model = exist_llm.get('model') + if exist_model and exist_model not in current_model_set and exist_llm.get('status') != 'unpublished': + await sor.U('llm', {'id': exist_llm.get('id')}, { + 'id': exist_llm.get('id'), + 'status': 'unpublished' + }) + llm_unpublished_count += 1 return { 'status': True, - 'msg': f"sync_llm_list同步模型成功，共同步{sync_products_count}个产品，共插入{new_llms_count}个模型，模型列表: {new_llms_list}" + 'msg': 'sync_llm_list同步模型成功', + 'data': { + 'provider_count': provider_count, + 'product_insert_count': product_insert_count, + 'product_update_count': product_update_count, + 'llm_insert_count': llm_insert_count, + 'llm_update_count': llm_update_count, + 'llm_unpublished_count': llm_unpublished_count, + 'model_list': synced_llms_list + } } except Exception as e: