Веб-скрейпінг у масштабі: виклики та рішення

Збір публічних даних у масштабі — це здебільшого інженерне завдання: як лишатися надійним, коли обсяги зростають, і водночас поважати сайти, які ви відвідуєте.

Цей посібник веде завдання від планування до моніторингу: правила та перевірки, бюджет запитів, коректний завантажувач із повторами, проксі, перевірка, зберігання та показники, за якими варто стежити. Код написано на Python; замініть заповнювачі власними даними.

1.Плануйте до масштабування

Чітко визначте, які публічні дані вам потрібні, перевірте умови сайту та robots.txt і вирішіть, як часто дані справді треба оновлювати. Збирати менше й рідше — найдешевший спосіб масштабуватися.

Дайте відповідь на ці запитання до першого запиту:

1. Чи публічні дані й чи можна ними користуватися? Прочитайте умови користування. Авторське право, права на бази даних і правила щодо персональних даних можуть діяти навіть для публічних сторінок.
2. Чи є API, стрічка або мапа сайту? Офіційні ендпоінти даних легші, стабільніші й бажані. Спершу скористайтеся ними.
3. Які поля вам потрібні й наскільки свіжими вони мають бути? Щоденне оновлення коштує в рази менше за щогодинне.
4. Який у вас бюджет? Визначте кількість запитів на день, ГБ на місяць і час роботи, які ви можете собі дозволити.
5. Як сайт зможе зв’язатися з вами? Позначте свій краулер зрозумілим User-Agent і контактною адресою.

2.Спершу перевірте правила

Прочитайте robots.txt, перш ніж запускати краулер на сайті, і дотримуйтесь його. Стандартна бібліотека вміє його розбирати, зокрема Crawl-delay, якщо сайт його задає. Завантажте файл своєю сесією, щоб запит також ішов через ваш проксі:

from urllib.robotparser import RobotFileParser
import requests

USER_AGENT = "my-crawler/1.0 (+https://example.com/crawler; contact: [email protected])"

proxy = "http://USERNAME:PASSWORD@HOST:PORT"
session = requests.Session()
session.proxies.update({"http": proxy, "https": proxy})
session.headers["User-Agent"] = USER_AGENT

def load_robots(base):
    r = session.get(base + "/robots.txt", timeout=(5, 20))
    robots = RobotFileParser()
    robots.parse(r.text.splitlines() if r.ok else [])      # no robots.txt: nothing is disallowed
    return robots

robots = load_robots("https://example.com")
delay = robots.crawl_delay("my-crawler") or 2               # the site's own value, or 2 seconds
print(robots.can_fetch("my-crawler", "https://example.com/page/1"), delay)

Персональні дані та закон

Публічна сторінка все одно може містити персональні дані, і до них можуть застосовуватися правила на кшталт GDPR. Збирайте лише потрібні поля, визначте мету збору, зберігайте дані не довше за необхідне й видаляйте їх на вимогу, де цього потребує закон. Це загальна інформація, а не юридична консультація: якщо ваш проєкт стосується даних людей, зверніться до юриста.

3.Контролюйте частоту запитів

Задайте загальний бюджет запитів і ліміти для кожного сайту. Додавайте затримки між запитами, обмежуйте паралельність і сповільнюйтеся, коли сайт відповідає повільно чи повертає помилки.

Арифметика проста. З одним запитом на 2 секунди на сайт 10 000 сторінок займуть близько 5,5 години. Якщо це надто повільно, зменште кількість сторінок або запитайте власника сайту про офіційну стрічку, а не підвищуйте темп. Цей обмежувач тримає мінімальну паузу для кожного хоста з невеликим випадковим розкидом, щоб потоки не стріляли синхронно:

import random
import threading
import time
from urllib.parse import urlsplit

class Throttle:
    """At least <delay> seconds between two requests to the same host."""

    def __init__(self, delay):
        self.delay = delay
        self.next_at = {}
        self.lock = threading.Lock()

    def wait(self, url):
        host = urlsplit(url).netloc
        with self.lock:
            now = time.monotonic()
            at = max(self.next_at.get(host, now), now)
            self.next_at[host] = at + self.delay + random.uniform(0, self.delay * 0.25)
        time.sleep(at - now)

throttle = Throttle(delay)

Також питайте сайт, чи змінилася сторінка, замість повторного завантаження. Сервери, що надсилають ETag, відповідають 304 Not Modified без тіла, що заощаджує час, ГБ і навантаження на сайт:

