feat: Item 支持指定 Pipeline 路由

- 新增 Item.__pipelines__ 属性，允许 Item 指定流向哪些 Pipeline
- 支持大小写不敏感匹配（csv/CSV/Csv 都有效）
- 未指定时流向所有 Pipeline（保持向后兼容）
- 修改 ItemBuffer 逻辑，支持 Pipeline 过滤

使用示例：
class ProductItem(Item):
    table_name = 'product'
    __pipelines__ = ['csv']  # 只流向 CSV Pipeline

class UserItem(Item):
    table_name = 'user'
    __pipelines__ = ['mysql']  # 只流向 MySQL Pipeline

class OrderItem(Item):
    table_name = 'order'
    __pipelines__ = ['csv', 'MySQL']  # 流向两者，大小写不敏感

Author: ShellMonster

feapder/buffer/item_buffer.py

@@ -217,11 +217,14 @@ def __pick_items(self, items, is_update_item=False):
         将每个表之间的数据分开 拆分后 原items为空
         @param items:
         @param is_update_item:
-        @return:
+        @return: (datas_dict, pipelines_dict)
         """
         datas_dict = {
             # 'table_name': [{}, {}]
         }
+        pipelines_dict = {
+            # 'table_name': ['csv', 'mysql'] or None
+        }
 
         while items:
             item = items.pop(0)
@@ -235,16 +238,26 @@ def __pick_items(self, items, is_update_item=False):
 
             if table_name not in datas_dict:
                 datas_dict[table_name] = []
+                # 保存这个 table 的 pipelines 配置（只需保存一次）
+                pipelines_dict[table_name] = getattr(item, '__pipelines__', None)
 
             datas_dict[table_name].append(item.to_dict)
 
             if is_update_item and table_name not in self._item_update_keys:
                 self._item_update_keys[table_name] = item.update_key
 
-        return datas_dict
+        return datas_dict, pipelines_dict
 
-    def __export_to_db(self, table, datas, is_update=False, update_keys=()):
+    def __export_to_db(self, table, datas, is_update=False, update_keys=(), allowed_pipelines=None):
         for pipeline in self._pipelines:
+            # 如果 item 指定了 pipelines，检查是否匹配（忽略大小写）
+            if allowed_pipelines is not None:
+                pipeline_name = pipeline.__class__.__name__.replace("Pipeline", "").lower()
+                # 将用户指定的 pipeline 名称也转为小写进行比较
+                allowed_pipelines_lower = [p.lower() for p in allowed_pipelines]
+                if pipeline_name not in allowed_pipelines_lower:
+                    continue  # 跳过不匹配的 pipeline
+
             if is_update:
                 if table == self._task_table and not isinstance(
                     pipeline, MysqlPipeline
@@ -287,14 +300,15 @@ def __add_item_to_db(
         if setting.ITEM_FILTER_ENABLE:
             items, items_fingerprints = self.__dedup_items(items, items_fingerprints)
 
-        # 分捡
-        items_dict = self.__pick_items(items)
-        update_items_dict = self.__pick_items(update_items, is_update_item=True)
+        # 分捡（返回值包含 pipelines_dict）
+        items_dict, items_pipelines = self.__pick_items(items)
+        update_items_dict, update_pipelines = self.__pick_items(update_items, is_update_item=True)
 
         # item批量入库
         failed_items = {"add": [], "update": [], "requests": []}
         while items_dict:
             table, datas = items_dict.popitem()
+            allowed_pipelines = items_pipelines.get(table)
 
             log.debug(
                 """
@@ -305,13 +319,14 @@ def __add_item_to_db(
                 % (table, tools.dumps_json(datas, indent=16))
             )
 
-            if not self.__export_to_db(table, datas):
+            if not self.__export_to_db(table, datas, allowed_pipelines=allowed_pipelines):
                 export_success = False
                 failed_items["add"].append({"table": table, "datas": datas})
 
         # 执行批量update
         while update_items_dict:
             table, datas = update_items_dict.popitem()
+            allowed_pipelines = update_pipelines.get(table)
 
             log.debug(
                 """
@@ -324,7 +339,7 @@ def __add_item_to_db(
 
             update_keys = self._item_update_keys.get(table)
             if not self.__export_to_db(
-                table, datas, is_update=True, update_keys=update_keys
+                table, datas, is_update=True, update_keys=update_keys, allowed_pipelines=allowed_pipelines
             ):
                 export_success = False
                 failed_items["update"].append(

feapder/network/item.py

@@ -20,6 +20,7 @@ def __new__(cls, name, bases, attrs):
         attrs.setdefault("__name_underline__", None)
         attrs.setdefault("__update_key__", None)
         attrs.setdefault("__unique_key__", None)
+        attrs.setdefault("__pipelines__", None)
 
         return type.__new__(cls, name, bases, attrs)
 
@@ -69,6 +70,7 @@ def to_dict(self):
                 "__name_underline__",
                 "__update_key__",
                 "__unique_key__",
+                "__pipelines__",
             ):
                 if key.startswith(f"_{self.__class__.__name__}"):
                     key = key.replace(f"_{self.__class__.__name__}", "")
@@ -145,6 +147,7 @@ def to_UpdateItem(self):
 
 class UpdateItem(Item):
     __update_key__ = []
+    __pipelines__ = None
 
     def __init__(self, **kwargs):
         super(UpdateItem, self).__init__(**kwargs)