Пример запуска Scrapy отдельным скриптом, без создания проекта

Scrapy представяет собой серъёзный фреймворк для обхода сайтов и извлечения из них структурированной информации. Отличается большим диапазоном решаемых задач (сбор данных, мониторинг, автоматическое тестирование, итд) а также отличной скоростью (благодаря асинхронным вызовам). Использование scrapy подразумевает создание отдельного проекта, со своей заданной структурой и настройкой логики/методов обхода в отдельных python классах scrapy.Spider .
Но что если мы хотим использовать все плюшки большого фремворка, не создавая непосредственно проект? Скажем надо сделать маленькую подзадачу в другом проекте: залогиниться, перейти на заданную страницу и вытащить что-то из html кода?
Столкнувшись с данной задачей мне не удалось быстро найти решение на просторах интернета, поэтому опишу здесь что у меня получилось:

# Устанавливаем библиотеку
pip3 install scrapy

Сценарий следующий:
1. Заходим на стартовую страницу сайта (SCRAPY_START_URL), на которой находится форма входа
2. После успешного залогинивания переходим на вторую страницу с данными SCRAPY_SECOND_URL
3. В функции action из html кода получаем необходимые данные, путём указания пути через аналоги css-селекторов
* Для вызова всего кода используем функцию get_data_by_id, которая по data_id конкретизирует вторую страницу.
* Во время исполнения будет создаваться временный файл items.json, который будет перезаписываться после каждого вызова (он сейчас специально удаляется, если не удалять, то новые результаты будут записываться в конец файла). items.json создается в Scrapy по умолчанию, и как-то без него скорее всего нельзя, но можно задать формат, если вдруг json вам не по душе.
* пример ниже:

import json
import os
import scrapy
from scrapy.crawler import CrawlerProcess, CrawlerRunner
from twisted.internet import reactor
from multiprocessing import Process, Queue


SCRAPY_START_URL = 'https://example.com/'
SCRAPY_USER = 'user_name'
SCRAPY_PASS = 'some_pass'
SCRAPY_SECOND_URL = 'https://example.com/data/'


class CPSpider(scrapy.Spider):
    name = "crawler_name_here"
    start_urls = [SCRAPY_START_URL]
    second_url = None

    def parse(self, response):
        yield scrapy.FormRequest.from_response(
            response,
            formxpath='//form[@id="loginForm"]',
            formdata={
                'auth_signin[username]': SCRAPY_USER,
                'auth_signin[password]': SCRAPY_PASS,
                'Action': '1',
            },
            callback=self.after_login)

    # запускается после успешного логина
    # переходим на странцу, откуда надо стянуть данные
    def after_login(self, response):
        yield scrapy.Request(
            url=CPSpider.second_url,
            callback=self.action)

    # вытаскиваем данные из html кода с помощью xpath
    def action(self, response):
        final_result_txt = response.xpath('//span[@class="data_display"]/b/text()').extract_first(default='some_default')
        print(final_result_txt)

        # создадим словарь и запишем полученный резальтат
        item = dict()
        item['final_result'] = final_result_txt
        return item


def get_data_by_id(data_id):
    if os.path.isfile('items.json'):
        os.remove('items.json')

    CPSpider.reseller_url = SCRAPY_SECOND_URL + str(data_id)

    # а вот здесь хитрая обертка
    # благодаря таким фокусам можно не запускать основное приложение
    def run_spider(spider):
        def f(q):
            try:
                runner = CrawlerRunner(settings={
                    # установим уровень ошибок выше дефолта
                    # чтобы не сыпалось много отладочного вывода
                    # если надо отдебажть, то ставим DEBUG
                    'LOG_LEVEL': 'ERROR',
                    'FEED_FORMAT': 'json',
                    'FEED_URI': 'items.json'
                })
                deferred = runner.crawl(spider)
                deferred.addBoth(lambda _: reactor.stop())
                reactor.run()
                q.put(None)
            except Exception as e:
                q.put(e)

        q = Queue()
        p = Process(target=f, args=(q,))
        p.start()
        result = q.get()
        p.join()

        if result is not None:
            raise result

    try:
        run_spider(CPSpider)
        # считываем результат, полученнный после работы scrapy
        with open('items.json') as json_file:
            res = json.load(json_file)
        if len(res):
            return res[0]['final_result']
    except BaseException as e:
        return 'error'

Добавить комментарий

Ваш e-mail не будет опубликован. Обязательные поля помечены *

Можно использовать следующие HTML-теги и атрибуты: <a href="" title=""> <abbr title=""> <acronym title=""> <b> <blockquote cite=""> <cite> <code> <del datetime=""> <em> <i> <q cite=""> <strike> <strong>