cache = {}        # url -> {"etag": ..., "body": ...}

def get_cached(url):
    headers = {}
    if url in cache and cache[url]["etag"]:
        headers["If-None-Match"] = cache[url]["etag"]
    r = session.get(url, headers=headers, timeout=(5, 20))
    if r.status_code == 304:
        return cache[url]["body"]                # unchanged: use the copy you already have
    r.raise_for_status()
    cache[url] = {"etag": r.headers.get("ETag"), "body": r.text}
    return r.text

4.Коректно обробляйте помилки

Використовуйте тайм-аути, повторюйте тимчасові помилки з експоненційною затримкою та припиняйте повтори при постійних збоях. Логуйте кожну помилку з достатнім контекстом, щоб розібратися пізніше.

Ця функція застосовує ці правила. Вона чекає довше після кожної невдалої спроби, враховує Retry-After і одразу зупиняється, якщо в доступі відмовлено:

def fetch(session, url, attempts=4):
    for attempt in range(attempts):
        throttle.wait(url)
        try:
            r = session.get(url, timeout=(5, 20))
        except requests.exceptions.RequestException:
            r = None                                          # connection problem
        if r is not None:
            if r.status_code < 400:
                return r
            if r.status_code in (404, 410):
                return r                                      # gone: nothing to retry
            if r.status_code in (401, 403, 407):
                raise PermissionError(f"{r.status_code} for {url}: stop and check the rules and your credentials")
        pause = 2 ** attempt + random.uniform(0, 1)            # 1 s, 2 s, 4 s ... plus jitter
        if r is not None and r.headers.get("Retry-After", "").isdigit():
            pause = max(pause, int(r.headers["Retry-After"]))  # the site says how long to wait
        time.sleep(pause)
    return None

Запобіжник захищає сайт і ваш бюджет. Коли надто багато останніх запитів завершуються помилкою, зупиніть краулінг цього хоста, зачекайте та з’ясуйте причину. Тиснути сильніше майже ніколи не допомагає:

from collections import deque

class CircuitBreaker:
    """Opens when too many of the latest requests failed."""

    def __init__(self, window=20, max_failures=10):
        self.results = deque(maxlen=window)
        self.max_failures = max_failures

    def record(self, ok):
        self.results.append(ok)

    @property
    def open(self):
        return list(self.results).count(False) >= self.max_failures

5.Розумно використовуйте сесії та ротацію

Змінюйте IP на вимогу між незалежними завданнями й залишайте той самий IP для процесів, що охоплюють кілька запитів. Повторно використовуйте з’єднання, де це можливо, щоб зменшити накладні витрати.

Кілька проксі розподіляють навантаження між вашими з’єднаннями. Вони не змінюють правил сайту: обмежувач вище працює для кожного хоста, тож увесь краулінг лишається в межах ліміту, скільки б проксі ви не використовували. Дайте кожному робочому потоку власну сесію та проксі й прочитайте про ротацію проксі: там розглянуто пули, паузи відпочинку та темп для кожного проксі. Цей краулер збирає все докупи:

import itertools
from concurrent.futures import ThreadPoolExecutor

proxy_urls = ["http://USER1:PASS1@HOST1:PORT1", "http://USER2:PASS2@HOST2:PORT2"]
turns = itertools.cycle(proxy_urls)
local = threading.local()
breaker = CircuitBreaker()

def thread_session():
    if not hasattr(local, "session"):
        s = requests.Session()
        proxy = next(turns)                                   # one proxy per worker thread
        s.proxies.update({"http": proxy, "https": proxy})
        s.headers["User-Agent"] = USER_AGENT
        local.session = s
    return local.session

def crawl(url):
    if breaker.open:
        return None                                           # stop quietly: investigate first
    try:
        r = fetch(thread_session(), url)
    except PermissionError as exc:
        print(exc)
        r = None
    breaker.record(r is not None)
    return r

urls = [f"https://example.com/page/{i}" for i in range(1, 9)]

with ThreadPoolExecutor(max_workers=2) as pool:               # no more workers than your plan allows
    results = list(pool.map(crawl, urls))

print(sum(r is not None for r in results), "of", len(urls), "pages fetched")

6.Зберігайте та перевіряйте дані

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

from dataclasses import dataclass

@dataclass
class Product:
    sku: str
    title: str
    price: float

def validate(row):
    try:
        product = Product(row["sku"].strip(), row["title"].strip(), float(row["price"]))
    except (KeyError, ValueError, AttributeError):
        return None
    if not product.sku or not product.title or product.price <= 0:
        return None
    return product

