Веб-краулер на Python: как построить масштабируемый краулер на asyncio

Веб-краулер на Python представляет собой программу, которая загружает страницы, извлекает данные и ссылки и планирует дальнейший обход. Его используют для технического аудита, мониторинга цен и подготовки наборов данных. В реальной работе скорость ограничивают правила сайта, сетевые задержки, память браузера и качество дедупликации URL. Поэтому сбор данных начинается не с количества вкладок, а с очереди с управляемой нагрузкой.

Масштабируемый краулер разделяет планирование, загрузку, парсинг и хранение. Такое разделение позволяет заменить HTTP-клиент на Playwright только для страниц с JavaScript, не переписывая остальную систему. Ниже приведены рабочая основа для asyncio, правила вежливого обхода и способы определить момент, когда увеличение числа воркеров перестаёт ускорять работу.

Из каких компонентов состоит масштабируемый веб-краулер

Масштабируемый веб-краулер состоит из frontier-очереди, набора воркеров, слоя fetcher, парсера и хранилища. У каждого компонента своя ответственность, поэтому медленный сайт, ошибка селектора или сбой записи не останавливают весь обход.

Frontier хранит нормализованные URL и решает, какой хост можно опрашивать следующим. Для небольшого сайта достаточно asyncio.Queue. Для нескольких доменов лучше использовать отдельную очередь на хост, собственный лимит конкурентности и время следующего разрешённого запроса. Этот принцип хорошо сочетается с визуальными сценариями автоматизации, когда браузерные действия нужно отделить от планировщика.

Fetcher получает URL и возвращает HTTP-статус, конечный адрес, заголовки и тело ответа. Сначала он выполняет менее ресурсоёмкий HTTP-запрос. Если контент формируется в браузере, нужен вход или пустой HTML не содержит целевых данных, задача переходит в Playwright. Такой HTTP-first подход использует браузер только при необходимости, а не как универсальный транспорт.

Парсер не должен открывать сетевые соединения. Он принимает готовый документ, извлекает canonical URL, JSON-LD, нужные поля и новые ссылки. Отдельный классификатор помечает страницу как ok, blocked, not_found, parse_error или retryable. Для динамических страниц важно учитывать признаки headless browsing, поскольку WAF оценивает не только User-Agent, но и поведение сессии.

Хранилище сохраняет результаты и состояние обхода. Для одного процесса обычно достаточно SQLite в режиме WAL, а для нескольких экземпляров краулера подойдёт PostgreSQL. JSONL удобен для экспорта, но слаб как единственное хранилище: после перезапуска трудно определить, какие URL завершены, а какие оставались в работе.

Разделение кажется избыточным, пока краулер обрабатывает двадцать страниц. При обходе десятков тысяч URL оно экономит время: зависший Playwright не блокирует HTTP-воркеры, а изменение структуры данных не требует переделывать frontier.

В frontier полезно хранить не строку URL, а структурированную задачу: url, depth, priority, attempt, ready_at, discovered_from и признак браузерной эскалации. Приоритет позволяет раньше обработать карточки товаров или свежие статьи, не дожидаясь всех страниц пагинации. Поле discovered_from помогает найти источник цикла или ошибочной ссылки.

Нормализуйте URL до дедупликации. Код ниже выполняет только базовую нормализацию схемы, регистра хоста, стандартных портов, фрагмента и пустого пути. Правила для завершающего / и query-параметров зависят от сайта. Удалите фрагмент после #, приведите host к нижнему регистру, уберите стандартные порты и примените единую политику к завершающему /. Параметр utm_source часто можно удалить, а page=2 меняет ресурс. Белый список значимых параметров надёжнее удаления всего после знака вопроса.

Для крупных обходов одного набора seen в RAM недостаточно. В нескольких процессах резервируйте задачи атомарно, например через FOR UPDATE SKIP LOCKED в PostgreSQL, и используйте lease со сроком действия, чтобы возвращать незавершённые задачи после сбоя. Bloom filter экономит память, но допускает ложные срабатывания и может пропустить новые URL. Если полнота важнее максимальной скорости, храните нормализованный URL или его криптографический хеш в таблице с уникальным индексом.

