Merge branch 'ShellMonster-feature/csv-pipeline'

Author: Boris-code

docs/source_code/Item.md

@@ -102,6 +102,26 @@ class SpiderDataItem(Item):
         self.title = self.title.strip()
 ```
 
+## 指定入库使用的pipelines
+
+```python
+
+from feapder import Item
+from feapder.pipelines.csv_pipeline import CsvPipeline
+
+
+class SpiderDataItem(Item):
+
+    __pipelines__ = [CsvPipeline()]
+
+    def __init__(self, *args, **kwargs):
+        # self.id = None
+        self.title = None
+```
+
+使用__pipelines__指定后，该item只会流经指定的pipelines处理
+
+
 ## 更新数据
 
 采集过程中，往往会有些数据漏采或解析出错，如果我们想更新已入库的数据，可将Item转为UpdateItem

docs/source_code/pipeline.md

@@ -2,11 +2,26 @@
 
 Pipeline是数据入库时流经的管道，用户可自定义，以便对接其他数据库。
 
-框架已内置mysql及mongo管道，其他管道作为扩展方式提供，可从[feapder_pipelines](https://github.com/Boris-code/feapder_pipelines)项目中按需安装
+框架已内置mysql、mongo、csv管道，其他管道作为扩展方式提供，可从[feapder_pipelines](https://github.com/Boris-code/feapder_pipelines)项目中按需安装
 
 项目地址：https://github.com/Boris-code/feapder_pipelines
 
-## 使用方式
+## 选择内置的pipeline
+
+在配置文件 `setting.py` 中的 `ITEM_PIPELINES` 中启用：
+
+```python
+ITEM_PIPELINES = [
+    "feapder.pipelines.mysql_pipeline.MysqlPipeline",
+    # "feapder.pipelines.mongo_pipeline.MongoPipeline",
+    # "feapder.pipelines.csv_pipeline.CsvPipeline",
+    # "feapder.pipelines.console_pipeline.ConsolePipeline",
+]
+```
+
+然后 爬虫中`yield`的`item`会流经选择的pipeline自动存储
+
+## 自定义pipeline
 
 注：item会被聚合成多条一起流经pipeline，方便批量入库
 

feapder/buffer/item_buffer.py

@@ -52,14 +52,18 @@ def __init__(self, redis_key, task_table=None):
                 # 'table_name': ['id', 'name'...] # 缓存table_name与__update_key__的关系
             }
 
+            self._item_pipelines = {
+                # 'table_name': ['pipeline1', 'pipeline2'] # 缓存table_name与pipelines的关系
+            }
+
             self._pipelines = self.load_pipelines()
 
             self._have_mysql_pipeline = MYSQL_PIPELINE_PATH in setting.ITEM_PIPELINES
             self._mysql_pipeline = None
 
             if setting.ITEM_FILTER_ENABLE and not self.__class__.dedup:
                 if setting.ITEM_FILTER_SETTING.get(
-                    "filter_type"
+                        "filter_type"
                 ) == Dedup.BloomFilter or setting.ITEM_FILTER_SETTING.get("name"):
                     self.__class__.dedup = Dedup(
                         to_md5=False, **setting.ITEM_FILTER_SETTING
@@ -217,7 +221,7 @@ def __pick_items(self, items, is_update_item=False):
         将每个表之间的数据分开 拆分后 原items为空
         @param items:
         @param is_update_item:
-        @return:
+        @return: 表名与数据的字典
         """
         datas_dict = {
             # 'table_name': [{}, {}]
@@ -232,22 +236,24 @@ def __pick_items(self, items, is_update_item=False):
             if not table_name:
                 table_name = item.table_name
                 self._item_tables[item_name] = table_name
+                self._item_pipelines[table_name] = item.pipelines
+
+            if is_update_item and table_name not in self._item_update_keys:
+                self._item_update_keys[table_name] = item.update_key
 
             if table_name not in datas_dict:
                 datas_dict[table_name] = []
 
             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
 
-    def __export_to_db(self, table, datas, is_update=False, update_keys=()):
-        for pipeline in self._pipelines:
+    def __export_to_db(self, table, datas, is_update=False, update_keys=(), used_pipelines=None):
+        pipelines = used_pipelines or self._pipelines  # 优先采用指定的pipelines
+        for pipeline in pipelines:
             if is_update:
                 if table == self._task_table and not isinstance(
-                    pipeline, MysqlPipeline
+                        pipeline, MysqlPipeline
                 ):
                     continue
 
@@ -267,7 +273,7 @@ def __export_to_db(self, table, datas, is_update=False, update_keys=()):
         # 若是任务表, 且上面的pipeline里没mysql，则需调用mysql更新任务
         if not self._have_mysql_pipeline and is_update and table == self._task_table:
             if not self.mysql_pipeline.update_items(
-                table, datas, update_keys=update_keys
+                    table, datas, update_keys=update_keys
             ):
                 log.error(
                     f"{self.mysql_pipeline.__class__.__name__} 更新数据失败. table: {table}  items: {datas}"
@@ -278,7 +284,7 @@ def __export_to_db(self, table, datas, is_update=False, update_keys=()):
         return True
 
     def __add_item_to_db(
-        self, items, update_items, requests, callbacks, items_fingerprints
+            self, items, update_items, requests, callbacks, items_fingerprints
     ):
         export_success = True
         self._is_adding_to_db = True
@@ -287,14 +293,15 @@ def __add_item_to_db(
         if setting.ITEM_FILTER_ENABLE:
             items, items_fingerprints = self.__dedup_items(items, items_fingerprints)
 
-        # 分捡
+        # 分捡（返回值包含 pipelines_dict）
         items_dict = self.__pick_items(items)
         update_items_dict = self.__pick_items(update_items, is_update_item=True)
 
         # item批量入库
         failed_items = {"add": [], "update": [], "requests": []}
         while items_dict:
             table, datas = items_dict.popitem()
+            used_pipelines = self._item_pipelines.get(table)
 
             log.debug(
                 """
@@ -305,13 +312,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, used_pipelines=used_pipelines):
                 export_success = False
                 failed_items["add"].append({"table": table, "datas": datas})
 
         # 执行批量update
         while update_items_dict:
             table, datas = update_items_dict.popitem()
+            used_pipelines = self._item_pipelines.get(table)
 
             log.debug(
                 """
@@ -324,7 +332,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, used_pipelines=used_pipelines
             ):
                 export_success = False
                 failed_items["update"].append(

feapder/db/mysqldb.py

@@ -202,7 +202,7 @@ def convert(col):
                 if isinstance(col, (datetime.date, datetime.time)):
                     return str(col)
                 elif isinstance(col, str) and (
-                    col.startswith("{") or col.startswith("[")
+                        col.startswith("{") or col.startswith("[")
                 ):
                     try:
                         # col = self.unescape_string(col)

feapder/network/item.py

@@ -9,6 +9,7 @@
 """
 
 import re
+from typing import List
 
 import feapder.utils.tools as tools
 
@@ -20,12 +21,14 @@ 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)
 
 
 class Item(metaclass=ItemMetaclass):
-    __unique_key__ = []
+    __unique_key__: List = []
+    __pipelines__: List = None
 
     def __init__(self, **kwargs):
         self.__dict__ = kwargs
@@ -64,11 +67,12 @@ def to_dict(self):
         propertys = {}
         for key, value in self.__dict__.items():
             if key not in (
-                "__name__",
-                "__table_name__",
-                "__name_underline__",
-                "__update_key__",
-                "__unique_key__",
+                    "__name__",
+                    "__table_name__",
+                    "__name_underline__",
+                    "__update_key__",
+                    "__unique_key__",
+                    "__pipelines__",
             ):
                 if key.startswith(f"_{self.__class__.__name__}"):
                     key = key.replace(f"_{self.__class__.__name__}", "")
@@ -123,6 +127,17 @@ def unique_key(self, keys):
         else:
             self.__unique_key__ = (keys,)
 
+    @property
+    def pipelines(self):
+        return self.__pipelines__ or self.__class__.__pipelines__
+
+    @pipelines.setter
+    def pipelines(self, pipelines):
+        if isinstance(pipelines, (tuple, list)):
+            self.__pipelines__ = pipelines
+        else:
+            self.__pipelines__ = (pipelines,)
+
     @property
     def fingerprint(self):
         args = []