rows = [{"sku": "A1", "title": "Phone case", "price": "9.90"}, {"sku": "", "title": "?", "price": "-1"}]
valid = [p for p in map(validate, rows) if p]

invalid_share = 1 - len(valid) / max(len(rows), 1)
if invalid_share > 0.05:
    print(f"{invalid_share:.0%} of the rows failed validation: the page may have changed")

Зберігайте дані так, щоб повторний запуск був безпечним. Upsert за стабільним ключем оновлює змінене й не створює дублікатів, а SQLite зі стандартної бібліотеки вистачає для багатьох завдань:

import sqlite3

db = sqlite3.connect("products.db")
db.execute("CREATE TABLE IF NOT EXISTS products (sku TEXT PRIMARY KEY, title TEXT, price REAL, seen_at TEXT)")

def save(p):
    db.execute(
        "INSERT INTO products VALUES (?, ?, ?, datetime('now')) "
        "ON CONFLICT(sku) DO UPDATE SET title = excluded.title, price = excluded.price, seen_at = excluded.seen_at",
        (p.sku, p.title, p.price),
    )
    db.commit()

for product in valid:
    save(product)

Зберігайте на якийсь час і сирі відповіді. Коли парсер ламається, ви можете виправити його та повторно запустити на збережених сторінках, а не завантажувати їх знову.

7.Стежте за показниками

Краулінг, якого ви не бачите, — це краулінг, якому не можна довіряти. Рахуйте, що відбувається, і виводьте підсумок наприкінці кожного запуску:

from collections import Counter

stats = Counter()

def count(r):
    stats["requests"] += 1
    if r is None:
        stats["failed"] += 1
        return
    stats[f"status_{r.status_code}"] += 1
    stats["bytes"] += len(r.content)

for r in results:
    count(r)

print(dict(stats), f"{stats['bytes'] / 1024**2:.2f} MB")

Стежте за п’ятьма показниками: часткою невдалих запитів, часткою відповідей 429, середнім часом відповіді, використаними ГБ порівняно з бюджетом і часткою некоректних записів. Коли один із них стрибає, призупиніть краулінг і з’ясуйте причину, перш ніж продовжувати.

8.Типові помилки

• Старт на повній швидкості. Починайте повільно й підвищуйте темп, лише поки частка помилок не зростає, а правила сайту це дозволяють.
• Відповідь на 403 і 429 додатковими проксі. Ці відповіді просять зупинитися або сповільнитися. Перегляньте умови, зменште темп чи зверніться до власника сайту.
• Ігнорування robots.txt та умов. Читайте їх до запуску краулера, а не після скарги.
• Повтор усіх помилок. Повторюйте лише тимчасові помилки, з наростаючою затримкою та обмеженням.
• Розбір без перевірки. Змінена сторінка тоді заповнює вашу базу порожніми чи хибними значеннями.
• Завантаження повних сторінок, коли є ендпоінт даних. Це марнує ГБ і даремно навантажує сайт.

9.Часті запитання

Чи легальний веб-скрейпінг?

Це залежить від даних, умов сайту та чинного законодавства. Збирайте лише публічні дані, якими маєте право користуватися, і дотримуйтесь robots.txt та обмежень частоти запитів.

З якою швидкістю можна збирати дані із сайту?

Так швидко, як дозволяє сайт. Використовуйте його Crawl-delay, якщо він заданий, починайте з одного запиту кожні кілька секунд на хост і сповільнюйтеся за перших ознак навантаження.

Чи потрібен мені браузер без інтерфейсу?

Лише тоді, коли дані з’являються після виконання JavaScript. Прості HTTP-запити швидші та легші за ГБ, а офіційний API чи ендпоінт даних ще кращі.

Скільки проксі мені потрібно?

Розділіть потрібну кількість запитів на хвилину на те, що один проксі може безпечно надсилати в межах лімітів сайту, і починайте з малого. Приклад розрахунку є в матеріалі про ротацію проксі.

Як знизити рахунок за трафік?

Використовуйте ендпоінти даних замість повних сторінок, надсилайте умовні запити, не завантажуйте зображення й шрифти в браузерах, кешуйте те, що не змінюється, і збирайте рідше. У матеріалі про мобільні проксі й проксі дата-центрів показано, як оцінити ГБ.

Лишилися запитання про мобільні проксі? Відвідайте нашу сторінку FAQ щоб отримати детальні відповіді, або зверніться до нашої підтримки — ми допоможемо.