Урок 4. Парсинг с пагинацией на Python

Каталоги, ленты новостей и результаты поиска почти всегда разбиты на страницы. Парсер, который читает только первую из них, собирает лишь малую часть данных. В этом уроке учим скрипт проходить по всем страницам: по ссылке "следующая", по номеру в адресе, с паузами, повторными попытками при сбоях и надежным условием остановки. Тренируемся на учебном сайте quotes.toscrape.com, разбор отдельной страницы которого мы делали в уроке Урок 2. Парсер на Python.

Какой бывает пагинация

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

На quotes.toscrape.com работают сразу два первых способа: адреса имеют вид /page/2/, а внизу страницы есть кнопка "Next" со ссылкой в разметке <li class="next"><a href="/page/2/">.

Подготовка: requests.Session и функция разбора

При обходе десятков страниц стоит использовать requests.Session. Сессия держит открытым TCP-соединение с сервером (не нужно заново устанавливать соединение и TLS для каждого запроса), хранит куки и общие заголовки. На сотне страниц это ускоряет работу в разы и снижает нагрузку на сайт.

import requests
from bs4 import BeautifulSoup

BASE_URL = "https://quotes.toscrape.com"

session = requests.Session()
session.headers.update({"User-Agent": "Mozilla/5.0 StudyParser/1.0"})

def parse_page(html: str) -> list[dict]:
    soup = BeautifulSoup(html, "lxml")
    items = []
    for q in soup.select("div.quote"):
        items.append({
            "text": q.select_one("span.text").get_text(strip=True),
            "author": q.select_one("small.author").get_text(strip=True),
            "tags": [a.get_text(strip=True) for a in q.select("a.tag")],
        })
    return items

Функция разбора возвращает список словарей для одной страницы. Обход страниц и разбор разделены: так проще тестировать и менять логику пагинации, не трогая селекторы.

Обход по ссылке "следующая страница"

Самый надежный способ: начинаем с первой страницы, на каждой ищем ссылку "Next" и переходим по ней. Когда ссылки нет, страницы закончились. Не нужно заранее знать их количество, и цикл не сломается, если сайт поменяет схему адресов.

import time
from urllib.parse import urljoin

def crawl_by_next(start_url: str, delay: float = 1.0) -> list[dict]:
    url = start_url
    all_items: list[dict] = []
    visited: set[str] = set()

    while url and url not in visited:
        visited.add(url)
        response = session.get(url, timeout=10)
        response.raise_for_status()

        all_items.extend(parse_page(response.text))

        soup = BeautifulSoup(response.text, "lxml")
        next_link = soup.select_one("li.next > a")
        url = urljoin(url, next_link["href"]) if next_link else None

        print(f"Собрано: {len(all_items)}")
        time.sleep(delay)
    return all_items

quotes = crawl_by_next(BASE_URL + "/")
print(len(quotes))     # -> 100

Два важных момента. Во-первых, urljoin превращает относительную ссылку /page/2/ в полный адрес. Во-вторых, множество visited защищает от вечного цикла: если на последней странице ссылка "Next" ведет на саму себя (такое встречается), парсер остановится.

Обход по номеру страницы в URL

Если адреса предсказуемы, можно просто увеличивать номер. Главный вопрос - когда остановиться. Варианты условия выхода:

• сервер вернул 404;
• на странице нет ни одного элемента (у quotes.toscrape.com на странице 11 код ответа 200, но цитат нет);
• достигнут известный номер последней страницы из блока пагинации;
• достигнут ваш собственный лимит max_pages, страховка от ошибок.

def crawl_by_number(max_pages: int = 50, delay: float = 1.0) -> list[dict]:
    all_items: list[dict] = []
    for page in range(1, max_pages + 1):
        response = session.get(f"{BASE_URL}/page/{page}/", timeout=10)
        if response.status_code == 404:
            break
        response.raise_for_status()

        items = parse_page(response.text)
        if not items:                 # пустая страница: данные закончились
            break
        all_items.extend(items)
        time.sleep(delay)
    return all_items

Для параметров в строке запроса не собирайте адрес вручную, а передавайте словарь: session.get(url, params={"page": page, "sort": "new"}). requests сам закодирует значения и подставит знаки ? и &.

Паузы между запросами

