如果为空,则使用 headers 进行 Scrapy 导出
Scrapy export with headers if empty
据我所知,没有人问过这个问题,我完全无法解决这个问题。我有一个蜘蛛,它有时不会 return 任何结果(要么不存在结果,要么由于 robots.txt 文件而无法抓取网站),这会导致一个空的,无标题,csv 文件。我有一个机器人想要拾取这个文件,所以当它是空的时候机器人没有意识到它已经完成并且在任何情况下没有 headers 无法理解它。
我想要的是每次都输出headers的csv文件,即使没有结果。我试过使用 json 代替,但遇到了同样的问题 - 如果出现错误或没有结果,则文件为空。
我很乐意在 spider 关闭时调用一些东西(无论出于何种原因,甚至是由于错误 url 而导致的初始化错误)并向文件写入一些东西。
谢谢。
我通过修改我的项目管道而不是在命令行中使用提要导出器解决了这个问题。这允许我使用 close_spider 来写,如果没有结果, header.
如果我遗漏了什么,我当然欢迎任何改进。
管道代码:
from scrapy.exceptions import DropItem
from scrapy.exporters import CsvItemExporter
import csv
from Generic.items import GenericItem
class GenericPipeline:
def __init__(self):
self.emails_seen = set()
self.files = {}
def open_spider(self, spider):
self.file = open("results.csv", 'wb')
self.exporter = CsvItemExporter(self.file)
self.exporter.start_exporting()
def close_spider(self, spider):
if not self.emails_seen:
header = GenericItem()
header["email"] = "None Found"
self.exporter.export_item(header)
self.exporter.finish_exporting()
self.file.close()
def process_item(self, item, spider):
if item["email"] in self.emails_seen:
raise DropItem(f"Duplicate item found: {item!r}")
else:
self.emails_seen.add(item["email"])
self.exporter.export_item(item)
return item
据我所知,没有人问过这个问题,我完全无法解决这个问题。我有一个蜘蛛,它有时不会 return 任何结果(要么不存在结果,要么由于 robots.txt 文件而无法抓取网站),这会导致一个空的,无标题,csv 文件。我有一个机器人想要拾取这个文件,所以当它是空的时候机器人没有意识到它已经完成并且在任何情况下没有 headers 无法理解它。
我想要的是每次都输出headers的csv文件,即使没有结果。我试过使用 json 代替,但遇到了同样的问题 - 如果出现错误或没有结果,则文件为空。
我很乐意在 spider 关闭时调用一些东西(无论出于何种原因,甚至是由于错误 url 而导致的初始化错误)并向文件写入一些东西。
谢谢。
我通过修改我的项目管道而不是在命令行中使用提要导出器解决了这个问题。这允许我使用 close_spider 来写,如果没有结果, header.
如果我遗漏了什么,我当然欢迎任何改进。
管道代码:
from scrapy.exceptions import DropItem
from scrapy.exporters import CsvItemExporter
import csv
from Generic.items import GenericItem
class GenericPipeline:
def __init__(self):
self.emails_seen = set()
self.files = {}
def open_spider(self, spider):
self.file = open("results.csv", 'wb')
self.exporter = CsvItemExporter(self.file)
self.exporter.start_exporting()
def close_spider(self, spider):
if not self.emails_seen:
header = GenericItem()
header["email"] = "None Found"
self.exporter.export_item(header)
self.exporter.finish_exporting()
self.file.close()
def process_item(self, item, spider):
if item["email"] in self.emails_seen:
raise DropItem(f"Duplicate item found: {item!r}")
else:
self.emails_seen.add(item["email"])
self.exporter.export_item(item)
return item