From 811de2a863f76fd9565e7d10525f414d1c68509a Mon Sep 17 00:00:00 2001 From: Hermes Agent Date: Tue, 23 Jun 2026 15:09:09 +0800 Subject: [PATCH] feat: add resource module interface dispatcher to ProductManager - _get_product_interface: resolve product -> resource_module -> interface dict - get_product_display_info: delegate to resource module pricing display - check_product_availability: delegate to resource module availability check - check_product_consumable: delegate to resource module pre-consumption check - execute_product / execute_product_stream: delegate to resource module service execution - calculate_product_cost: delegate to resource module cost calculation - Register all dispatchers on ServerEnv in load_product_management() --- product_management/__init__.py | 43 ++++++++ product_management/core.py | 179 +++++++++++++++++++++++++++++++++ 2 files changed, 222 insertions(+) diff --git a/product_management/__init__.py b/product_management/__init__.py index 1d9d3ec..159a3ff 100644 --- a/product_management/__init__.py +++ b/product_management/__init__.py @@ -75,6 +75,42 @@ async def import_categories_and_products(resource_module, org_id, parent_categor ) +async def get_product_display_info(product_id=None, product_code=None): + """Get product pricing display info via resource module interface.""" + manager = get_manager() + return await manager.get_product_display_info(product_id, product_code) + + +async def check_product_availability(product_id=None, product_code=None, user_org_id=None): + """Check product availability via resource module interface.""" + manager = get_manager() + return await manager.check_product_availability(product_id, product_code, user_org_id) + + +async def check_product_consumable(product_id=None, product_code=None, user_id=None, user_org_id=None): + """Pre-consumption check via resource module interface.""" + manager = get_manager() + return await manager.check_product_consumable(product_id, product_code, user_id, user_org_id) + + +async def execute_product(product_id=None, product_code=None, request_data=None, user_id=None, user_org_id=None): + """Execute product service via resource module interface.""" + manager = get_manager() + return await manager.execute_product(product_id, product_code, request_data, user_id, user_org_id) + + +async def execute_product_stream(product_id=None, product_code=None, request_data=None, user_id=None, user_org_id=None): + """Execute product service (streaming) via resource module interface.""" + manager = get_manager() + return await manager.execute_product_stream(product_id, product_code, request_data, user_id, user_org_id) + + +async def calculate_product_cost(product_id=None, product_code=None, usage_data=None, user_org_id=None): + """Calculate product usage cost via resource module interface.""" + manager = get_manager() + return await manager.calculate_product_cost(product_id, product_code, usage_data, user_org_id) + + def load_product_management(): """Register all functions with ServerEnv so they can be called from .ui/.dspy files.""" env = ServerEnv() @@ -87,4 +123,11 @@ def load_product_management(): env.get_operator_config = get_operator_config env.set_operator_config = set_operator_config env.import_categories_and_products = import_categories_and_products + # Resource module interface dispatchers + env.get_product_display_info = get_product_display_info + env.check_product_availability = check_product_availability + env.check_product_consumable = check_product_consumable + env.execute_product = execute_product + env.execute_product_stream = execute_product_stream + env.calculate_product_cost = calculate_product_cost return True diff --git a/product_management/core.py b/product_management/core.py index 3c1cd5a..d4ce5d0 100644 --- a/product_management/core.py +++ b/product_management/core.py @@ -1122,3 +1122,182 @@ class ProductManager: return {'success': False, 'error': f'资源模块 "{resource_module}" 未返回导入数据'} return result + + # ─── Resource Module Interface Dispatcher ─── + + async def _get_product_interface(self, product_id=None, product_code=None): + """Resolve product to its resource module's interface dict. + + Returns: (interface_dict, product_record, error_message) + """ + dbname = self._get_dbname() + conditions = ["status = '1'"] + params = {} + if product_id: + conditions.append("id = ${pid}$") + params['pid'] = product_id + elif product_code: + conditions.append("product_code = ${code}$") + params['code'] = product_code + else: + return None, None, '需要指定 product_id 或 product_code' + + where = " AND ".join(conditions) + async with DBPools().sqlorContext(dbname) as sor: + rows = await sor.sqlExe( + f"SELECT * FROM product WHERE {where}", params) + if not rows: + return None, None, '产品不存在或已下线' + product = dict(rows[0]) + + # Get resource_module from category + cat_id = product.get('category_id') + if cat_id: + async with DBPools().sqlorContext(dbname) as sor: + cat_rows = await sor.sqlExe( + "SELECT resource_module FROM product_category WHERE id = ${cid}$", + {'cid': cat_id}) + if cat_rows and cat_rows[0].get('resource_module'): + resource_module = cat_rows[0]['resource_module'] + else: + return None, product, '产品类别未绑定资源模块' + else: + return None, product, '产品无类别信息' + + # Get interface from ServerEnv or importlib + env = ServerEnv() + iface = getattr(env, 'product_interface', None) + if iface and iface.get('module_name') == resource_module: + return iface, product, None + + # Fallback: dynamic import + import importlib + try: + mod = importlib.import_module(f'{resource_module}.init') + # Trigger load if not loaded yet + load_fn_name = f'load_{resource_module}' + load_fn = getattr(mod, load_fn_name, None) + if load_fn: + load_fn() + iface = getattr(env, 'product_interface', None) + if iface and iface.get('module_name') == resource_module: + return iface, product, None + except Exception as e: + return None, product, f'加载资源模块 {resource_module} 失败: {e}' + + return None, product, f'资源模块 {resource_module} 未注册 product_interface' + + async def get_product_display_info(self, product_id=None, product_code=None): + """获取产品定价展示信息（通过资源模块接口）。""" + iface, product, err = await self._get_product_interface( + product_id, product_code) + if err: + return {'success': False, 'message': err} + + fn = iface.get('get_product_display') + if not fn: + return {'success': False, 'message': '资源模块未实现 get_product_display'} + + return await fn(product['product_code']) + + async def check_product_availability(self, product_id=None, product_code=None, + user_org_id=None): + """检查产品可用性（通过资源模块接口）。""" + iface, product, err = await self._get_product_interface( + product_id, product_code) + if err: + return {'available': False, 'reason': err} + + fn = iface.get('check_product_availability') + if not fn: + return {'available': False, 'reason': '资源模块未实现 check_product_availability'} + + return await fn(product['product_code'], user_org_id) + + async def check_product_consumable(self, product_id=None, product_code=None, + user_id=None, user_org_id=None): + """消费前综合预检（通过资源模块接口）。""" + iface, product, err = await self._get_product_interface( + product_id, product_code) + if err: + return {'consumable': False, 'reason': err, + 'min_balance': 0, 'pricing_available': False} + + fn = iface.get('check_product_consumable') + if not fn: + return {'consumable': False, 'reason': '资源模块未实现 check_product_consumable', + 'min_balance': 0, 'pricing_available': False} + + if not user_id: + env = ServerEnv() + user_id = await env.get_user() + if not user_org_id: + user_org_id = self._get_current_org_id() + + return await fn(product['product_code'], user_id, user_org_id) + + async def execute_product(self, product_id=None, product_code=None, + request_data=None, user_id=None, user_org_id=None): + """执行产品服务（通过资源模块接口）。""" + iface, product, err = await self._get_product_interface( + product_id, product_code) + if err: + return {'success': False, 'message': err, 'status': 'FAILED'} + + fn = iface.get('execute_product_service') + if not fn: + return {'success': False, 'message': '资源模块未实现 execute_product_service', + 'status': 'FAILED'} + + if not user_id: + env = ServerEnv() + user_id = await env.get_user() + if not user_org_id: + user_org_id = self._get_current_org_id() + + return await fn(product['product_code'], user_id, user_org_id, + request_data or {}) + + async def execute_product_stream(self, product_id=None, product_code=None, + request_data=None, user_id=None, + user_org_id=None): + """执行产品服务-流式（通过资源模块接口，返回异步生成器）。""" + iface, product, err = await self._get_product_interface( + product_id, product_code) + if err: + async def _err_gen(): + yield {'chunk': None, 'usage_data': None, 'done': True, 'error': err} + return _err_gen() + + fn = iface.get('execute_product_service_stream') + if not fn: + async def _noimpl_gen(): + yield {'chunk': None, 'usage_data': None, 'done': True, + 'error': '资源模块未实现 execute_product_service_stream'} + return _noimpl_gen() + + if not user_id: + env = ServerEnv() + user_id = await env.get_user() + if not user_org_id: + user_org_id = self._get_current_org_id() + + return fn(product['product_code'], user_id, user_org_id, + request_data or {}) + + async def calculate_product_cost(self, product_id=None, product_code=None, + usage_data=None, user_org_id=None): + """计算消费费用（通过资源模块接口）。""" + iface, product, err = await self._get_product_interface( + product_id, product_code) + if err: + return {'success': False, 'message': err} + + fn = iface.get('calculate_product_cost') + if not fn: + return {'success': False, 'message': '资源模块未实现 calculate_product_cost'} + + if not user_org_id: + user_org_id = self._get_current_org_id() + + return await fn(product['product_code'], usage_data or {}, user_org_id)