Без пауз цикл отправит запросы так быстро, как позволит сеть. Для сайта это выглядит как атака, и вас справедливо заблокируют. Стандартная пауза для учебных и небольших проектов - 1-2 секунды. Чтобы запросы выглядели менее механическими, добавляют случайный разброс:

import random
time.sleep(random.uniform(1.0, 2.5))

Если в файле robots.txt указана директива Crawl-delay, соблюдайте ее. Если сервер ответил кодом 429 (Too Many Requests), снижайте темп: в заголовке Retry-After часто указано, сколько секунд нужно подождать. Подробнее о функции паузы читайте в статье time.sleep в Python.

Обработка ошибок и повторы

На длинном обходе обязательно случится сбой: таймаут, обрыв соединения, ответ 502 от перегруженного сервера. Падать из-за этого на 80-й странице обидно, поэтому запрос оборачивают в функцию с повторными попытками и растущей паузой.

RETRY_STATUSES = {429, 500, 502, 503, 504}

def get_with_retry(url: str, retries: int = 3, backoff: float = 2.0) -> requests.Response:
    for attempt in range(1, retries + 1):
        try:
            response = session.get(url, timeout=10)
            if response.status_code in RETRY_STATUSES:
                raise requests.HTTPError(f"status {response.status_code}", response=response)
            response.raise_for_status()
            return response
        except (requests.ConnectionError, requests.Timeout, requests.HTTPError) as e:
            status = e.response.status_code if e.response is not None else None
            if attempt == retries or (status and status not in RETRY_STATUSES):
                raise
            wait = backoff ** attempt
            print(f"Ошибка {e!r}, попытка {attempt}, ждем {wait:.0f} с")
            time.sleep(wait)

Логика такая: временные ошибки (сеть, таймаут, коды 429 и 5xx) повторяем с паузой 2, 4, 8 секунд, а постоянные (404, 403) пробрасываем сразу, потому что повтор их не исправит. Такая схема называется экспоненциальной задержкой.

Можно не писать повторы вручную, а подключить встроенный механизм urllib3 к сессии:

from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry

retry = Retry(total=3, backoff_factor=1, status_forcelist=[429, 500, 502, 503, 504])
session.mount("https://", HTTPAdapter(max_retries=retry))

После этого каждый session.get автоматически повторяет неудачные запросы. О перехвате исключений в целом рассказано в материале про обработку исключений.

Итоговый парсер со всеми страницами

Соберем все приемы в один скрипт: сессия с повторами, обход по ссылке "Next", паузы, лимит страниц, промежуточный вывод и сохранение в CSV.

import csv
import random
import time
from urllib.parse import urljoin

import requests
from bs4 import BeautifulSoup
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry

START_URL = "https://quotes.toscrape.com/"
MAX_PAGES = 100

def make_session() -> requests.Session:
    s = requests.Session()
    s.headers["User-Agent"] = "Mozilla/5.0 StudyParser/1.0"
    retry = Retry(total=3, backoff_factor=1, status_forcelist=[429, 500, 502, 503, 504])
    s.mount("https://", HTTPAdapter(max_retries=retry))
    return s

def parse(soup: BeautifulSoup) -> list[dict]:
    return [
        {
            "text": q.select_one("span.text").get_text(strip=True),
            "author": q.select_one("small.author").get_text(strip=True),
            "tags": ", ".join(a.get_text(strip=True) for a in q.select("a.tag")),
        }
        for q in soup.select("div.quote")
    ]

def crawl() -> list[dict]:
    session = make_session()
    url, page, rows = START_URL, 0, []
    while url and page < MAX_PAGES:
        page += 1
        try:
            resp = session.get(url, timeout=10)
            resp.raise_for_status()
        except requests.RequestException as e:
            print(f"Страница {page} пропущена: {e}")
            break
        soup = BeautifulSoup(resp.text, "lxml")
        rows.extend(parse(soup))
        nxt = soup.select_one("li.next > a")
        url = urljoin(url, nxt["href"]) if nxt else None
        print(f"Страница {page}: всего {len(rows)} записей")
        time.sleep(random.uniform(1.0, 2.0))
    return rows

if __name__ == "__main__":
    data = crawl()
    with open("all_quotes.csv", "w", newline="", encoding="utf-8-sig") as f:
        writer = csv.DictWriter(f, fieldnames=["text", "author", "tags"])
        writer.writeheader()
        writer.writerows(data)
    print("Готово:", len(data))