Как написать веб-краулер на asyncio и Playwright

Краулер на asyncio стоит строить вокруг воркеров, которые работают в течение всего обхода, и очереди с управляемой нагрузкой. В учебном примере очередь не ограничена, но количество добавляемых URL ограничивает параметр limit.

Рабочей системе нужна ограниченная очередь и отдельный координатор frontier, который предотвращает взаимную блокировку воркеров. Не создавайте отдельную задачу для каждого URL через неограниченный asyncio.gather(): большой sitemap быстро превращает список корутин в лишнее потребление памяти.

Подготовьте Python 3.11 или новее и установите зависимости. В примере Beautiful Soup используется для HTML-парсинга, а Playwright нужен для страниц, где целевой контент появляется после выполнения JavaScript.

1. создайте виртуальное окружение командой python -m venv .venv
2. активируйте его для своей операционной системы
3. установите пакеты командой pip install aiohttp beautifulsoup4 playwright
4. загрузите Chromium командой playwright install chromium
5. добавьте тестовый домен, который разрешает автоматизированный обход

Каркас ниже показывает дедупликацию до постановки в очередь, task_done() в блоке finally, лимит воркеров и один повторно используемый браузер. Это учебный минимум, а не готовый обход любого сайта. Функция нормализации не проверяет встроенные учётные данные, DNS-ответы или доступность внутренних IP-адресов.

Для недоверенных URL нужны отдельная проверка адресов, контроль DNS и сетевые ограничения, включая защиту от SSRF. Пример не реализует robots.txt, задержки, повторные попытки и постоянное состояние. Вызов raise_for_status() передаёт все HTTP-ошибки общему обработчику, поэтому учебный код не различает 404, 429 и 503 и не использует Retry-After.

Правила классификации статусов и повторов, описанные ниже, реализуйте отдельно. Запускайте пример только на собственном или явно разрешённом тестовом ресурсе после проверки правил доступа.

import asyncio
from dataclasses import dataclass
from urllib.parse import urldefrag, urljoin, urlsplit, urlunsplit

import aiohttp
from bs4 import BeautifulSoup
from playwright.async_api import Error as PlaywrightError, async_playwright

@dataclass
class Job:
    url: str
    depth: int = 0

def normalize(base: str, href: str) -> str | None:
    absolute, _ = urldefrag(urljoin(base, href))
    try:
        parsed = urlsplit(absolute)
        if parsed.scheme.lower() not in {"http", "https"} or not parsed.hostname:
            return None
        if parsed.username is not None or parsed.password is not None:
            return None
        scheme = parsed.scheme.lower()
        host = parsed.hostname.lower()
        port = parsed.port
        if ":" in host:  # IPv6
            host = f"[{host}]"
        if port and not ((scheme == "http" and port == 80) or
                         (scheme == "https" and port == 443)):
            host = f"{host}:{port}"
        # Оставляем query-параметры: некоторые из них определяют ресурс.
        # Политику завершающего / и tracking-параметров задают отдельно.
        return urlunsplit((scheme, host, parsed.path or "/", parsed.query, ""))
    except ValueError:
        return None

