Источники данных и код скрапера

Проверка robots.txt

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

Что такое robots.txt?

Файл robots.txt содержит инструкции для поисковых роботов и скраперов о том, какие страницы можно сканировать, а какие — нет.

Как проверить robots.txt

Habr robots.txt

curl https://habr.com/robots.txt

Содержимое:

User-agent: *
Allow: /
Disallow: /api/
Disallow: /ajax/
Disallow: /backend/
Disallow: /sandbox/
Disallow: /search/

Dzen robots.txt

curl https://dzen.ru/robots.txt

Содержимое:

User-agent: *
Allow: /
Disallow: /api/
Disallow: /auth/
Disallow: /settings/

Интерпретация правил

Директива	Значение
`User-agent: *`	Правила применяются ко всем роботам
`Allow: /`	Сканирование всего сайта разрешено
`Disallow: /path/`	Сканирование указанного пути запрещено

Источники данных

Habr

URL: https://habr.com

Особенности:

Техническая направленность статей
Структурированный HTML
Наличие хабов для фильтрации по темам
Рейтинг статей

Структура URL:

https://habr.com/ru/hub/{hub_name}/page{page_number}/

Примеры хабов:

python — статьи о Python
data_science — Data Science
machine_learning — Машинное обучение

CSS-селекторы:

/* Статьи */
article.tm-articles-list__item

/* Заголовок */
a.tm-title__link

/* Автор */
a.tm-user-info__username

/* Рейтинг */
span.tm-votes-meter__value

/* Дата */
time

Dzen

URL: https://dzen.ru

Особенности:

Широкий спектр тем
Большой объём контента
Наличие тегов для фильтрации
Временные метки

Структура URL:

https://dzen.ru/news/{tag}/page{page_number}/

Примеры тегов:

python — новости о Python
technology — технологии
programming — программирование

CSS-селекторы:

/* Статьи */
div.card-compact-view

/* Заголовок */
a.card-compact-view__title-link

/* Автор */
span.card-compact-view__author

/* Время */
span.card-compact-view__time

BaseScraper класс

Полный код базового класса скрапера:

import requests
from bs4 import BeautifulSoup
import json
from abc import ABC, abstractmethod
from typing import List, Dict, Optional
import time
import logging


class BaseScraper(ABC):
    """
    Базовый класс для веб-скраперов.
    Содержит общую логику для всех скраперов.
    """

    def __init__(self, base_url: str):
        """
        Инициализация скрапера

        Args:
            base_url: Базовый URL сайта
        """
        self.base_url = base_url
        self.session = requests.Session()
        self.headers = {
            "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36",
            "Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8",
            "Accept-Language": "ru-RU,ru;q=0.9,en-US;q=0.8,en;q=0.7",
            "Accept-Encoding": "gzip, deflate, br",
            "Connection": "keep-alive",
            "Upgrade-Insecure-Requests": "1",
        }
        self.logger = logging.getLogger(__name__)

    def check_robots_txt(self) -> bool:
        """
        Проверка robots.txt

        Returns:
            True, если скрапинг разрешён
        """
        try:
            url = f"{self.base_url}/robots.txt"
            response = self.session.get(url, timeout=5, headers=self.headers)

            if response.status_code == 200:
                robots_content = response.text
                self.logger.info(f"robots.txt найден: {robots_content[:200]}...")

                # Простая проверка на наличие Disallow: /
                if "Disallow: /" in robots_content and "Allow: /" not in robots_content:
                    self.logger.warning("robots.txt запрещает сканирование")
                    return False

            return True

        except Exception as e:
            self.logger.warning(f"Не удалось проверить robots.txt: {e}")
            return True  # Продолжаем, если не удалось проверить

    def get_page(self, url: str, params: Optional[Dict] = None) -> Optional[requests.Response]:
        """
        Получение страницы

        Args:
            url: URL страницы
            params: Параметры запроса

        Returns:
            Response объект или None при ошибке
        """
        try:
            response = self.session.get(
                url,
                params=params,
                headers=self.headers,
                timeout=10
            )
            response.raise_for_status()
            return response

        except requests.RequestException as e:
            self.logger.error(f"Ошибка при запросе {url}: {e}")
            return None

    def parse_html(self, html: str) -> BeautifulSoup:
        """
        Парсинг HTML

        Args:
            html: HTML строка

        Returns:
            BeautifulSoup объект
        """
        return BeautifulSoup(html, "html.parser")

    @abstractmethod
    def extract_articles(self, soup: BeautifulSoup) -> List[Dict]:
        """
        Извлечение статей из HTML
        (абстрактный метод, должен быть реализован в наследниках)

        Args:
            soup: BeautifulSoup объект

        Returns:
            Список словарей со статьями
        """
        pass

    def scrape_page(self, url: str) -> Optional[List[Dict]]:
        """
        Скрапинг одной страницы

        Args:
            url: URL страницы

        Returns:
            Список словарей со статьями или None
        """
        response = self.get_page(url)
        if not response:
            return None

        soup = self.parse_html(response.text)
        articles = self.extract_articles(soup)

        self.logger.info(f"Извлечено {len(articles)} статей с {url}")
        return articles

    def scrape_multiple_pages(self, urls: List[str]) -> List[Dict]:
        """
        Скрапинг нескольких страниц

        Args:
            urls: Список URL

        Returns:
            Список всех статей
        """
        all_articles = []

        # Проверка robots.txt перед началом
        if not self.check_robots_txt():
            self.logger.error("Скрапинг запрещён robots.txt")
            return []

        for i, url in enumerate(urls, 1):
            self.logger.info(f"Обработка страницы {i}/{len(urls)}: {url}")
            articles = self.scrape_page(url)

            if articles:
                all_articles.extend(articles)

            # Задержка между запросами
            if i < len(urls):
                time.sleep(1)

        return all_articles

    def save_to_json(self, data: List[Dict], filename: str) -> None:
        """
        Сохранение данных в JSON

        Args:
            data: Список данных
            filename: Имя файла
        """
        try:
            with open(filename, "w", encoding="utf-8") as file:
                json.dump(data, file, ensure_ascii=False, indent=4)
            self.logger.info(f"Данные сохранены в {filename}")
        except Exception as e:
            self.logger.error(f"Ошибка при сохранении в JSON: {e}")

    def save_to_database(self, data: List[Dict], db_name: str, table_name: str) -> None:
        """
        Сохранение данных в базу данных SQLite

        Args:
            data: Список данных
            db_name: Имя базы данных
            table_name: Имя таблицы
        """
        import sqlite3

        if not data:
            self.logger.warning("Нет данных для сохранения")
            return

        try:
            # Создание таблицы
            columns = list(data[0].keys())
            column_defs = ", ".join([f"{col} TEXT" for col in columns])

            connection = sqlite3.connect(db_name)
            cursor = connection.cursor()

            cursor.execute(f"""
                CREATE TABLE IF NOT EXISTS {table_name} (
                    id INTEGER PRIMARY KEY AUTOINCREMENT,
                    {column_defs}
                )
            """)

            # Вставка данных
            placeholders = ", ".join(["?"] * len(columns))
            insert_query = f"INSERT INTO {table_name} ({', '.join(columns)}) VALUES ({placeholders})"

            for article in data:
                values = [str(article.get(col, "")) for col in columns]
                cursor.execute(insert_query, values)

            connection.commit()
            connection.close()
            self.logger.info(f"Данные сохранены в {db_name} (таблица: {table_name})")

        except Exception as e:
            self.logger.error(f"Ошибка при сохранении в базу данных: {e}")

    def run(self, urls: List[str], output_file: str = None) -> List[Dict]:
        """
        Основной метод запуска скрапера

        Args:
            urls: Список URL для скрапинга
            output_file: Имя файла для сохранения (опционально)

        Returns:
            Список собранных данных
        """
        self.logger.info(f"Запуск скрапера для {len(urls)} страниц")
        data = self.scrape_multiple_pages(urls)
        self.logger.info(f"Собрано {len(data)} статей")

        if output_file:
            self.save_to_json(data, output_file)

        return data