Вывод (сокращенно):

Страница 1: всего 10 записей
Страница 2: всего 20 записей
...
Страница 10: всего 100 записей
Готово: 100

Почему данные копятся в списке словарей

Список словарей - универсальный промежуточный формат. Его можно записать в CSV через DictWriter, в JSON одной строкой json.dump, превратить в таблицу pandas через pd.DataFrame(rows) или отфильтровать обычным списковым включением. Ключи словаря одновременно служат заголовками столбцов.

Если страниц тысячи, не держите все в памяти до конца: дописывайте строки в CSV сразу после разбора каждой страницы, открыв файл до цикла. Тогда при аварийной остановке уже собранные данные сохранятся, и парсер можно будет продолжить с нужной страницы.

Дозапись в CSV и продолжение после сбоя

Ниже вариант обхода по номерам страниц, который пишет каждую страницу в файл сразу и умеет продолжить работу с места остановки. Номер последней обработанной страницы хранится в маленьком текстовом файле. Если парсер упадет на 57-й странице, следующий запуск начнет с 58-й, а не с начала.

import csv
from pathlib import Path

OUT = Path("quotes_stream.csv")
STATE = Path("last_page.txt")
FIELDS = ["page", "text", "author", "tags"]

def crawl_resumable(max_pages: int = 50) -> None:
    start = int(STATE.read_text()) + 1 if STATE.exists() else 1
    new_file = not OUT.exists()
    with OUT.open("a", newline="", encoding="utf-8-sig") as f:
        writer = csv.DictWriter(f, fieldnames=FIELDS)
        if new_file:
            writer.writeheader()
        for page in range(start, max_pages + 1):
            resp = get_with_retry(f"{BASE_URL}/page/{page}/")
            items = parse_page(resp.text)
            if not items:
                print("Данные закончились на странице", page)
                break
            for item in items:
                writer.writerow({"page": page, **item, "tags": ", ".join(item["tags"])})
            f.flush()                          # гарантируем запись на диск
            STATE.write_text(str(page))        # запоминаем прогресс
            time.sleep(random.uniform(1.0, 2.0))

Здесь используются функции get_with_retry и parse_page из предыдущих разделов. Метод flush() сбрасывает буфер записи на диск после каждой страницы, чтобы при аварийном завершении в файле не потерялись последние строки. Конструкция {"page": page, **item} объединяет номер страницы с полями записи в один словарь, а список тегов превращается в строку, потому что CSV не умеет хранить списки.

Чтобы начать сбор заново, достаточно удалить файлы quotes_stream.csv и last_page.txt. Для очень больших объемов вместо CSV удобнее SQLite: уникальный индекс по ссылке на запись не даст сохранить дубликаты, даже если одна и та же страница будет обработана дважды.

Проверка собранных данных

После обхода убедитесь, что результат правдоподобен. Простые проверки занимают пару строк и экономят часы на поиске ошибок потом:

• число записей совпадает с ожидаемым (10 страниц по 10 цитат дают 100);
• нет дубликатов: len({r["text"] for r in rows}) == len(rows);
• нет пустых обязательных полей: [r for r in rows if not r["author"]] должен быть пустым списком;
• последняя страница действительно последняя: на ней нет ссылки "Next".

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

Частые вопросы

Ответы на типичные вопросы читателей, которые помогут избежать частых ошибок.

Как узнать общее количество страниц?

Посмотрите блок пагинации: часто там есть ссылка на последнюю страницу или текст "Страница 1 из 37". Номер можно извлечь регулярным выражением. Но надежнее не полагаться на него, а идти по ссылке "следующая" до конца.

Почему парсер зациклился на последней странице?

Ссылка "следующая" на последней странице ведет на эту же страницу, или сайт на несуществующий номер отдает первую страницу. Храните множество посещенных адресов и проверяйте, что на новой странице появились новые записи.

Как ускорить обход сотен страниц?

Сначала используйте requests.Session, это уже заметно быстрее. Дальше можно выполнять несколько запросов параллельно через асинхронный клиент httpx, но с ограничением числа одновременных запросов, чтобы не перегружать сайт.

Как обходить страницы, если номер передается в параметре ?page=?

Передавайте номер через словарь: session.get(url, params={"page": n}). Остальные параметры фильтра и сортировки положите в тот же словарь, чтобы они не потерялись при переходе на следующую страницу.