async def crawl(start_url: str, limit: int = 200, workers: int = 8):
    start_url = normalize(start_url, start_url)
    if not start_url:
        raise ValueError("Некорректный начальный адрес")
    if limit < 1 or workers < 1:
        raise ValueError("limit и workers должны быть положительными")

    queue: asyncio.Queue[Job] = asyncio.Queue()
    seen = {start_url}
    results = []
    start_host = urlsplit(start_url).netloc
    timeout = aiohttp.ClientTimeout(total=20)
    connector = aiohttp.TCPConnector(limit=workers, limit_per_host=2)
    browser_slots = asyncio.Semaphore(2)

    async with aiohttp.ClientSession(
        timeout=timeout,
        connector=connector,
        headers={"User-Agent": "ExampleAuditBot/1.0 contact@example.com"},
    ) as session, async_playwright() as pw:
        browser = await pw.chromium.launch(headless=True)
        try:
            context = await browser.new_context()
            try:
                async def fetch(url: str) -> tuple[str, str]:
                    # Проверяем каждый HTTP-переход до следующего запроса.
                    final_url = url
                    for _ in range(6):
                        if urlsplit(final_url).netloc != start_host:
                            raise ValueError("URL за пределами разрешённого хоста")
                        async with session.get(final_url, allow_redirects=False) as response:
                            if response.status in {301, 302, 303, 307, 308}:
                                location = response.headers.get("Location")
                                if not location:
                                    raise ValueError("Перенаправление без Location")
                                next_url = normalize(final_url, location)
                                if not next_url or urlsplit(next_url).netloc != start_host:
                                    raise ValueError("Перенаправление за пределы разрешённого хоста")
                                final_url = next_url
                                continue
                            response.raise_for_status()
                            html = await response.text(errors="replace")
                            final_url = str(response.url)
                            break
                    else:
                        raise ValueError("Слишком много перенаправлений")
                    if "data-requires-javascript" not in html:
                        return final_url, html

                    async with browser_slots:
                        page = await context.new_page()
                        # Запрещаем браузеру запрашивать другие хосты.
                        async def restrict_route(route):
                            if urlsplit(route.request.url).netloc != start_host:
                                await route.abort()
                            else:
                                await route.continue_()
                        await page.route("**/*", restrict_route)
                        try:
                            async with asyncio.timeout(45):
                                await page.goto(final_url, wait_until="domcontentloaded", timeout=30_000)
                                if urlsplit(page.url).netloc != start_host:
                                    raise ValueError("Браузер перешёл на другой хост")
                                return page.url, await page.content()
                        finally:
                            await page.close()

                async def worker():
                    while True:
                        job = await queue.get()
                        try:
                            final_url, html = await fetch(job.url)
                            soup = BeautifulSoup(html, "html.parser")
                            results.append({
                                "url": final_url,
                                "title": soup.title.get_text(strip=True) if soup.title else None,
                                "depth": job.depth,
                            })
                            for tag in soup.select("a[href]"):
                                link = normalize(final_url, tag["href"])
                                if not link or urlsplit(link).netloc != start_host:
                                    continue
                                if link not in seen and len(seen) < limit:
                                    seen.add(link)
                                    await queue.put(Job(link, job.depth + 1))
                        except Exception as error:
                            # Учебный пример: записываем ошибку, а воркер продолжает работу.
                            results.append({"url": job.url, "error": type(error).__name__})
                        finally:
                            queue.task_done()

                tasks = [asyncio.create_task(worker()) for _ in range(workers)]
                try:
                    await queue.put(Job(start_url))
                    await queue.join()
                finally:
                    for task in tasks:
                        task.cancel()
                    await asyncio.gather(*tasks, return_exceptions=True)
            finally:
                await context.close()
        finally:
            await browser.close()
    return results

if __name__ == "__main__":
    data = asyncio.run(crawl("https://example.com"))
    print(data)

Для быстрой проверки нормализации URL используйте несколько простых тестов. Они проверяют стандартные порты, IPv6, фрагменты и сохранение query-параметров:

assert normalize("https://EXAMPLE.com:443/a", "#section") == "https://example.com/a"
assert normalize("http://example.com", "/?page=2") == "http://example.com/?page=2"
assert normalize("http://[::1]:80/", "/path") == "http://[::1]/path"
assert normalize("https://example.com", "mailto:info@example.com") is None

Эти тесты проверяют только преобразование адреса, а не безопасность сетевого доступа. Для недоверенных URL дополнительно нужны проверка DNS и сетевые ограничения.

Ключевая деталь здесь не Playwright, а жизненный цикл задачи. Каждый queue.put() должен завершаться одним queue.task_done(). Иначе queue.join() будет ждать бесконечно. Такой сбой часто выглядит как медленная сеть, хотя причина находится в исключении внутри воркера.

Все воркеры в примере используют один BrowserContext, а семафор ограничивает количество одновременно открытых страниц. limit_per_host=2 ограничивает одновременные HTTP-соединения, но не число запросов в секунду; частоту обхода контролируйте отдельно. Для публичных страниц без авторизации этого часто достаточно.

