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

@@ -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

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)