Anony331
diff --git a/Collapse file
‎feapder/buffer/item_buffer.py‎
Copy file name to clipboardExpand all lines: feapder/buffer/item_buffer.py
+35-17Lines changed: 35 additions & 17 deletions b/Collapse file
‎feapder/buffer/item_buffer.py‎
Copy file name to clipboardExpand all lines: feapder/buffer/item_buffer.py
+35-17Lines changed: 35 additions & 17 deletions
diff --git a/Collapse file
‎feapder/setting.py‎
Copy file name to clipboardExpand all lines: feapder/setting.py
+4-2Lines changed: 4 additions & 2 deletions b/Collapse file
‎feapder/setting.py‎
Copy file name to clipboardExpand all lines: feapder/setting.py
+4-2Lines changed: 4 additions & 2 deletions
@@ -45,6 +45,9 @@ def __init__(self, redis_key, task_table=None):
 
             self._table_item = setting.TAB_ITEM
             self._table_request = setting.TAB_REQUSETS.format(redis_key=redis_key)
+            self._table_failed_items = setting.TAB_FAILED_ITEMS.format(
+                redis_key=redis_key
+            )
 
             self._item_tables = {
                 # 'xxx_item': {'tab_item': 'xxx:xxx_item'} # 记录item名与redis中item名对应关系
@@ -246,39 +249,37 @@ def __pick_items(self, items, is_update_item=False):
 
         return datas_dict
 
-    def __export_to_db(self, tab_item, datas, is_update=False, update_keys=()):
-        to_table = tools.get_info(tab_item, ":s_(.*?)_item$", fetch_one=True)
-
+    def __export_to_db(self, table, datas, is_update=False, update_keys=()):
         # 打点 校验
-        self.check_datas(table=to_table, datas=datas)
+        self.check_datas(table=table, datas=datas)
 
         for pipeline in self._pipelines:
             if is_update:
-                if to_table == self._task_table and not isinstance(
+                if table == self._task_table and not isinstance(
                     pipeline, MysqlPipeline
                 ):
                     continue
 
-                if not pipeline.update_items(to_table, datas, update_keys=update_keys):
+                if not pipeline.update_items(table, datas, update_keys=update_keys):
                     log.error(
-                        f"{pipeline.__class__.__name__} 更新数据失败. table: {to_table}  items: {datas}"
+                        f"{pipeline.__class__.__name__} 更新数据失败. table: {table}  items: {datas}"
                     )
                     return False
 
             else:
-                if not pipeline.save_items(to_table, datas):
+                if not pipeline.save_items(table, datas):
                     log.error(
-                        f"{pipeline.__class__.__name__} 保存数据失败. table: {to_table}  items: {datas}"
+                        f"{pipeline.__class__.__name__} 保存数据失败. table: {table}  items: {datas}"
                     )
                     return False
 
         # 若是任务表, 且上面的pipeline里没mysql，则需调用mysql更新任务
-        if not self._have_mysql_pipeline and is_update and to_table == self._task_table:
+        if not self._have_mysql_pipeline and is_update and table == self._task_table:
             if not self.mysql_pipeline.update_items(
-                to_table, datas, update_keys=update_keys
+                table, datas, update_keys=update_keys
             ):
                 log.error(
-                    f"{pipeline.__class__.__name__} 更新数据失败. table: {to_table}  items: {datas}"
+                    f"{self.mysql_pipeline.__class__.__name__} 更新数据失败. table: {table}  items: {datas}"
                 )
                 return False
 
@@ -299,36 +300,46 @@ def __add_item_to_db(
         update_items_dict = self.__pick_items(update_items, is_update_item=True)
 
         # item批量入库
+        failed_items = {}
         while items_dict:
             tab_item, datas = items_dict.popitem()
+            table = tools.get_info(tab_item, ":s_(.*?)_item$", fetch_one=True)
 
             log.debug(
                 """
                 -------------- item 批量入库 --------------
                 表名: %s
                 datas: %s
                     """
-                % (tab_item, tools.dumps_json(datas, indent=16))
+                % (table, tools.dumps_json(datas, indent=16))
             )
 
-            export_success = self.__export_to_db(tab_item, datas)
+            export_success = self.__export_to_db(table, datas)
+            if not export_success:
+                failed_items["add"] = {"table": table, "datas": datas}
+                break
 
         # 执行批量update
         while update_items_dict:
             tab_item, datas = update_items_dict.popitem()
+            table = tools.get_info(tab_item, ":s_(.*?)_item$", fetch_one=True)
+
             log.debug(
                 """
                 -------------- item 批量更新 --------------
                 表名: %s
                 datas: %s
                     """
-                % (tab_item, tools.dumps_json(datas, indent=16))
+                % (table, tools.dumps_json(datas, indent=16))
             )
 
             update_keys = self._item_update_keys.get(tab_item)
             export_success = self.__export_to_db(
-                tab_item, datas, is_update=True, update_keys=update_keys
+                table, datas, is_update=True, update_keys=update_keys
             )
+            if not export_success:
+                failed_items["update"] = {"table": table, "datas": datas}
+                break
 
         if export_success:
             # 执行回调
@@ -361,10 +372,17 @@ def __add_item_to_db(
                 )
 
             if self.export_falied_times > setting.EXPORT_DATA_MAX_RETRY_TIMES:
+                failed_items["requests"] = requests
+                self.redis_db.sadd(self._table_failed_items, failed_items)
+
                 # 删除做过的request
                 if requests:
                     self.redis_db.zrem(self._table_request, requests)
-                log.error("入库超过最大重试次数，不再重试")
+                log.error(
+                    "入库超过最大重试次数，不再重试, 数据记录到redis, items:\n {}".format(
+                        tools.dumps_json(failed_items)
+                    )
+                )
             else:
                 tip = ["入库不成功"]
                 if callbacks:
 
@@ -7,6 +7,8 @@
 TAB_REQUSETS = "{redis_key}:z_requsets"
 # 任务失败模板
 TAB_FAILED_REQUSETS = "{redis_key}:z_failed_requsets"
+# 数据保存失败模板
+TAB_FAILED_ITEMS = "{redis_key}:s_failed_items"
 # 爬虫状态表模版
 TAB_SPIDER_STATUS = "{redis_key}:z_spider_status"
 # item 表模版
@@ -41,8 +43,8 @@
     "feapder.pipelines.mysql_pipeline.MysqlPipeline",
     # "feapder.pipelines.mongo_pipeline.MongoPipeline",
 ]
-EXPORT_DATA_MAX_FAILED_TIMES = 10 # 导出数据时最大的失败次数，包括保存和更新，超过这个次数报警
-EXPORT_DATA_MAX_RETRY_TIMES = 10 # 导出数据时最大的重试次数，包括保存和更新，超过这个次数则放弃重试
+EXPORT_DATA_MAX_FAILED_TIMES = 10  # 导出数据时最大的失败次数，包括保存和更新，超过这个次数报警
+EXPORT_DATA_MAX_RETRY_TIMES = 10  # 导出数据时最大的重试次数，包括保存和更新，超过这个次数则放弃重试
 
 # 爬虫相关
 # COLLECTOR