Автоматический сбор статистики сайтов по расписанию на Python с разбором HTML и записью данных в таблицу

Цифры, собранные раз в месяц вручную, показывают почти ничего: между замерами теряется всё, что было в промежутке. Скрипт, снимающий показатели ежедневно в одно время, за квартал даёт таблицу из девяноста строк с трендом, провалами и сезонностью. Разбираем, как написать сборщик на Python, чем запускать его по расписанию, в каком формате хранить результат и какие правила соблюдать при обращении к чужим сайтам.

Какие данные стоит собирать и где проходит граница допустимого

Автоматический сбор имеет смысл там, где значение меняется медленно, а отслеживать его нужно долго: цена у конкурента, позиция по запросу, число объявлений, курс или тариф. Разовый замер бесполезен, ценность появляется только в ряду наблюдений.

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

1. Файл robots.txt в корне сайта описывает, какие разделы владелец просит не обходить, и его игнорирование это сознательное нарушение заявленных условий;
2. Частота обращений держится в разумных пределах, поскольку один запрос в сутки не создаёт нагрузки, а запрос каждые две секунды в течение часа даёт тысячи обращений и приводит к блокировке адреса;
3. Состав данных ограничивается сведениями, опубликованными для неограниченного круга лиц, а персональные данные не собирают и не хранят, даже если они технически доступны на странице;
4. В заголовке User-Agent указывается назначение скрипта и способ связи, чтобы администратор чужого сайта мог задать вопрос, а не блокировать адрес молча.

HEADERS = {
    "User-Agent": "stat-collector/1.0 (+https://example.local/bot; Адрес электронной почты защищен от спам-ботов. Для просмотра адреса в браузере должен быть включен Javascript.)"
}

Когда нужен разбор HTML, а когда есть готовый интерфейс

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

Признаки стандартные: страница обновляет цифры без перезагрузки, в адресе встречаются слова api, json или graphql, есть документация либо выгрузка в формате CSV.

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

Разница в усилиях колоссальная: разбор JSON это несколько строк и предсказуемая структура, разбор HTML это навигация по классам, которая ломается при любом изменении вёрстки.

Если готового источника нет, остаётся разметка, и для её разбора берут BeautifulSoup:

pip install requests beautifulsoup4 lxml

Каркас надёжного сборщика с таймаутом, повторами и журналом

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

import logging
import time
from email.utils import parsedate_to_datetime

import requests

logging.basicConfig(
    filename="/var/log/stat-collector.log",
    level=logging.INFO,
    format="%(asctime)s %(levelname)s %(message)s",
)

HEADERS = {"User-Agent": "stat-collector/1.0 (+https://example.local/bot)"}
TIMEOUT = (10, 30)


def retry_after(header: str | None, default: float) -> float:
    """Секунды ожидания из Retry-After: число либо дата в формате HTTP."""
    if not header:
        return default
    header = header.strip()
    if header.isdigit():
        return float(header)
    try:
        moment = parsedate_to_datetime(header)
    except (TypeError, ValueError):
        logging.warning("не разобрал Retry-After: %r, беру %.0f с", header, default)
        return default
    return max(0.0, (moment.timestamp() - time.time()))


def fetch(url: str, attempts: int = 3, pause: float = 5.0):
    for n in range(1, attempts + 1):
        try:
            r = requests.get(url, headers=HEADERS, timeout=TIMEOUT)
            if r.status_code == 200:
                return r
            if r.status_code == 429:
                wait = retry_after(r.headers.get("Retry-After"), pause * n * 2)
                logging.warning("429 от %s, ждём %.0f с", url, wait)
                time.sleep(wait)
                continue
            logging.error("код %s для %s", r.status_code, url)
            return None
        except requests.RequestException as exc:
            logging.warning("попытка %s из %s для %s: %s", n, attempts, url, exc)
            time.sleep(pause * n)
    return None

Разобрать здесь стоит каждое решение. Таймаут передаётся парой чисел, где первое ограничивает время установки соединения, а второе время ожидания ответа. Без таймаута один зависший запрос останавливает весь скрипт до срабатывания системных ограничений.

Код 429 означает, что сервер попросил снизить темп, и в заголовке Retry-After часто указан конкретный срок. Игнорировать его значит гарантированно получить блокировку.

Формат значения допускает два варианта: число секунд либо дату в текстовом формате HTTP, например с указанием дня недели, месяца и часового пояса. Прямое преобразование в число на втором варианте завершится исключением, и сборщик упадёт именно тогда, когда сервер просит подождать. Поэтому строку из цифр принимают как секунды, остальное читают как дату и вычисляют остаток до неё, а при неудаче берут паузу по умолчанию.

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

Журнал в файл обязателен: скрипт запускается ночью, и без записи утром не понять, собралось что-нибудь или нет.

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

Разбор страницы и извлечение числовых значений

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

from bs4 import BeautifulSoup


