В эпоху информационного переизбытка собирать нужные данные вручную — всё равно что вычерпывать океан чайной ложкой. Веб-краулеры стали незаменимыми помошниками для тех, кто каждый день работает с большими объёмами данных из интернета. Эти умные "пауки" умеют самостоятельно путешествовать по сайтам, находить нужную информацию и упаковывать её в удобный для анализа формат.
Scrapy — это мощный асинхронный фреймворк, способный одновременно обрабатывать сотни запросов, автоматически следовать по ссылкам, извлекать структурированные данные и экспортировать их в любом удобном формате. От простого сбора новостей до комплексного мониторинга цен — Scrapy адаптируется под задачи любой сложности. Когда я впервые столкнулся с необходимостью собрать данные о ценах конкурентов с тысяч страниц, то потратил неделю на создание скрипта с BeautifulSoup. Тот же функционал на Scrapy удалось реализовать за пару часов, причём с гораздо более элегантным кодом.
Автоматизация сбора данных: пишем веб-краулер на Python и Scrapy
Веб-краулеры — это программные роботы, которые методично перемещаются по сети, словно пауки по паутине, следуя по ссылкам и собирая информацию со страниц. В отличие от ручного сбора, где человек вынужден щелкать по каждой странице, копировать и вставлять данные, краулер делает всё автоматически, экономя сотни часов рутинной работы. Применений веб-краулерам — не счесть. Маркетологи используют их для мониторинга цен конкурентов, аналитики предлагают клиентам отчёты по рыночным трендам, журналисты отслеживают обновления на десятках новостных ресурсов, а исследователи накапливают данные для своих научных работ.
В мире скрейпинга Scrapy стоит особняком, хотя конкуренцию ему составляют несколько популярных инструментов:
BeautifulSoup — прост в освоении, отлично подходит для небольших проектов. Но у него нет встроенных механизмов для обхода нескольких страниц, обработки JavaScript или управления сессиями. Это больше парсер, чем краулер.
Selenium — незаменим, когда нужно имитировать реальное взаимодействие пользователя с браузером или обрабатывать JavaScript-содержимое. Однако он ресурсоёмкий и медленный — не лучший выбор для масштабных задач.
Scrapy — полноценный фреймворк с асинхронной архитектурой, системой конвейеров для обработки данных, встроенной поддержкой экспорта и множеством инструментов для тонкой настройки.
История Scrapy начинается в 2008 году, когда Pablo Hoffman и Shane Evans создали первую версию фреймворка для внутреннего использования в компании Mydeco. Изначально проект назывался "Insophia's Scraping Framework", но вскоре был переименован в Scrapy в честь краулеров-пауков (spiders) и Python. С тех пор фреймворк прошёл огромный путь развития, став open-source продуктом и собрав вокруг себя активное сообщество.
Погружаясь в мир веб-скрейпинга, полезно освоить ключевую терминологию:
Краулинг — процес систематического обхода веб-страниц по ссылкам,
Скрейпинг — извлечение конкретных данных из веб-страниц,
Селекторы — выражения (CSS или XPath), помогающие находить нужные элементы на странице,
Robots.txt — файл с правилами для роботов, указывающий, какие страницы можно или нельзя сканировать,
User-Agent — идентификатор браузера или бота, отправляемый серверу,
Прокси — промежуточные серверы для маскировки запросов.
Краулер на Python
Всем доброго времени суток,
столкнулся с проблемкой, которую хотел бы оптимизировать, задача... Как получить содержимое div контейнера с использованием scrapy?
Вечер добрый,
Пытаюсь распарсить данную страницу vulners.com в качестве параметра передаю... Как сделать связку scrapy + selenium?
Доброго времени суток, есть задача соскрапить несколько тысяч фото по одному запросу в... Как задать собственные имена скачиваемых файлов в Scrapy?
Доброго времени суток. Осваиваю Scrapy, стоит задача скачать изображения по спарсенным ссылкам....
Техническая часть
Настройка окружения для Scrapy — задача несложная, но требует внимания к деталям. В отличие от монолитных библиотек, Scrapy — это фреймворк с множеством зависимостей, и правильная настройка поможет избежать головной боли в будущем.
Для начала, убедитесь что у вас установлен Python 3.6 или новее — Scrapy работает именно с этими версиями. Установка самого фреймворка выполняется через pip:
На Windows иногда возникают проблемы с зависимостями, особенно Twisted — библиотеки, отвечающей за асинхронное выполнение. В таком случае полезно скачать предкомпилированные wheel-пакеты с неофициальных ресурсов вроде Christoph Gohlke's repository. Сам регулярно натыкаюсь на эту проблему, когда помогаю настроить окружение новичкам. После установки Scrapy вы можете создать свой первый проект:
| Python | 1
| scrapy startproject myproject |
|
| Python | 1
2
3
4
5
6
7
8
9
10
| myproject/
scrapy.cfg # Файл конфигурации проекта
myproject/ # Модуль Python с кодом проекта
__init__.py
items.py # Определение структуры данных
middlewares.py # Промежуточные обработчики
pipelines.py # Конвейеры для обработки данных
settings.py # Настройки проекта
spiders/ # Директория для пауков
__init__.py |
|
Теперь займемся созданием первого паука. В Scrapy пауки — это классы Python, наследующие от базового класса scrapy.Spider. Создадим файл в директории spiders, например, quotes_spider.py:
| Python | 1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
| import scrapy
class QuotesSpider(scrapy.Spider):
name = 'quotes' # Уникальное имя паука
start_urls = ['https://quotes.toscrape.com/'] # Начальные URL
def parse(self, response):
# Метод, обрабатывающий ответы с сайта
for quote in response.css('div.quote'):
yield {
'text': quote.css('span.text::text').get(),
'author': quote.css('small.author::text').get(),
'tags': quote.css('div.tags a.tag::text').getall(),
}
# Переход на следующую страницу, если она есть
next_page = response.css('li.next a::attr(href)').get()
if next_page is not None:
yield response.follow(next_page, self.parse) |
|
Метод parse — мозг паука. Он получает ответ от сервера и извлекает данные, используя селекторы. В Scrapy доступны два типа селекторов: CSS и XPath. CSS-селекторы проще для новичков и похожи на те, что используются в JavaScript, а XPath мощнее для сложных случаев. Запуск паука выполняется командой:
По умолчанию Scrapy выводит результаты в консоль, но вы можете сохранить их в файл:
| Python | 1
| scrapy crawl quotes -o quotes.json |
|
| Python | 1
| scrapy crawl quotes -o quotes.csv |
|
В основе Scrapy лежит асинхронная архитектура, построеная на Twisted. Пока один сервер обрабатывает запрос, паук уже отправляет следующий, не дожидаясь ответа. Это позволяет обрабатывать сотни страниц одновременно без создания отдельных потоков или процессов. Именно поэтому Scrapy значительно производительнее простых решений на requests + BeautifulSoup. Однако асинхронность требует аккуратности в коде. Любое блокирующее действие, например, запись в файл или длительная операция с данными, может тормозить всю систему. Для таких случаев Scrapy предлагает использовать конвейеры (pipelines) и сигналы, позволяющие разделить логику сбора данных и их обработки.
Одно из самых полезных и недооцененных средств Scrapy — Scrapy shell. Это интерактивная консоль для экспериментов с селекторами:
| Python | 1
| scrapy shell 'https://quotes.toscrape.com/' |
|
| Python | 1
2
3
4
5
| >>> response.css('div.quote')
[H2]Вернет список элементов div с классом quote[/H2]
>>> response.css('div.quote span.text::text').getall()
# Вернет список всех текстов цитат |
|
Для более сложных проектов полезно создавать специализированые классы для хранения данных — Items. В файле items.py можно определить структуру извлекаемых данных:
| Python | 1
2
3
4
5
6
| import scrapy
class QuoteItem(scrapy.Item):
text = scrapy.Field()
author = scrapy.Field()
tags = scrapy.Field() |
|
Middleware в Scrapy делятся на две категории: для обработки запросов/ответов и для работы с объектами-пауками. Рассмотрим пример создания middleware для ротации User-Agent — частой необходимости при работе с сайтами, блокирующими ботов:
| Python | 1
2
3
4
5
6
7
8
9
10
11
12
13
14
| import random
from scrapy import signals
class RandomUserAgentMiddleware:
user_agents = [
'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36',
'Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/605.1.15',
'Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko)'
# и другие варианты
]
def process_request(self, request, spider):
request.headers['User-Agent'] = random.choice(self.user_agents)
return None |
|
| Python | 1
2
3
| DOWNLOADER_MIDDLEWARES = {
'myproject.middlewares.RandomUserAgentMiddleware': 400,
} |
|
Аналогичным образом можно настроить работу через прокси:
| Python | 1
2
3
4
5
6
7
8
9
10
| class RotatingProxyMiddleware:
proxies = [
'http://proxy1.example.com:8000',
'http://proxy2.example.com:8031',
'http://proxy3.example.com:8080',
]
def process_request(self, request, spider):
request.meta['proxy'] = random.choice(self.proxies)
return None |
|
XPath — мощный инструмент, особенно для сложных HTML-структур, где CSS не справляется. Например, выбрать все ссылки, содержащие слово "Python":
| Python | 1
| response.xpath('//a[contains(text(), "Python")]/@href').getall() |
|
| Python | 1
| response.xpath('//*[@data-category="books"]').getall() |
|
| Python | 1
2
3
4
| # Найти все цены со скидкой
product = response.css('div.product')
original_price = product.xpath('.//span[@class="original-price"]/text()').get()
discount_price = product.css('span.discount-price::text').get() |
|
Иногда структура страницы не позволяет удобно выбрать элемент ни с CSS, ни с XPath. В таких случаях на помошь приходит метод re() для применения регулярных выражений:
| Python | 1
2
| # Извлечь ID продукта из URL
product_id = response.css('a.product::attr(href)').re(r'product/(\d+)')[0] |
|
| Python | 1
2
3
4
5
| # Вместо этого:
title = response.xpath('//div[@class="product"]/h2/text()').get()
# Можно написать так:
title = response.css('div.product').xpath('./h2/text()').get() |
|
Теперь поговорим о том, как Scrapy управляется с хранением и экспортом данных. Помимо упомянутых ранее форматов JSON и CSV, фреймворк поддерживает:- JSON Lines (каждый объект на отдельной строке),
- XML,
- Pickle,
- Marshal,
- Custom Feed Exporters (пользовательские экспортеры).
Но наиболе интересный подход — использование Item Pipelines для продвинутой обработки и хранения данных. Pipeline — это класс с методами, которые вызываются последовательно для каждого элемента (Item), извлеченного пауком. Вот пример Pipeline для удаления дубликатов и сохранения в MongoDB:
| Python | 1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
| import pymongo
class DuplicatesPipeline:
def __init__(self):
self.ids_seen = set()
def process_item(self, item, spider):
if item['id'] in self.ids_seen:
raise DropItem(f"Duplicate item found: {item!r}")
else:
self.ids_seen.add(item['id'])
return item
class MongoDBPipeline:
collection_name = 'scraped_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 |
|
| Python | 1
2
3
4
| ITEM_PIPELINES = {
'myproject.pipelines.DuplicatesPipeline': 300,
'myproject.pipelines.MongoDBPipeline': 800,
} |
|
| Python | 1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
| class AsyncImageDownloaderPipeline:
def process_item(self, item, spider):
if 'image_urls' in item:
deferreds = []
for url in item['image_urls']:
d = defer.Deferred()
d.addCallback(self.handle_image_download, item)
d.addErrback(self.handle_error, item)
deferreds.append(d)
# Асинхронная загрузка изображений
reactor.callLater(0, self.download_image, url, d)
return defer.DeferredList(deferreds).addCallback(lambda _: item)
return item
def download_image(self, url, d):
# Реализация асинхронной загрузки
pass |
|
Работа с большими объёмами данных часто требует оптимизации хранения. Один из моих любимых приёмов — использование кастомного Item Exporter, который сжимает данные перед сохранением:
| Python | 1
2
3
4
5
6
7
8
9
10
11
| from scrapy.exporters import JsonLinesItemExporter
import gzip
class GzipJsonLinesItemExporter(JsonLinesItemExporter):
def __init__(self, file, **kwargs):
self._gzfile = gzip.GzipFile(fileobj=file)
super(GzipJsonLinesItemExporter, self).__init__(self._gzfile, **kwargs)
def finish_exporting(self):
super(GzipJsonLinesItemExporter, self).finish_exporting()
self._gzfile.close() |
|
| Python | 1
2
3
| FEED_EXPORTERS = {
'json.gz': 'myproject.exporters.GzipJsonLinesItemExporter',
} |
|
| Python | 1
| scrapy crawl quotes -o quotes.json.gz |
|
Практическое применение
Один из самых распространённых кейсов — мониторинг цен конкурентов. Представьте, что вы создаёте систему, которая ежедневно проверяет цены на тысячах страниц товаров. Вот каркас такого решения:
| Python | 1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
| class PriceSpider(scrapy.Spider):
name = 'price_monitor'
def __init__(self, product_urls_file=None, *args, **kwargs):
super(PriceSpider, self).__init__(*args, **kwargs)
# Загрузка списка URL из файла
with open(product_urls_file) as f:
self.start_urls = [line.strip() for line in f]
def parse(self, response):
# Извлечение информации о товаре
product = {
'url': response.url,
'name': response.css('h1.product-title::text').get(),
'price': self.extract_price(response),
'availability': 'In stock' if response.css('span.in-stock') else 'Out of stock',
'timestamp': datetime.now().isoformat()
}
yield product
def extract_price(self, response):
price_raw = response.css('span.price::text').get()
if not price_raw:
return None
# Удаление валютных символов и преобразование в число
return float(re.sub(r'[^\d.]', '', price_raw)) |
|
extract_price тут не просто для красоты — в реальной жизни цены часто представлены в разных форматах, с разными валютными знаками и разделителями. Выделение этой логики в отдельный метод делает код более модульным и понятным.
Оптимизация производительности при скрейпинге — дело тонкое, балансирующее между скоростью, надёжностью и "дружелюбностью" к целевому сайту. Порой кажется, что это искуство одновременно гладить кота по шерсти и против неё. Вот несколько техник, доказавших свою эффективность:
1. Настройка параллелизма. Scrapy позволяет контролировать количество одновременных запросов к домену:
| Python | 1
2
3
4
| # В settings.py
CONCURRENT_REQUESTS = 32 # Общее число одновременных запросов
CONCURRENT_REQUESTS_PER_DOMAIN = 8 # Запросов к одному домену
DOWNLOAD_DELAY = 0.5 # Задержка между запросами в секундах |
|
| Python | 1
2
3
| def process_links(self, links):
# Фильтрация ссылок до их обработки
return [link for link in links if not re.search(r'(login|logout|cart)', link.url)] |
|
| Python | 1
2
3
4
| # settings.py
HTTPCACHE_ENABLED = True
HTTPCACHE_EXPIRATION_SECS = 86400 # 24 часа
HTTPCACHE_DIR = 'httpcache' |
|
Одна из самых сложных задач в веб-скрейпинге — обход ограничений сайтов. Многие ресурсы активно сопротивляются автоматичесскому сбору данных, используя CAPTCHA, блокировку IP-адресов и детекцию браузера. Вот несколько подходов, помогающих обойти эти ограничения:
| Python | 1
2
3
4
5
6
7
8
9
10
11
| class AntiBlockMiddleware:
def process_request(self, request, spider):
# Имитация реального пользователя
request.headers['User-Agent'] = random.choice(USER_AGENTS)
request.headers['Accept-Language'] = 'en-US,en;q=0.9'
request.headers['Accept'] = 'text/html,application/xhtml+xml,application/xml'
request.headers['Referer'] = 'https://www.google.com/'
# Замедление запросов с небольшой рандомизацией
time.sleep(random.uniform(0.5, 1.5))
return None |
|
| Python | 1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
| class JsSpider(scrapy.Spider):
name = 'js_content'
def start_requests(self):
for url in self.start_urls:
yield SplashRequest(
url,
self.parse,
args={'wait': 2} # Ждем 2 секунды для выполнения JS
)
def parse(self, response):
# Теперь контент загружен через JavaScript
dynamic_content = response.css('div#js-generated-content::text').get()
yield {'content': dynamic_content} |
|
| Python | 1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
| class SeleniumSpider(scrapy.Spider):
name = 'selenium_login'
def start_requests(self):
yield SeleniumRequest(
url='https://example.com/login',
callback=self.after_login,
wait_time=5,
screenshot=True,
script='document.querySelector("form").submit();'
)
def after_login(self, response):
# Теперь мы залогинены и можем продолжать скрейпинг
yield scrapy.Request('https://example.com/protected-area', self.parse_protected) |
|
| Python | 1
2
3
4
5
6
7
8
9
10
11
12
13
| import nltk
from nltk.sentiment import SentimentIntensityAnalyzer
class SentimentPipeline:
def __init__(self):
nltk.download('vader_lexicon')
self.analyzer = SentimentIntensityAnalyzer()
def process_item(self, item, spider):
if 'review_text' in item:
sentiment = self.analyzer.polarity_scores(item['review_text'])
item['sentiment'] = sentiment['compound'] # От -1 (негативный) до 1 (позитивный)
return item |
|
Установка ScrapyD предельно проста:
| Python | 1
2
| pip install scrapyd
pip install scrapyd-client |
|
scrapy.cfg:
| Python | 1
2
3
| [deploy]
url = http://localhost:6800/
project = myproject |
|
scrapyd) вы можете разворачивать своих пауков:
И запускать их через HTTP API:
| Python | 1
| curl http://localhost:6800/schedule.json -d project=myproject -d spider=quotes |
|
Но распределенный скрейпинг порождает новые проблемы. Например, как избежать дублирования работы? Как отслеживать прогрес? На помощь приходит Redis и система блумфильтров:
| Python | 1
2
3
4
5
6
7
8
9
10
11
12
13
| import redis
from scrapy.dupefilters import BaseDupeFilter
from scrapy.utils.request import request_fingerprint
class RedisDupeFilter(BaseDupeFilter):
def __init__(self, redis_host='localhost', redis_port=6379):
self.server = redis.Redis(host=redis_host, port=redis_port)
self.key = 'scrapy:dupefilter'
def request_seen(self, request):
fp = request_fingerprint(request)
added = self.server.sadd(self.key, fp)
return added == 0 # returns True if request was seen before |
|
| Python | 1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
| from fastapi import FastAPI, Query
from motor.motor_asyncio import AsyncIOMotorClient
from typing import List, Optional
app = FastAPI(title="Scraped Data API")
@app.on_event("startup")
async def startup_db_client():
app.mongodb_client = AsyncIOMotorClient("mongodb://localhost:27017")
app.mongodb = app.mongodb_client.scraped_data
@app.get("/products/", response_model=List[dict])
async def get_products(
category: Optional[str] = Query(None),
min_price: Optional[float] = Query(None),
max_price: Optional[float] = Query(None),
limit: int = 10
):
query = {}
if category:
query["category"] = category
if min_price or max_price:
query["price"] = {}
if min_price:
query["price"]["$gte"] = min_price
if max_price:
query["price"]["$lte"] = max_price
cursor = app.mongodb.products.find(query).limit(limit)
return [doc async for doc in cursor] |
|
Когда объемы скрейпинга растут, управление потоком задач становится критическим. Что делать, если вам нужно обработать миллионы URL с разной приоритетностью? Celery в связке с Redis — моё фаворитное решение:
| Python | 1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
| from celery import Celery
import subprocess
app = Celery('scrapy_tasks', broker='redis://localhost:6379/0')
@app.task
def crawl(spider_name, domain=None):
command = ['scrapy', 'crawl', spider_name]
if domain:
command.extend(['-a', f'domain={domain}'])
subprocess.call(command)
# В другом файле можно планировать задачи:
from scrapy_tasks import crawl
# Запустить паука с низким приоритетом
crawl.apply_async(args=['low_priority_spider'], queue='low')
# Запустить паука с высоким приоритетом
crawl.apply_async(args=['high_priority_spider'], queue='high') |
|
В одном из моих проектов мы пошли еще дальше — создали систему автоматического обучения и переобучения селекторов. Когда структура сайта менялась (это происходит постоянно), наш "умный" краулер узнавал эти изменения и корректировал селекторы в реальном времени:
| Python | 1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
| class AdaptiveSpider(scrapy.Spider):
def __init__(self, *args, **kwargs):
super(AdaptiveSpider, self).__init__(*args, **kwargs)
self.selectors = self.load_selectors()
self.adaptation_history = []
def load_selectors(self):
# Загрузка селекторов из базы данных
return {
'title': ['h1.product-title::text', 'div.product h1::text', 'h1::text'],
'price': ['span.price::text', 'div.price span::text', 'p.price::text']
}
def extract_with_adaptivity(self, response, field):
for selector in self.selectors[field]:
result = response.css(selector).get()
if result:
return result, selector
# Если все селекторы провалились, попробуем найти новый
if field == 'price':
# Эвристика для поиска цен: ищем элементы с символами валют
candidates = response.xpath('//*[contains(text(), "$") or contains(text(), "€")]/text()').getall()
if candidates:
new_selector = self.infer_selector(response, candidates[0])
self.adaptation_history.append((field, new_selector))
return candidates[0], new_selector
return None, None |
|
Мониторинг и отказоустойчивость: создаём надёжные краулеры
Написать краулер, который работает на вашем ноутбуке в идеальных условиях, — половина дела. Реальный мир веб-скрейпинга полон неожиданностей: сайты меняются, сервера ломаются, соединения обрываются. Вот несколько ключевых стратегий, которые приведут ваши краулеры к индустриальным стандартам отказоустойчивости.
Глубокий мониторинг работы пауков
Любой долгоживущий краулер нуждается в механизмах мониторинга. Простейший, но невероятно эффективный подход — внедрение периодических чекпойнтов и метрик:
| Python | 1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
| class MetricsCollectorExtension:
def __init__(self, stats):
self.stats = stats
self.start_time = time.time()
self.items_scraped = 0
self.reporting_interval = 60 # секунды между отчётами
self.last_report_time = self.start_time
def item_scraped(self, item, spider):
self.items_scraped += 1
current_time = time.time()
if current_time - self.last_report_time > self.reporting_interval:
elapsed = current_time - self.start_time
items_per_minute = self.items_scraped / (elapsed / 60)
spider.logger.info(
f"Производительность: {items_per_minute:.2f} элементов/мин, "
f"Всего: {self.items_scraped}, "
f"Прошло времени: {elapsed:.2f} сек"
)
self.last_report_time = current_time
@classmethod
def from_crawler(cls, crawler):
extension = cls(crawler.stats)
crawler.signals.connect(extension.item_scraped, signal=signals.item_scraped)
return extension |
|
| Python | 1
2
3
4
| # settings.py
EXTENSIONS = {
'myproject.extensions.MetricsCollectorExtension': 100,
} |
|
| Python | 1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
| class DataValidatorPipeline:
def __init__(self):
self.price_pattern = re.compile(r'^\d+(\.\d{1,2})?$') # Ожидаемый формат цены
self.non_empty_fields = ['title', 'url'] # Поля, которые не должны быть пустыми
self.invalid_count = 0
self.total_count = 0
self.error_threshold = 0.1 # 10% ошибок — сигнал для тревоги
def process_item(self, item, spider):
self.total_count += 1
has_errors = False
# Проверка обязательных полей
for field in self.non_empty_fields:
if field not in item or not item[field]:
spider.logger.warning(f"Отсутствует обязательное поле {field} в {item}")
has_errors = True
# Проверка формата цены
if 'price' in item and item['price']:
price_str = str(item['price'])
if not self.price_pattern.match(price_str):
spider.logger.warning(f"Некорректный формат цены: {price_str}")
has_errors = True
if has_errors:
self.invalid_count += 1
# Проверка превышения порога ошибок
error_rate = self.invalid_count / self.total_count
if error_rate > self.error_threshold and self.total_count > 10:
spider.logger.critical(
f"ТРЕВОГА! Превышен порог ошибок данных: {error_rate:.2%}. "
f"Возможно, структура сайта изменилась."
)
# Здесь можно добавить отправку уведомления
return item |
|
Интеграция и автоматизация
Современный подход к скрейпингу уже не ограничивается запуском скрипта из консоли. Интеграция веб-краулеров с другими системами часто становится ключом к созданию действительно полезных решений:
| Python | 1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
| # Интеграция с Slack для уведомлений
from slack_sdk import WebClient
class SlackNotifier:
def __init__(self, token, channel):
self.client = WebClient(token=token)
self.channel = channel
def send_notification(self, message, attachments=None):
try:
self.client.chat_postMessage(
channel=self.channel,
text=message,
attachments=attachments
)
return True
except Exception as e:
print(f"Ошибка отправки сообщения в Slack: {e}")
return False
# В spider.py
def closed(self, reason):
stats = self.crawler.stats.get_stats()
items_count = stats.get('item_scraped_count', 0)
error_count = stats.get('log_count/ERROR', 0)
message = f"Краулер {self.name} завершил работу по причине: {reason}\n"
message += f"Собрано элементов: {items_count}\n"
message += f"Ошибок: {error_count}"
notifier = SlackNotifier(
token=self.settings.get('SLACK_TOKEN'),
channel=self.settings.get('SLACK_CHANNEL')
)
notifier.send_notification(message) |
|
| Python | 1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
| from airflow import DAG
from airflow.operators.bash_operator import BashOperator
from airflow.operators.python_operator import PythonOperator
from datetime import datetime, timedelta
import yaml
import os
# Загрузка конфигурации краулеров
with open('/path/to/crawlers_config.yaml') as f:
crawlers_config = yaml.safe_load(f)
default_args = {
'owner': 'data_team',
'depends_on_past': False,
'start_date': datetime(2023, 1, 1),
'email_on_failure': True,
'email_on_retry': False,
'retries': 1,
'retry_delay': timedelta(minutes=5),
}
dag = DAG(
'dynamic_web_crawlers',
default_args=default_args,
description='Динамически генерируемые задачи краулеров',
schedule_interval='0 */6 * * *', # Каждые 6 часов
catchup=False
)
def post_processing(crawler_name, **context):
# Обработка данных после успешного запуска краулера
import pandas as pd
from sqlalchemy import create_engine
# Путь к файлу с результатами
output_file = f'/path/to/data/{crawler_name}.json'
# Загрузка и обработка данных
data = pd.read_json(output_file, lines=True)
# Пример обработки: очистка и трансформация
if 'description' in data.columns:
data['description'] = data['description'].str.strip()
# Загрузка в базу данных
engine = create_engine('postgresql://user:password@localhost:5432/scraped_data')
data.to_sql(f'{crawler_name}_items', engine, if_exists='replace', index=False)
return f"Обработано {len(data)} записей для {crawler_name}"
# Динамическое создание задач для каждого краулера
for crawler in crawlers_config:
name = crawler['name']
schedule = crawler.get('schedule', '0 0 * * *') # По умолчанию - ежедневно в полночь
# Задача запуска краулера
crawl_task = BashOperator(
task_id=f'crawl_{name}',
bash_command=f'cd /path/to/scrapy/project && scrapy crawl {name} -o /path/to/data/{name}.json',
dag=dag
)
# Задача постобработки
process_task = PythonOperator(
task_id=f'process_{name}',
python_callable=post_processing,
op_kwargs={'crawler_name': name},
dag=dag
)
# Определение зависимостей
crawl_task >> process_task |
|
Обработка ошибок и перезапуск
Опытные разработчики скрейперов знают: даже самый идеальный краулер может столкнуться с непредвиденными ситуациями. Сайты временно падают, соединение прерывается, прокси-серверы выходят из строя. Грамотная стратегия обработки ошибок — ключ к долгосрочной надежности:
| Python | 1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
| class ErrorHandlingMiddleware:
def __init__(self, settings):
self.retry_codes = set(settings.getlist('RETRY_HTTP_CODES', [500, 502, 503, 504, 408, 429]))
self.retry_times = settings.getint('RETRY_TIMES', 2)
self.retry_backoff = settings.getbool('RETRY_BACKOFF', True)
self.backoff_factor = settings.getfloat('RETRY_BACKOFF_FACTOR', 0.5)
self.max_backoff = settings.getfloat('RETRY_BACKOFF_MAX', 60)
self.stats = None
@classmethod
def from_crawler(cls, crawler):
middleware = cls(crawler.settings)
middleware.stats = crawler.stats
return middleware
def process_exception(self, request, exception, spider):
# Логика для обработки различных типов исключений
if isinstance(exception, ConnectionRefusedError):
# Возможно, сервер перегружен или блокирует нас
spider.logger.warning(f"Соединение отклонено для {request.url}")
return self._retry_request(request, exception, spider)
# Особая обработка для Timeout
if isinstance(exception, TimeoutError):
spider.logger.warning(f"Таймаут запроса для {request.url}")
# Увеличиваем таймаут на повторных запросах
new_request = request.copy()
new_request.meta['download_timeout'] = request.meta.get('download_timeout', 30) * 1.5
return self._retry_request(new_request, exception, spider)
# Общая обработка остальных исключений
return None
def process_response(self, request, response, spider):
# Проверка HTTP-кодов для повторных попыток
if response.status in self.retry_codes:
return self._retry_request(request, None, spider)
return response
def _retry_request(self, request, exception, spider):
retries = request.meta.get('retry_times', 0) + 1
if retries <= self.retry_times:
# Рассчитываем время задержки с экспоненциальным ростом
if self.retry_backoff:
delay = min(self.backoff_factor * (2 ** (retries - 1)), self.max_backoff)
else:
delay = 0
# Статистика
if self.stats:
self.stats.inc_value('retry/count')
# Подготовка нового запроса
retry_request = request.copy()
retry_request.meta['retry_times'] = retries
retry_request.dont_filter = True
# Добавление задержки
if delay:
spider.logger.info(f"Повторная попытка {retries}/{self.retry_times} для {request.url} через {delay} сек")
return request.replace(dont_filter=True, meta=retry_request.meta)
else:
return retry_request
# Превышено количество повторов
spider.logger.error(f"Исчерпаны попытки для {request.url}")
if self.stats:
self.stats.inc_value('retry/max_reached')
# Можно создать задачу для последующего ручного разбора
self._save_failed_request(request, exception, spider)
return None
def _save_failed_request(self, request, exception, spider):
# Сохранение информации о неудачном запросе для последующего анализа
failed_dir = 'failed_requests'
os.makedirs(failed_dir, exist_ok=True)
timestamp = datetime.now().strftime('%Y%m%d_%H%M%S')
filename = f"{failed_dir}/{spider.name}_{timestamp}.json"
with open(filename, 'w') as f:
json.dump({
'url': request.url,
'headers': dict(request.headers),
'method': request.method,
'exception': str(exception) if exception else None,
'timestamp': timestamp
}, f, indent=2) |
|
Еще одна мощная техника — создание системы постепенной деградации. Если основной метод извлечения данных не сработал, попробовать запасной, менее оптимальный:
| Python | 1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
| def extract_with_fallback(self, response, field):
# Первичная стратегия: CSS-селектор
value = response.css(f'{field}::text').get()
if value:
return value
# Запасная стратегия 1: XPath
value = response.xpath(f'//*[contains(@class, "{field}")]/text()').get()
if value:
return value
# Запасная стратегия 2: регулярное выражение
pattern = FIELD_PATTERNS.get(field)
if pattern:
matches = re.search(pattern, response.text)
if matches:
return matches.group(1)
# Запасная стратегия 3: поиск по ключевым словам
for keyword in FIELD_KEYWORDS.get(field, []):
if keyword in response.text:
# Извлечь контекст вокруг ключевого слова
start = response.text.find(keyword)
context = response.text[max(0, start-100):start+200]
# Здесь можно применить более сложную эвристику для извлечения значения
return f"Контекст вокруг '{keyword}': {context}"
# Все стратегии провалились
return None |
|
Лучшие практики для долгосрочных проектов
Если ваш веб-краулер — это не разовый скрипт, а долгоживущая система, стоит инвестировать время в несколько дополнительных практик:
1. Тесты на регрессию — создайте набор HTML-страниц, отражающих структуру целевого сайта, и пишите автоматические тесты для своих селекторов. Вот пример простого теста:
| Python | 1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
| import unittest
from scrapy.http import TextResponse
from scrapy.selector import Selector
from your_project.spiders.your_spider import YourSpider
class SpiderTest(unittest.TestCase):
def setUp(self):
self.spider = YourSpider()
# Загрузка тестовых HTML-файлов
with open('tests/samples/product_page.html', 'r') as f:
self.product_html = f.read()
# Создание mock-объекта Response
self.response = TextResponse(
url='https://example.com/product/123',
body=self.product_html.encode('utf-8')
)
def test_parse_product(self):
# Проверка работы парсера
results = list(self.spider.parse_product(self.response))
# Тесты на извлечение данных
self.assertEqual(len(results), 1)
item = results[0]
self.assertEqual(item['name'], 'Test Product')
self.assertEqual(item['price'], '99.99')
self.assertIsNotNone(item['description']) |
|
| Python | 1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
| class VersionedSelectors:
def __init__(self):
self.selectors = {
# Версия 1 (до рефакторинга сайта в мае 2023)
1: {
'name': 'h1.product-title::text',
'price': 'span.price-current::text',
'description': 'div.product-desc p::text',
},
# Версия 2 (после рефакторинга)
2: {
'name': 'h1.product-name::text',
'price': 'div.price-box span.current::text',
'description': 'div.description-content::text',
}
}
# Правила определения версии
self.version_detection = {
1: lambda response: response.css('span.price-current').get() is not None,
2: lambda response: response.css('div.price-box').get() is not None
}
def detect_version(self, response):
for version, detector in self.version_detection.items():
if detector(response):
return version
# По умолчанию используем последнюю версию
return max(self.selectors.keys())
def get_selector(self, response, field):
version = self.detect_version(response)
return self.selectors[version].get(field)
# Использование в пауке
def parse_product(self, response):
versioned = VersionedSelectors()
name_selector = versioned.get_selector(response, 'name')
name = response.css(name_selector).get()
# ... остальной код |
|
| Python | 1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
| # extractors.py
class PriceExtractor:
def __init__(self):
self.currency_symbols = {'$', '€', '£', '¥', '₽'}
self.decimal_separators = {'.', ','}
self.thousand_separators = {' ', '.', ','}
def normalize_price(self, price_text):
if not price_text:
return None
# Удаление символов валют
for symbol in self.currency_symbols:
price_text = price_text.replace(symbol, '')
# Обработка разделителей
price_text = price_text.strip()
# Определение разделителя дробной части
decimal_sep = None
for sep in self.decimal_separators:
if sep in price_text[-3:]: # Ищем в последних 3 символах
decimal_sep = sep
break
if not decimal_sep:
# Нет дробной части, просто удаляем разделители тысяч
for sep in self.thousand_separators:
price_text = price_text.replace(sep, '')
return price_text
# Есть дробная часть
parts = price_text.split(decimal_sep)
if len(parts) != 2:
return None # Некорректный формат
# Обработка целой части
whole_part = parts[0]
for sep in self.thousand_separators:
whole_part = whole_part.replace(sep, '')
# Сборка результата
return whole_part + '.' + parts[1]
# В spider.py
from .extractors import PriceExtractor
class ProductSpider(scrapy.Spider):
# ...
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
self.price_extractor = PriceExtractor()
def parse_item(self, response):
# ...
price_text = response.css('span.price::text').get()
normalized_price = self.price_extractor.normalize_price(price_text)
# ... |
|
| Python | 1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
| class HealthCheckSpider(scrapy.Spider):
name = 'health_check'
allowed_domains = ['example.com']
start_urls = ['https://example.com/test-page']
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
self.expected_fields = {
'title': 'Example Domain',
'headings_count': 1,
'links_count_min': 1
}
self.health_check_passed = True
self.issues = []
def parse(self, response):
# Проверка заголовка страницы
title = response.css('title::text').get()
if title != self.expected_fields['title']:
self.health_check_passed = False
self.issues.append(f"Title mismatch: expected '{self.expected_fields['title']}', got '{title}'")
# Проверка количества заголовков
headings = response.css('h1::text').getall()
if len(headings) != self.expected_fields['headings_count']:
self.health_check_passed = False
self.issues.append(f"Headings count mismatch: expected {self.expected_fields['headings_count']}, got {len(headings)}")
# Проверка наличия ссылок
links = response.css('a::attr(href)').getall()
if len(links) < self.expected_fields['links_count_min']:
self.health_check_passed = False
self.issues.append(f"Links count below minimum: expected at least {self.expected_fields['links_count_min']}, got {len(links)}")
# Вывод результатов проверки
yield {
'health_check_passed': self.health_check_passed,
'issues': self.issues if not self.health_check_passed else [],
'url': response.url,
'timestamp': datetime.now().isoformat()
}
def closed(self, reason):
if not self.health_check_passed:
# Отправка уведомления об ошибках
message = "Health check failed:\n" + "\n".join([f"- {issue}" for issue in self.issues])
# Здесь код для отправки уведомления через Slack, Email и т.д.
else:
self.logger.info("Health check passed successfully") |
|
Тонкое искуство веб-скрейпинга — в балансе между техническими навыками и этическим подходом. Создавайте краулеры, которые бережно относятся к ресурсам целевых сайтов, избегайте агрессивного скрейпинга, и ваши проекты по сбору данных будут работать надежно и долго.
Scrapy crawl как объект
Я бы хотел запускать своего паука из функции как объект, чтобы при каждом запуске изменялось его... Парсинг на scrapy
Добрый день. проблема распарсить сайт на питоне с помощью фреймворка Scrapy. вещь в наших краях не... Сохранить видимый текст страницы используя Scrapy
Задание у меня такое, мне необходимо со случайного сайта сохранить видимый текст. Причём с шести... Scrapy не переходит по странице
Привет всем! Почему паук не переходит по страницам использую правило(что не так делаю), тут код:
... Scrapy передача респонса
Добрый день!
Спасибо!
ну не поленитесь переписать хоть парсер на фреймворке scrapy
Вcем привет. Пытаюcь cпарcить некоторые данный c cайта c помощью фреймворка scrapy,однако, не могу... Взаимодействие двух пауков в фреймфорке scrapy
Здравствуйте, как сделать в scrapy так, чтобы один паук вернул данные, которые пропарсил, а второй... Scrapy, определенное количество пауков за раз
Есть управляющий скрипт, который запускает нескольких пауков
s = get_project_settings()
process... Не понимаю CrawlSpider, Rule и LinkExtractor в Scrapy
Всем привет.
Я не понимаю, как работает CrawlSpider, Rule и LinkExtractor в Скрапи.
... Авторизация, Парсер Aliexpress на Scrapy
Приветствую!
Что делаю не так? Помогите с авторизацией на али, пожалуйста.
Все запросы к... Проблема с куками и сессией Selenium + scrapy
Задачка следующая. Есть сайт, написанный на Ангулар. Нужно выбрать город и в этой сессии уже начать... Scrapy-splash + Lua script, проблема с redirect после нажатия кнопки поиска
Всем привет.
Я экономист и полнейший нуб в webscraping, но мне для дипломной работы в маге...
|
Очень интересно)). Инструмент действительно стоящий, обязательно воспользуюсь.