Если источникам нужны разные cookies или прокси:

• создайте отдельный context для каждой логической сессии
• не запускайте отдельный процесс Chromium для каждого URL
• закрывайте context после завершения группы задач

Context требует меньше ресурсов, чем отдельный процесс браузера. В рабочей системе также контролируйте состояние общего контекста: после критической ошибки или потери соединения с браузером его может потребоваться пересоздать.

Предусмотрите:

• максимальное время завершения задачи
• корректную отмену корутин
• очистку ресурсов
• внешний контроль и завершение дочерних процессов Chromium

Блок finally помогает закрыть страницы и контексты при обычном завершении или отмене корутин, но не гарантирован при аварийном завершении процесса Python.

Проверка хоста и перехват запросов через page.route() ограничивают часть переходов и запросов, но не гарантируют полную изоляцию браузера.

Учитывайте ограничения:

• не все сетевые обращения обязательно проходят через этот обработчик
• возможны особенности Service Worker
• DNS может вернуть внутренний IP-адрес даже для разрешённого имени

Для недоверенных источников используйте изолированную среду с сетевыми правилами, блокирующими внутренние адреса и службы метаданных.

Параметр limit ограничивает количество уникальных URL, добавленных в обход, а не число успешно обработанных страниц.

Ограниченная очередь, которую те же воркеры читают и заполняют, может вызвать deadlock: все корутины будут ждать свободного места, но некому будет забрать следующий URL.

В рабочей версии:

• передайте добавление найденных ссылок отдельному координатору frontier
• примените backpressure между координатором frontier, fetcher и writer

Рабочей системе также нужны проверка robots.txt до запроса, ограничение частоты на хост, retry с задержками, постоянное хранилище frontier и отдельный writer. Воркеры передают готовые записи во вторую ограниченную очередь, а writer пакетно добавляет их в базу. Пакет из 100 записей обычно создаёт меньше накладных расходов, чем 100 отдельных транзакций. Если база работает медленнее, очередь результатов постепенно заполняется и ограничивает скорость загрузки новых страниц. Это предотвращает перегрузку системы.

В рабочем проекте замените признак data-requires-javascript реальной проверкой. Например, передавайте страницу браузеру, когда нужного CSS-селектора нет, HTML короче ожидаемого минимума или ответ содержит challenge-страницу. Похожие конвейеры полезны для автоматизированного OSINT-сбора, где источники имеют разные форматы и режимы загрузки.

Как обрабатывать ошибки и соблюдать robots.txt

Надёжный краулер отличает постоянные ошибки от временных и проверяет правила доступа до загрузки URL. В учебном коде выше robots.txt ещё не реализован. 404 обычно не нужно повторять, а 429, 502, 503 и сетевой timeout требуют отложенной попытки с exponential backoff и jitter.

Сначала загрузите /robots.txt для каждого хоста и кэшируйте результат. Проверяйте разрешение для фактического User-Agent, учитывайте Crawl-delay, если сайт его задаёт, и ищите ссылки на sitemap. Отсутствие запрета в robots.txt не отменяет условия использования сайта, требования к персональным данным или ограничения доступа.

Зафиксируйте порядок обработки ответа:

1. завершите URL без повтора для статусов 400, 401, 403 и 404, если политика источника не требует другого
2. прочитайте Retry-After для 429 и 503; значение может быть количеством секунд или HTTP-датой
3. рассчитайте паузу как min(cap, base * 2 ** attempt) + jitter
4. верните задачу в frontier с будущим временем готовности
5. переведите URL в failed после исчерпания лимита попыток

Не удерживайте воркер в asyncio.sleep() десятки секунд. Когда время следующей попытки хранится в frontier как ready_at, воркер может обрабатывать другой хост. Отдельные очереди на хост эффективнее простого увеличения workers.

Для Playwright задавайте два timeout:

• короткий для навигации
• более длинный для всей задачи