def parse_price(html: str):
    soup = BeautifulSoup(html, "lxml")
    node = soup.select_one("div.product-card span.price")
    if node is None:
        logging.error("элемент цены не найден, вёрстка изменилась")
        return None
    raw = node.get_text(strip=True)
    cleaned = (
        raw.replace("\u00a0", "")
           .replace(" ", "")
           .replace("\u2009", "")
           .replace(",", ".")
    )
    digits = "".join(ch for ch in cleaned if ch.isdigit() or ch == ".")
    try:
        return float(digits)
    except ValueError:
        logging.error("не удалось разобрать значение: %r", raw)
        return None

Селектор берут конкретный, но без лишней глубины: путь из семи вложенных элементов ломается при любом изменении структуры, а уникальный класс живёт годами.

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

Отдельная тонкость с динамической разметкой: если цифры подставляются скриптом, в исходном HTML их нет и запрос вернёт пустую оболочку. Признак простой, значение видно в браузере, но отсутствует в тексте ответа. Лечится это поиском источника данных, а не запуском браузера.

Хранение результатов в CSV или SQLite с защитой от дублей и пропусков дней

Для ряда наблюдений из пары колонок достаточно CSV: файл читается любой таблицей и не требует служб.

import csv
import os
from datetime import date

CSV_PATH = "/srv/stats/prices.csv"
FIELDS = ["date", "source", "value"]


def append_row(row: dict):
    new = not os.path.exists(CSV_PATH)
    with open(CSV_PATH, "a", newline="", encoding="utf-8") as fh:
        writer = csv.DictWriter(fh, fieldnames=FIELDS)
        if new:
            writer.writeheader()
        writer.writerow(row)

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

SQLite решает задачу штатно через первичный ключ:

import sqlite3

DB_PATH = "/srv/stats/stats.db"


def init_db():
    with sqlite3.connect(DB_PATH) as db:
        db.execute("""
            CREATE TABLE IF NOT EXISTS metrics (
                day     TEXT NOT NULL,
                source  TEXT NOT NULL,
                value   REAL,
                PRIMARY KEY (day, source)
            )
        """)


def upsert(day: str, source: str, value: float):
    with sqlite3.connect(DB_PATH) as db:
        db.execute("""
            INSERT INTO metrics (day, source, value) VALUES (?, ?, ?)
            ON CONFLICT(day, source) DO UPDATE SET value = excluded.value
        """, (day, source, value))

Конструкция ON CONFLICT делает запись идемпотентной: сколько бы раз ни отработал скрипт за сутки, в таблице останется одно значение за день, и сбор после сбоя можно перезапускать.

Выгрузка в таблицу остаётся одной командой:

sqlite3 -header -csv /srv/stats/stats.db \
  "SELECT * FROM metrics ORDER BY day DESC LIMIT 40" > /srv/stats/last40.csv

Запуск по расписанию через cron с блокировкой и уведомлением о сбое

Расписание задаётся одной строкой. Ежедневный сбор в четыре утра:

0 4 * * * /srv/stats/run.sh >> /var/log/stat-collector.cron.log 2>&1

Обёртка run.sh нужна для трёх вещей: блокировка от одновременного запуска, проверка результата и уведомление при сбое.

#!/usr/bin/env bash
set -uo pipefail

LOCK=/run/stat-collector.lock
exec 200>"$LOCK"
if ! flock -n 200; then
  echo "предыдущий запуск ещё идёт"
  exit 0
fi

/usr/bin/python3 /srv/stats/collect.py
code=$?

if [ "$code" -ne 0 ]; then
  curl -s -X POST -H 'Content-Type: application/json' \
    -d "{\"text\":\"сбор статистики завершился с кодом $code\"}" \
    "$(cat /etc/stat-alert.conf)" >/dev/null
fi

Блокировка через flock защищает от наложения запусков, когда сбор занял больше отведённого интервала и второй экземпляр стартовал поверх первого.

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

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

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

from datetime import datetime, timezone

def today() -> str:
    return datetime.now(timezone.utc).astimezone().date().isoformat()

Второй вопрос, пропуски. Если сервер был выключен, строка за день не появится и на графике возникнет дыра. Заполнять её интерполяцией не стоит, поскольку восстановленное значение неотличимо от настоящего. Разумнее помечать пропуск отдельным признаком.

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

import sys
day = sys.argv[1] if len(sys.argv) > 1 else today()

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

Контроль качества данных и признаки того, что сбор сломался

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

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

def sanity_check(day: str, source: str, value: float) -> bool:
    with sqlite3.connect(DB_PATH) as db:
        rows = db.execute(
            "SELECT value FROM metrics WHERE source=? ORDER BY day DESC LIMIT 7",
            (source,),
        ).fetchall()
    if not rows:
        return True
    prev = [r[0] for r in rows]
    avg = sum(prev) / len(prev)
    if avg and abs(value - avg) / avg > 0.5:
        logging.warning("аномалия: %s = %s при среднем %.2f", source, value, avg)
        return False
    return True

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

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

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