refactor: import engine calls resource module's load_product_category_product

- core.py: import_categories_and_products() now takes resource_module as first arg,
  uses importlib to load {resource_module}.init, calls load_product_category_product(parent_category_id)
- Resource module owns the full import logic (read source + write product_management)
- __init__.py: updated wrapper signature
- import_category_products.dspy: updated call

product_management/__init__.py

@@ -67,11 +67,11 @@ async def set_operator_config(category_id, config_name, config_json, user_id=Non
     return await manager.set_operator_config(category_id, config_name, config_json, user_id, org_id)
 
 
-async def import_categories_and_products(org_id, parent_category_id, user_id, import_data):
+async def import_categories_and_products(resource_module, org_id, parent_category_id, user_id):
-    """Generic import engine for sub-categories and products."""
+    """Import sub-categories and products from a resource module."""
     manager = get_manager()
     return await manager.import_categories_and_products(
-        org_id, parent_category_id, user_id, import_data
+        resource_module, org_id, parent_category_id, user_id
     )
 
 

product_management/core.py

@@ -1093,125 +1093,32 @@ class ProductManager:
 
     # ─── Product Import from Resource Modules ───
 
-    async def import_categories_and_products(self, org_id, parent_category_id,
+    async def import_categories_and_products(self, resource_module, org_id,
-                                              user_id, import_data):
+                                              parent_category_id, user_id):
-        """Generic import engine: create sub-categories and products from structured data.
+        """Import sub-categories and products from a resource module.
 
-        import_data format:
+        Dynamically imports the resource module and calls its
-        {
+        load_product_category_product(parent_category_id) function,
-            'categories': [
+        which is responsible for both reading source data and writing
-                {'name': '子类别名', 'source_id': 'ext_id', 'parent_source_id': None,
+        sub-categories + products under the given parent_category_id.
-                 'sort_order': 1, 'description': '', 'has_product': '1', 'product_type': 'llm_model'},
-                ...
-            ],
-            'products': [
-                {'product_code': 'code1', 'product_name': '产品1',
-                 'category_source_id': 'ext_id', 'product_type': 'llm_model',
-                 'brief_intro': '', 'price': 0, 'currency': 'CNY', 'sort_order': 0},
-                ...
-            ]
-        }
-
-        source_id is the external module's ID, used to avoid duplicates.
-        Products link to categories via category_source_id matching a category's source_id.
         """
-        import time
+        import importlib
-        dbname = self._get_dbname()
-        now = time.strftime('%Y-%m-%d %H:%M:%S')
-        created_cats = 0
-        skipped_cats = 0
-        created_prods = 0
-        skipped_prods = 0
 
-        async with DBPools().sqlorContext(dbname) as sor:
+        try:
-            # Phase 1: Create sub-categories under parent_category_id
+            mod = importlib.import_module(f'{resource_module}.init')
-            # Map source_id -> new category id
+            load_fn = getattr(mod, 'load_product_category_product', None)
-            source_to_id = {}
+        except Exception as e:
+            return {'success': False, 'error': f'无法加载资源模块 "{resource_module}": {e}'}
 
-            for cat in import_data.get('categories', []):
+        if load_fn is None:
-                source_id = cat.get('source_id', '')
+            return {'success': False, 'error': f'资源模块 "{resource_module}" 未提供 load_product_category_product 函数'}
-                # Check if already imported (by name + parent)
-                existing = await sor.sqlExe(
-                    """SELECT id FROM product_category
-                       WHERE name = ${name}$ AND parent_id = ${parent_id}$ AND org_id = ${org_id}$""",
-                    {'name': cat['name'], 'parent_id': parent_category_id, 'org_id': org_id}
-                )
-                if existing:
-                    source_to_id[source_id] = existing[0]['id']
-                    skipped_cats += 1
-                    continue
 
-                new_id = getID()
+        try:
-                source_to_id[source_id] = new_id
+            result = await load_fn(parent_category_id)
-                # Determine parent: if parent_source_id is set, use that mapping
+        except Exception as e:
-                psrc = cat.get('parent_source_id')
+            return {'success': False, 'error': f'调用 {resource_module}.load_product_category_product() 失败: {e}'}
-                actual_parent = source_to_id.get(psrc, parent_category_id) if psrc else parent_category_id
 
-                cat_data = {
+        if not result:
-                    'id': new_id,
+            return {'success': False, 'error': f'资源模块 "{resource_module}" 未返回导入数据'}
-                    'parent_id': actual_parent,
-                    'name': cat['name'],
-                    'description': cat.get('description', ''),
-                    'has_product': cat.get('has_product', '1'),
-                    'product_type': cat.get('product_type', ''),
-                    'product_type_title': cat.get('product_type_title', ''),
-                    'sort_order': str(cat.get('sort_order', 0)),
-                    'icon': cat.get('icon', ''),
-                    'status': '1',
-                    'resource_module': cat.get('resource_module', ''),
-                    'org_id': org_id,
-                    'created_by': user_id,
-                    'created_at': now,
-                    'updated_at': now
-                }
-                await sor.C('product_category', cat_data)
-                created_cats += 1
 
-            # Phase 2: Create products under the mapped categories
+        return result
-            for prod in import_data.get('products', []):
-                cat_source_id = prod.get('category_source_id', '')
-                target_cat_id = source_to_id.get(cat_source_id)
-                if not target_cat_id:
-                    skipped_prods += 1
-                    continue
-
-                # Check duplicate by product_code + org
-                existing_prod = await sor.sqlExe(
-                    """SELECT id FROM product
-                       WHERE product_code = ${code}$ AND org_id = ${org_id}$""",
-                    {'code': prod['product_code'], 'org_id': org_id}
-                )
-                if existing_prod:
-                    skipped_prods += 1
-                    continue
-
-                prod_id = getID()
-                prod_data = {
-                    'id': prod_id,
-                    'category_id': target_cat_id,
-                    'product_code': prod['product_code'],
-                    'product_name': prod['product_name'],
-                    'product_type': prod.get('product_type', ''),
-                    'brief_intro': prod.get('brief_intro', ''),
-                    'detail_intro': prod.get('detail_intro', ''),
-                    'extra_json': prod.get('extra_json', ''),
-                    'enabled_date': prod.get('enabled_date'),
-                    'expired_date': prod.get('expired_date'),
-                    'status': prod.get('status', '1'),
-                    'price_type': prod.get('price_type', '1'),
-                    'price': str(prod.get('price', 0)),
-                    'currency': prod.get('currency', 'CNY'),
-                    'sort_order': str(prod.get('sort_order', 0)),
-                    'org_id': org_id,
-                    'created_by': user_id,
-                    'created_at': now,
-                    'updated_at': now
-                }
-                await sor.C('product', prod_data)
-                created_prods += 1
-
-        return {
-            'success': True,
-            'message': f'导入完成: 新增 {created_cats} 个子类别, {created_prods} 个产品; '
-                       f'跳过 {skipped_cats} 个已存在类别, {skipped_prods} 个已存在产品'
-        }

wwwroot/api/import_category_products.dspy

@@ -23,15 +23,14 @@ try:
     if not resource_module:
         raise ValueError('该类别未配置资源模块，请先在类别编辑中设置资源模块')
 
-    # Look up the registered import function by module name
+    # Call the import engine
     env = request._run_ns
-    import_fn_name = f'import_products_from_{resource_module}'
+    import_result = await env.import_categories_and_products(
-    import_fn = getattr(env, import_fn_name, None)
+        resource_module=resource_module,
-    if import_fn is None:
+        org_id=org_id,
-        raise ValueError(f'资源模块 "{resource_module}" 未注册产品导入功能')
+        parent_category_id=category_id,
-
+        user_id=user_id
-    # Call the resource module's import function
+    )
-    import_result = await import_fn(org_id=org_id, parent_category_id=category_id, user_id=user_id)
 
     if import_result.get('success'):
         msg = import_result.get('message', '导入完成')