Примеры использования

Базовое использование

from scrapers import HabrScraper, DzenScraper

# Скрапинг Habr
habr_scraper = HabrScraper()
habr_urls = ["https://habr.com/ru/hub/python/page1/"]
habr_data = habr_scraper.run(habr_urls, "data/habr.json")

# Скрапинг Dzen
dzen_scraper = DzenScraper()
dzen_urls = ["https://dzen.ru/news/python/page1/"]
dzen_data = dzen_scraper.run(dzen_urls, "data/dzen.json")

Использование с логированием

import logging
from scrapers import HabrScraper

# Настройка логирования
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)

scraper = HabrScraper()
data = scraper.run(["https://habr.com/ru/hub/python/page1/"])

Сохранение в базу данных

from scrapers import HabrScraper

scraper = HabrScraper()
data = scraper.run(["https://habr.com/ru/hub/python/page1/"])

scraper.save_to_database(data, "articles.db", "habr")

Обработка ошибок

Retries (повторные попытки)

import time
from functools import wraps


def retry(max_retries=3, delay=1):
    """Декоратор для повторных попыток"""
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            for attempt in range(max_retries):
                try:
                    return func(*args, **kwargs)
                except Exception as e:
                    if attempt == max_retries - 1:
                        raise
                    time.sleep(delay)
            return None
        return wrapper
    return decorator


class BaseScraper(ABC):
    @retry(max_retries=3, delay=2)
    def get_page(self, url: str) -> Optional[requests.Response]:
        """Получение страницы с retry"""
        # ...

Лучшие практики

1. Уважение к серверу

# Всегда добавляйте задержки между запросами
time.sleep(1)  # 1 секунда между запросами

# Не перегружайте сервер
max_concurrent_requests = 5

2. Обработка ошибок

try:
    data = scraper.run(urls)
except Exception as e:
    logger.error(f"Ошибка скрапинга: {e}")
    # Сохранение прогресса или уведомление

3. Валидация данных

def validate_article(article: Dict) -> bool:
    """Валидация статьи"""
    required_fields = ["заголовок", "ссылка", "источник"]
    return all(field in article for field in required_fields)

# Использование
valid_articles = [a for a in articles if validate_article(a)]

4. Логирование

import logging

logger = logging.getLogger(__name__)

logger.info("Запуск скрапера")
logger.warning("Мало данных собрано")
logger.error("Ошибка при запросе")