Полное руководство по краулингу списков

Извлечение данных с сайтов, которые показывают информацию в виде списков (каталоги товаров, вакансии, результаты поиска), непросто. Сайты разбивают элементы на несколько страниц, прячут их за кнопками «Load More» или используют бесконечную прокрутку. Без правильного подхода ваш набор данных будет неполным.

Это руководство охватывает основы краулинга списков: как надёжно извлекать структурированные данные с помощью Python и инструментов автоматизации.

Что такое list crawling?

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

Зачем нужен краулинг списков

Современные сайты организуют данные по предсказуемым шаблонам. Каждая карточка товара, вакансия или запись в каталоге следует одной и той же HTML-структуре. Это делает их идеальными для автоматизированного извлечения. Сложность в том, чтобы получить доступ ко всем элементам, когда они распределены по пагинации, бесконечной прокрутке или динамической подгрузке.

Основные компоненты обхода списков

Identification: Найдите HTML-контейнер, который содержит каждый повторяющийся элемент (карточка товара, вакансия). Это определяет, какие данные можно извлечь.

Extraction: Сопоставляйте конкретные поля внутри каждого контейнера (name, price, location) со своей схемой данных. Единообразное сопоставление полей обеспечивает чистый вывод.

Traversal: Обходите все элементы с помощью пагинации, бесконечной прокрутки или кнопок «Load More». Это обеспечивает полный сбор данных.

Real-World Applications

E-Commerce Price Monitoring: Отслеживайте изменения цен, наличие товара и предложения конкурентов, извлекая названия товаров, цены, рейтинги и статус наличия.

Job Market Analysis: Собирайте названия должностей, диапазоны зарплат, требуемые навыки, названия компаний и локации from job sites. Используйте эти данные для оповещений о вакансиях, анализа трендов и сравнения зарплат.

Business Intelligence: Извлекайте из каталогов контактные данные, часы работы, оценки клиентов и категории услуг для генерации лидов и конкурентного анализа.

Content Aggregation: Собирайте данные с новостных сайтов, блогов и контентных платформ, чтобы строить базы данных, отслеживать трендовые темы и следить за конкретными темами.

Identifying List Types

Прежде чем писать код, разберитесь, как целевой сайт структурирует свои списки. Способ загрузки определяет вашу стратегию обхода.

Standard Pagination: Фиксированное количество элементов на странице с пронумерованными ссылками или кнопками «Далее». Номера страниц отображаются в URL в виде ?page=2 или /page/2/. Проще всего собирать.

Load More Buttons: Один и тот же URL, новые элементы добавляются по клику. Либо автоматизируйте нажатие кнопки, либо напрямую перехватывайте фоновый API-запрос.

Infinite Scroll: Новые элементы подгружаются автоматически по мере прокрутки вниз. Это часто встречается в соцсетях и на современных платформах электронной коммерции. Требуется имитировать прокрутку или перехватывать вызовы базового API.

Search Result Pages: Результаты зависят от параметров запроса, фильтров и сортировки. Ваш краулер должен управлять этими параметрами, одновременно обрабатывая пагинацию. Посмотрите мою статью о лучшие SERP API-сервисы, что позволяет обойти логику сбора данных и получать результаты напрямую через API.

Выбор метода краулинга

Метод Лучше всего подходит для Плюсы Минусы Парсинг через HTTP-запросы Статический HTML с пагинацией по URL Быстрый, легковесный, простой Не работает с контентом, отрисованным JavaScript Headless-браузер Динамический контент, требующий JavaScript Поддерживает сложные взаимодействия Требует много ресурсов, сложнее масштабировать Корпоративный API Продакшн-системы, где важна надежность Управляемая инфраструктура со встроенной антибот-защитой Дополнительные расходы

Метод 1: HTTP-запросы с BeautifulSoup

Если элементы списка присутствуют в исходном HTML-ответе, извлекайте их с помощью HTTP-запросов и HTML-парсинга.

Setup

pip install requests beautifulsoup4

Basic List Crawler

импорт csv
импорт время
с сайта urllib.parse импорт urljoin
импорт requests
с сайта bs4 импорт BeautifulSoup
BASE_URL = "https://books.toscrape.com/"

