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'