Если page.goto() зависнет, внешний asyncio.timeout() отменит операцию после заданного времени, но сам по себе не закроет страницу. Всегда закрывайте page или context в finally, чтобы освободить слот пула. Иначе несколько десятков неудачных переходов могут оставить процессы Chromium в памяти.

Повторяйте только идемпотентные операции:

• обычный GET подходит для повторного выполнения
• нажатие кнопки, создающей заказ или меняющей данные, не относится к краулингу

Ведите retry budget для хоста, а не только отдельного URL. Если за последние пять минут половина запросов вернула 429 или 503, приостановите новые задачи домена на короткий cooldown. Иначе тысяча URL с тремя повторами превратит временную проблему в три тысячи лишних запросов.

Circuit breaker можно реализовать с тремя состояниями:

1. closed: запросы выполняются по обычному графику
2. open: после превышения порога ошибок frontier перестаёт выдавать задачи хоста
3. half_open: после cooldown разрешается один тестовый запрос

Успешный тест возвращает систему в closed, а повторная ошибка переводит её обратно в open.

Храните причины сбоев отдельно:

• dns_error
• connect_timeout
• read_timeout
• http_429
• browser_timeout
• parse_error

Эти ошибки требуют разных решений. Один счётчик failed не объяснит, почему из десяти тысяч страниц собрано только семь.

Как обходить sitemap, структурированные данные и обновлённые страницы

Для известного сайта начинайте с sitemap, а не с главной страницы. Sitemap даёт контролируемый список URL, часто содержит lastmod и не заставляет краулер открывать навигационные страницы только для поиска ссылок.

Поддержите sitemap index, обычные XML-файлы и gzip. До распаковки задайте максимальный размер результата, иначе маленький архив может развернуться в гигабайты. Нормализуйте URL, а <lastmod> храните отдельно от фактического времени обхода. Значение может быть устаревшим или неточным, поэтому не используйте его как единственный сигнал повторного обхода.

Организуйте первый проход так:

1. загрузите robots.txt и найдите объявленные sitemap
2. разберите sitemap index до конечных XML-файлов
3. отфильтруйте URL по разрешённому хосту и нужному path-префиксу
4. добавьте страницы в frontier с приоритетом по типу и lastmod
5. загрузите HTML через HTTP
6. передайте сложные страницы в Playwright только после проверки сигналов эскалации
7. запишите content hash, ETag, Last-Modified и время следующего обхода

Сначала ищите JSON-LD. Структурированные данные часто содержат название, цену, валюту, доступность или дату публикации в более стабильном формате, чем CSS-селекторы. Затем используйте microdata, Open Graph и только после этого селекторы страницы. Такая последовательность уменьшает число поломок после редизайна.

Интервал повторного обхода не должен быть одинаковым для всех URL. Часто меняющейся странице товара нужен более короткий цикл, чем архивной статье. Простое правило: сокращайте интервал после изменения content hash и постепенно увеличивайте после нескольких обходов без изменений. Для парсинга динамических страниц дополнительно сохраняйте версию селекторов, чтобы отличать реальное отсутствие данных от сбоя парсера.

В запросах повторного обхода используйте If-None-Match с ETag или If-Modified-Since, если сервер поддерживает эти заголовки. Ответ 304 Not Modified экономит трафик и CPU на парсинг. Он также не должен перезаписывать последнюю успешную запись пустым телом.

После завершения lease задача может быть выдана повторно, поэтому запись результатов должна быть идемпотентной. Используйте уникальный ключ URL и операцию upsert, чтобы не создавать дубликаты.

Чем веб-краулер отличается от веб-скрапера и браузерной автоматизации

Веб-краулер находит и планирует URL, веб-скрапер извлекает поля со страницы, а браузерная автоматизация воспроизводит действия в интерфейсе. Один проект может включать все три части, но смешивание их в одном классе усложняет тестирование и масштабирование.

Для каталога с серверным HTML достаточно aiohttp и парсера. Для SPA, где цена приходит после XHR-запроса, сначала проверьте сетевой API. Playwright оставьте для случаев, когда нужны рендеринг, cookies, сессия или взаимодействие со страницей. Браузер требует больше памяти, CPU и времени запуска, чем HTTP-клиент.