def extract_items_from_page(soup, base_url):
    """Extract all items from a single page."""
    items = []
    
    для card в soup.select("article.product_pod"):
        link_el = card.select_one("h3 a")
        price_el = card.select_one("p.price_color")
        rating_el = card.select_one("p.star-rating")
        
        если не link_el или не price_el:
            продолжить
        
        item = {
            "title": link_el.get("title", "").strip(),
            "detail_url": urljoin(base_url, link_el.get("href", "")),
            "цена": price_el.get_text(strip=Правда),
            "рейтинг": extract_rating(rating_el)
        }
        items.append(item)
    
    возврат items

def extract_rating(rating_element):
    """Extract rating from CSS classes."""
    если не rating_element:
        возврат ""
    classes = rating_element.get("class", [])
    возврат " ".join(c для c в classes если c != "star-rating")

def find_next_page(soup, current_url):
    """Find and return the next page URL."""
    next_link = soup.select_one("li.next a")
    если next_link:
        возврат urljoin(current_url, next_link.get("href", ""))
    возврат Нет

def crawl_book_list(max_pages=5):
    """Main crawling function."""
    url = BASE_URL
    all_items = []
    page_count = 1
    
    в то время как url и page_count <= max_pages:
        печать(f"Crawling page {page_count}: {url}")
        
        response = requests.get(url, timeout=15)
        response.raise_for_status()
        
        soup = BeautifulSoup(response.text, "html.parser")
        items = extract_items_from_page(soup, url)
        печать(f "Found {len(items)} элементов на странице {page_count}")
        
        если не items:
            печать("No items found, stopping crawl")
            перерыв
        
        all_items.extend(items)
        url = find_next_page(soup, url)
        page_count  = 1
        
        time.sleep(1.5)
    
    печать(f"Total items collected: {len(all_items)}")
    возврат all_items

def save_results(items, filename="books_catalog.csv"):
    """Save extracted items to CSV."""
    если не items:
        печать("Нет элементов для сохранения")
        возврат
    
    с открыть(filename, "w", newline="", кодировка="utf-8") в роли f:
        writer = csv.DictWriter(f, fieldnames=items[0].keys())
        writer.writeheader()
        writer.writerows(items)
    
    печать(f"Saved {len(items)} items to {filename}")

если __name__ == "__main__":
    books = crawl_book_list(max_pages=5)
    save_results(books)

Этот код разграничивает зоны ответственности: одна функция извлекает элементы, другая находит ссылку на следующую страницу, а основной цикл отвечает за обход.

Метод 2: Автоматизация headless-браузера

Когда JavaScript динамически генерирует содержимое списка, используйте полноценный браузер. Playwright даёт хороший контроль над автоматизацией браузера.

Setup

pip install playwright
playwright install chromium

Handling Infinite Scroll

импорт csv
импорт время
с сайта playwright.sync_api импорт sync_playwright
TARGET_URL = "https://www.scrapingcourse.com/infinite-scrolling/"

def scroll_and_wait(page, scroll_pause=2.0):
    """Scroll to bottom and wait for content to load."""
    page.evaluate("window.scrollTo(0, document.body.scrollHeight)")
    time.sleep(scroll_pause)

def get_item_count(page, selector):
    """Count currently visible items."""
    возврат page.locator(selector).count()

def extract_items(страница):
    """Extract all visible items from the page."""
    items = []
    
    для card в page.locator(".product-item").все():
        item = {
            "имя": card.locator(".product-name").inner_text().strip(),
            "цена": card.locator(".product-price").inner_text().strip(),
            "url": card.locator("a").first.get_attribute("href") или ""
        }
        items.append(item)
    
    возврат items

def crawl_infinite_scroll(max_scrolls=15):
    """Crawl an infinite scrolling list."""
    с sync_playwright() в роли playwright:
        browser = playwright.chromium.launch(headless=Правда)
        page = browser.new_page()
        
        page.goto(TARGET_URL, wait_until="networkidle")
        page.wait_for_selector(".product-item", timeout=10000)
        
        previous_count = 0
        stagnant_scrolls = 0
        max_stagnant = 3
        
        для scroll_num в ассортимент(max_scrolls):
            current_count = get_item_count(page, ".product-item")
            печать(f"Scroll {scroll_num   1}: {current_count} items visible")
            
            если current_count == previous_count:
                stagnant_scrolls  = 1
                если stagnant_scrolls >= max_stagnant:
                    печать(«Новых элементов после нескольких прокруток нет, останавливаемся»)
                    перерыв
            else:
                stagnant_scrolls = 0
            
            previous_count = current_count
            scroll_and_wait(page)
        
        items = extract_items(page)
        browser.close()
        
        печать(f"Extracted {len(items)} общее количество элементов")
        возврат items

