Как я могу запустить импорт Mongodb после завершения Scrapy Spider? - PullRequest
Новая
       9

Как я могу запустить импорт Mongodb после завершения Scrapy Spider?

0 голосов
/ 22 сентября 2018

Я использую Scrapy, библиотеку Python, для очистки веб-сайтов и создания выходных файлов JSON через равные промежутки времени.Для эффективности я хочу сделать массовые апсеты этих json-файлов в Mongodb после завершения каждого паука.

Я считаю, что могу сделать upsert следующим образом: 

mongoimport -c <collection> -d <db> --mode merge --file test.json

Однако яХотите знать, каков наилучший способ вызвать этот импорт, когда паук завершил?И как?

Я надеялся, что смогу использовать метод close_spider, описанный здесь: https://doc.scrapy.org/en/latest/topics/item-pipeline.html#writing-your-own-item-pipeline

Однако, поиграв с ним, я обнаружил, что файл json только создается, а не записывается.когда внутри этого метода.

Было бы неплохо, если бы у меня был какой-то способ прослушать новый файл в определенном каталоге, а затем выполнить приведенный выше оператор импорта.

Возможно, все это можно сделать в скрипте bash? S

Ответы [ 3 ]

0 голосов
/ 23 сентября 2018

Этот метод работает для меня (в вашем файле паука): 

import os
from scrapy import signals
from scrapy.xlib.pydispatch import dispatcher

class MySpider(scrapy.Spider):

    def __init__(self):
        dispatcher.connect(self.spider_closed, signals.spider_closed)

    def spider_closed(self, spider):
        os.system("your_command")
0 голосов
/ 23 сентября 2018

Одним из решений является использование pyinotify для просмотра файла в выбранном каталоге.Я получил идею от здесь и адаптировал ее для выполнения оператора импорта mongo. 

class MyEventHandler(pyinotify.ProcessEvent):

    def process_IN_ACCESS(self, event):
        print("ACCESS event:", event.pathname)

    def process_IN_ATTRIB(self, event):
        print("ATTRIB event:", event.pathname)

    def process_IN_CLOSE_NOWRITE(self, event):
        print("CLOSE_NOWRITE event:", event.pathname)

    def process_IN_CLOSE_WRITE(self, event):
        print("CLOSE_WRITE event:", event.pathname)
        result = os.system('mongoimport -c kray4 -d kray4 --mode merge --file /home/kevin/PycharmProjects/spider/krawler/output/test.json')
        print("Result: " + str(result))

    def process_IN_CREATE(self, event):
        print("CREATE event:", event.pathname)

    def process_IN_DELETE(self, event):
        print("DELETE event:", event.pathname)

    def process_IN_MODIFY(self, event):
        print("MODIFY event:", event.pathname)

    def process_IN_OPEN(self, event):
        print("OPEN event:", event.pathname)


def main():
    # watch manager
    wm = pyinotify.WatchManager()

    wm.add_watch('/home/kevin/PycharmProjects/spider/krawler/output/test.json', pyinotify.ALL_EVENTS, rec=True)

    # event handler
    eh = MyEventHandler()

    # notifier
    notifier = pyinotify.Notifier(wm, eh)
    #command = 'echo 1 > /proc/sys/net/ipv4/ip_forward'
    notifier.loop()

if __name__ == '__main__':
    main()
0 голосов
/ 22 сентября 2018

Вы можете писать предметы напрямую в Mongo, используя Item Pipelines .Взгляните на этот пример из документации Scrapy :

Запись элементов в MongoDB

В этом примере мы будем записывать элементы в MongoDB с помощью pymongo.Адрес MongoDB и имя базы данных указываются в настройках Scrapy;Коллекция MongoDB названа в честь класса элемента.

Основная задача этого примера - показать, как использовать метод from_crawler () и как правильно очищать ресурсы .:

import pymongo

class MongoPipeline(object):

    collection_name = 'scrapy_items'

    def __init__(self, mongo_uri, mongo_db):
        self.mongo_uri = mongo_uri
        self.mongo_db = mongo_db

    @classmethod
    def from_crawler(cls, crawler):
        return cls(
            mongo_uri=crawler.settings.get('MONGO_URI'),
            mongo_db=crawler.settings.get('MONGO_DATABASE', 'items')
        )

    def open_spider(self, spider):
        self.client = pymongo.MongoClient(self.mongo_uri)
        self.db = self.client[self.mongo_db]

    def close_spider(self, spider):
        self.client.close()

    def process_item(self, item, spider):
        self.db[self.collection_name].insert_one(dict(item))
        return item
...