HTTP-first архитектура даёт ещё одно преимущество: код парсинга не знает, откуда получен HTML. В тесте можно заменить fetcher локальным fixture-файлом, проверить селекторы без сети и воспроизвести ошибку через месяц.

Как найти узкие места по трафику, памяти и CPU

Предел конкурентности определяет самый медленный ресурс: лимит сайта, пул соединений, пропускная способность сети, RAM браузера, CPU парсера или скорость записи в базу. Количество воркеров не нужно угадывать. Подбирайте его серией коротких измерений на разрешённом тестовом наборе.

Начните с 1, 2, 4 и 8 воркеров. Для каждого запуска измерьте pages/s, p50 и p95 времени ответа, долю ошибок, RSS всего дерева процессов, CPU и размер очереди. Добавляйте следующий уровень конкурентности только тогда, когда пропускная способность заметно растёт, а ошибки и p95 остаются приемлемыми.

Для предварительной консервативной оценки одного хоста используйте concurrency ≈ target_requests_per_second × p95_latency_seconds. Это ориентир, а не точный расчёт: p95 отражает задержку 95-го процентиля, тогда как для средней одновременной нагрузки нужна средняя задержка.

Если сайт разрешает один запрос в секунду, а p95 равен трём секундам, три одновременных запроса можно использовать как начальную точку теста, а не гарантированно достаточный уровень. Ограничивайте частоту запуска отдельно от числа соединений и проверяйте результат измерениями.

Записывайте показатели в структурированные логи:

• queued и active для состояния frontier
• ok, retry, blocked, parse_error для результатов
• http_seconds, browser_seconds, parse_seconds для этапов
• rss_mb и cpu_percent для ресурсов процесса
• records_written и write_seconds для хранилища

Если парсер надолго блокирует event loop, перенесите ресурсоёмкую обработку в ProcessPoolExecutor. Потоки помогут только тогда, когда библиотека освобождает GIL. Проверьте измерением: обработайте один HTML-набор в event loop, thread pool и process pool.

Если потребление памяти Chromium растёт, ограничьте количество одновременных страниц, блокируйте ненужные изображения, шрифты и видео и повторно используйте context внутри совместимой сессии. Не держите сотни открытых page. При сборе данных через прокси отдельно считайте трафик: браузер может загружать ресурсы, отсутствующие в HTTP-first сценарии.

Начинайте диагностику с анализа графиков, а не с предположений о библиотеке. Рост queued при стабильном active означает, что fetcher или лимиты хоста не справляются. Рост очереди записи при пустом frontier указывает на базу. Постепенное увеличение RSS после каждой страницы может говорить о незакрытых ресурсах, но также бывает следствием кэширования или управления памятью Chromium.

Таблица подсказывает возможную причину, но подтвердить её можно только повторным тестом с изменением одного параметра. Если одновременно увеличить workers, заменить parser и отключить изображения, источник прироста останется неизвестным.

После каждого теста сохраняйте конфигурацию рядом с результатами: версию Python, версию Playwright, тип машины, количество воркеров и набор URL. Иначе цифры двух запусков нельзя честно сравнить.

Как организовать непрерывный краулинг и браузерные сессии с Afina

Непрерывный обход требует checkpoint, стабильных браузерных сред и понятной ответственности за каждый хост. Храните frontier и статусы в базе, а секреты, прокси и данные сессий не добавляйте в код или логи.

Когда части разрешённых источников нужна постоянная браузерная сессия, профили Afina можно использовать как отдельный инфраструктурный слой для браузерных задач, прокси и повторяемых сценариев. Сам краулер остаётся ответственным за очередь, robots.txt, лимиты и сохранение результатов. Материал предоставлен исключительно в ознакомительных и образовательных целях.

Не передавайте весь sitemap в браузерный слой. Сначала выполните недорогой HTTP-проход, а в профиль направляйте только URL с подтверждённой потребностью в JavaScript или постоянной сессии. Так проще прогнозировать расходы и искать причину сбоя.