def export_to_csv(items, filename="products.csv"):
    """Export items to CSV file."""
    если не items:
        печать("No items to export")
        возврат
    
    с открыть(filename, "w", newline="", кодировка="utf-8") в роли f:
        writer = csv.DictWriter(f, fieldnames=items[0].keys())
        writer.writeheader()
        writer.writerows(items)
    
    печать(f"Exported {len(items)} items to {filename}")

если __name__ == "__main__":
    products = crawl_infinite_scroll(max_scrolls=15)
    export_to_csv(products)

Краулер останавливается, когда количество элементов остаётся неизменным в течение трёх попыток прокрутки. Это означает, что вы дошли до конца списка.

Метод 3: корпоративное решение от Bright Data

Для продакшн-систем, которым нужны масштабируемость и надежность, Bright Data предоставляет управляемую инфраструктуру без операционной рутины.

Вариант A: Web Scraper API

Bright Data Web Scraper API предлагает готовые скрейперы для популярных сайтов. Он автоматически справляется с рендерингом JavaScript, антибот-защитой и управлением прокси.

Синхронный скрапинг (для обработки до 20 URL-адресов с результатами в режиме реального времени):

импорт requests
def scrape_urls_sync(urls, dataset_id, api_key):
    """Scrape URLs using Bright Data Web Scraper API."""
    endpoint = "https://api.brightdata.com/datasets/v3/scrape"
    
    параметры = {
        "dataset_id": dataset_id,
        "формат": "json"
    }
    
    headers = {
        "Авторизация": f "Носитель {api_key}",
        "Content-Type": "application/json"
    }
    
    payload = [{"url": url} для url в urls]
    
    response = requests.post(
        endpoint,
        params=params,
        headers=headers,
        json=payload,
        timeout=60
    )
    response.raise_for_status()
    
    возврат response.json()

если __name__ == "__main__":
    API_KEY = "YOUR_API_KEY"
    DATASET_ID = "gd_YOUR_DATASET_ID"  # Get from Bright Data dashboard
    
    урлы = [
        "https://example.com/product/1",
        "https://example.com/product/2"
    ]
    
    results = scrape_urls_sync(urls, DATASET_ID, API_KEY)
    печать(f "Scraped {len(результаты)} items")

Асинхронный сбор данных (для больших объёмов):

импорт requests
импорт время
def trigger_async_scrape(urls, dataset_id, api_key):
    """Start async scrape job."""
    endpoint = "https://api.brightdata.com/datasets/v3/trigger"
    
    параметры = {
        "dataset_id": dataset_id,
        "формат": "json"
    }
    
    headers = {
        "Авторизация": f "Носитель {api_key}",
        "Content-Type": "application/json"
    }
    
    payload = [{"url": url} для url в urls]
    
    response = requests.post(
        endpoint,
        params=params,
        headers=headers,
        json=payload
    )
    response.raise_for_status()
    
    возврат response.json()["snapshot_id"]

def check_snapshot_status(snapshot_id, api_key):
    """Check if snapshot is ready."""
    endpoint = f "https://api.brightdata.com/datasets/v3/snapshot/{snapshot_id}"
    
    заголовки = {"Авторизация": f "Носитель {api_key}"}
    
    response = requests.get(endpoint, headers=headers)
    response.raise_for_status()
    
    возврат response.json()["status"]

def download_snapshot(snapshot_id, api_key):
    """Download completed snapshot data."""
    endpoint = f "https://api.brightdata.com/datasets/v3/snapshot/{snapshot_id}"
    
    параметры = {"формат": "json"}
    headers = {"Авторизация": f "Носитель {api_key}"}
    
    response = requests.get(endpoint, params=params, headers=headers)
    response.raise_for_status()
    
    возврат response.json()

def scrape_urls_async(urls, dataset_id, api_key, poll_interval=10):
    """Scrape URLs with polling."""
    snapshot_id = trigger_async_scrape(urls, dataset_id, api_key)
    печать(f"Job started. Snapshot ID: {snapshot_id}")
    
    в то время как Правда:
        status = check_snapshot_status(snapshot_id, api_key)
        печать(f"Status: {status}")
        
        если status == "ready":
            возврат download_snapshot(snapshot_id, api_key)
        elif status == "failed":
            поднять Exception("Задача скрейпинга не выполнена")
        
        time.sleep(poll_interval)

если __name__ == "__main__":
    API_KEY = "YOUR_API_KEY"
    DATASET_ID = "gd_YOUR_DATASET_ID"
    
    урлы = ["https://example.com/product/"   str(i) для i в ассортимент(100)]
    
    results = scrape_urls_async(urls, DATASET_ID, API_KEY)
    печать(f "Scraped {len(результаты)} items")

Вариант B: Browser API

Для кастомного сбора данных с полным контролем над браузером Bright Data предлагает Browser API предоставляет удаленные браузеры со встроенным обходом блокировок. Подключайтесь через Playwright или Puppeteer:

импорт asyncio
с сайта playwright.async_api импорт async_playwright
AUTH = "brd-customer-CUSTOMER_ID-zone-ZONE_NAME:ZONE_PASSWORD"
async def scrape_with_browser_api(url):
    """Scrape using Bright Data Browser API with Playwright."""
    async с async_playwright() в роли playwright:
        endpoint_url = f"wss://{@brd.superproxy.io">AUTH}@brd.superproxy.io:9222"
        
        browser = ожидайте playwright.chromium.connect_over_cdp(endpoint_url)
        
        попробуйте:
            page = ожидайте browser.new_page()
            ожидайте page.goto(url, timeout=120000)
            
            # Your custom extraction logic
            items = []
            для card в ожидайте page.locator(".product-item").все():
                item = {
                    "name": ожидайте card.locator(".name").inner_text(),
                    "price": ожидайте card.locator(".price").inner_text()
                }
                items.append(item)
            
            return items
        finally:
            await browser.close()

if __name__ == "__main__":
    results = asyncio.run(scrape_with_browser_api("https://example.com/products")) print(f"Extracted {len(results)} items")

Вариант C: Unlocker API

Для более простых случаев, когда вам нужен только HTML без блокировок, используйте Unlocker API. Возвращает чистый HTML/JSON и автоматически справляется с прокси, CAPTCHA и браузерным фингерпринтингом:

импорт requests
с сайта bs4 импорт BeautifulSoup
def fetch_with_unlocker(url, api_key, zone_name):
    """Fetch URL content using Bright Data Unlocker API."""
    endpoint = "https://api.brightdata.com/request"
    
    заголовки = {
        "Авторизация": f "Носитель {api_key}",
        "Content-Type": "application/json"
    }
    
    payload = {
        "зона": имя_зоны,
        "url": url,
        "формат": "сырой"
    }
    
    response = requests.post(endpoint, headers=headers, json=payload)
    response.raise_for_status()
    
    возврат response.text

def crawl_with_unlocker(base_url, api_key, zone_name, max_pages=5):
    """Crawl paginated list using Unlocker API."""
    all_items = []
    
    для page_num в ассортимент(1, max_pages   1):
        url = f"{base_url}?page={page_num}"
        печать(f"Fetching page {page_num}")
        
        html = fetch_with_unlocker(url, api_key, zone_name)
        soup = BeautifulSoup(html, "html.parser")
        
        items = []
        для card в soup.select(".product-card"):
            items.append({
                "имя": card.select_one(".name").get_text(strip=Правда),
                "цена": card.select_one(".price").get_text(strip=Правда)
            })
        
        если не items:
            перерыв
        
        all_items.extend(items)
    
    возврат all_items

Best Practices

Respect Rate Limits: Добавляйте задержки между запросами. Для большинства сайтов достаточно задержки в 1–2 секунды. Для более крупных обходов распределяйте запросы во времени.

Handle Errors: Проблемы с сетью и неожиданные структуры страниц неизбежны. Используйте блоки try/except и проверяйте данные перед сохранением.

Check Data Quality: Регулярно проверяйте извлеченные данные. Отсутствующие поля или обрезанные списки означают, что у ваших селекторов или логики обхода есть проблемы.

Адаптация к изменениям: Сайты часто меняют HTML-структуру. Создавайте краулеры с легко обновляемыми селекторами и регулярно их тестируйте.

Итоги

List crawling превращает разрозненные веб-данные в структурированные наборы данных. Выбор метода зависит от технологий, на которых построен целевой сайт, и ваших требований к масштабу.

Начинайте с HTTP-запросов для статических списков. Используйте headless-браузеры, когда требуется рендеринг JavaScript. Рассмотрите корпоративные API вроде Bright Data, если вам нужны надежность, масштабируемость и меньшие затраты на поддержку.

Основы остаются теми же: выявляйте повторяющиеся шаблоны, извлекайте данные последовательно и полностью обходите список