# [FILE] src/scraper/engine.py # [REFACTORING_TARGET] Преобразование модуля с функциями в класс Scraper. # ANCHOR: Scraper_Class_Module # Семантика: Инкапсулирует всю логику, связанную с HTTP-запросами и парсингом HTML. import logging import time from urllib.parse import urljoin import requests from bs4 import BeautifulSoup from typing import List, Optional from requests.adapters import HTTPAdapter from urllib3.util.retry import Retry from core.models import ProductVariant # [FIX] Импорт ProductVariant from core.settings import ScraperSelectors class Scraper: """ [MAIN-CONTRACT] @description: Класс, ответственный за взаимодействие с сайтом и извлечение данных. @invariant: Использует одну и ту же HTTP-сессию для всех запросов. """ def __init__(self, session: requests.Session, selectors: ScraperSelectors, base_url: str): # [INIT] Инициализация с зависимостями. self.session = session self.selectors = selectors self.base_url = base_url self.logger = logging.getLogger(self.__class__.__name__) # [ENHANCEMENT] Настройка retry стратегии для HTTP запросов self._setup_retry_strategy() def _setup_retry_strategy(self): """[HELPER] Настраивает retry стратегию для HTTP запросов.""" retry_strategy = Retry( total=3, # Максимум 3 попытки backoff_factor=1, # Экспоненциальная задержка: 1, 2, 4 секунды status_forcelist=[429, 500, 502, 503, 504], # Коды ошибок для retry allowed_methods=["HEAD", "GET", "OPTIONS"] # Разрешенные методы ) adapter = HTTPAdapter(max_retries=retry_strategy) self.session.mount("http://", adapter) self.session.mount("https://", adapter) self.logger.debug("[DEBUG] Retry стратегия настроена для HTTP запросов.") def _clean_price(self, price_str: str) -> int: """[HELPER] Очищает строку цены и возвращает целое число.""" self.logger.debug(f"[DEBUG] Очистка цены: '{price_str}'") try: # Удаляем все символы кроме цифр digits = ''.join(filter(str.isdigit, price_str)) if not digits: self.logger.warning(f"[WARNING] Не удалось извлечь цифры из цены: '{price_str}'") return 0 cleaned_price = int(digits) if cleaned_price <= 0: self.logger.warning(f"[WARNING] Некорректная цена (<= 0): {cleaned_price}") return 0 self.logger.debug(f"[DEBUG] Цена после очистки: {cleaned_price}") return cleaned_price except (ValueError, TypeError) as e: self.logger.error(f"[ERROR] Ошибка при обработке цены '{price_str}': {e}") return 0 def _fetch_page(self, url: str, request_id: str) -> Optional[str]: """[HELPER] Приватный метод для скачивания HTML-содержимого страницы.""" log_prefix = f"_fetch_page(id={request_id})" self.logger.debug(f"{log_prefix} - Запрос к URL: {url}") try: response = self.session.get(url, timeout=30) # Увеличил timeout до 30 секунд response.raise_for_status() # [ENHANCEMENT] Проверка на валидный HTML if not response.text.strip(): self.logger.warning(f"{log_prefix} - Получен пустой ответ от {url}") return None # [ENHANCEMENT] Проверка на блокировку или капчу if "captcha" in response.text.lower() or "blocked" in response.text.lower(): self.logger.error(f"{log_prefix} - [BLOCKED] Обнаружена капча или блокировка на {url}") return None self.logger.debug(f"{log_prefix} - [COHERENCE_CHECK_PASSED] Страница успешно получена, статус {response.status_code}.") return response.text except requests.exceptions.Timeout: self.logger.error(f"{log_prefix} - [TIMEOUT] Превышено время ожидания для {url}") return None except requests.exceptions.ConnectionError as e: self.logger.error(f"{log_prefix} - [CONNECTION_ERROR] Ошибка соединения для {url}: {e}") return None except requests.exceptions.HTTPError as e: self.logger.error(f"{log_prefix} - [HTTP_ERROR] HTTP ошибка для {url}: {e.response.status_code}") return None except requests.RequestException as e: self.logger.error(f"{log_prefix} - [COHERENCE_CHECK_FAILED] Сетевая ошибка при запросе {url}: {e}", exc_info=True) return None except Exception as e: self.logger.critical(f"{log_prefix} - [CRITICAL] Непредвиденная ошибка при запросе {url}: {e}", exc_info=True) return None def get_base_product_urls(self, catalog_url: str, run_id: str) -> List[str]: """[ACTION] Собирает URL всех товаров с основной страницы каталога. @pre: `catalog_url` должен быть доступен. @post: Возвращает список уникальных URL базовых продуктов. """ log_prefix = f"get_base_urls(id={run_id})" self.logger.info(f"{log_prefix} - Начало сбора базовых URL с: {catalog_url}") html = self._fetch_page(catalog_url, log_prefix) if not html: self.logger.error(f"{log_prefix} - [CRITICAL] Не удалось получить HTML страницы каталога, возвращаю пустой список.") return [] try: soup = BeautifulSoup(html, 'html.parser') links = soup.select(self.selectors.catalog_product_link) if not links: self.logger.warning(f"{log_prefix} - [WARNING] Не найдено ни одной ссылки на товар с селектором: {self.selectors.catalog_product_link}") return [] unique_urls = set() for link in links: href = link.get('href') if href: full_url = urljoin(self.base_url, href) unique_urls.add(full_url) else: self.logger.debug(f"{log_prefix} - Пропуск ссылки без href: {link}") self.logger.info(f"{log_prefix} - Найдено {len(unique_urls)} уникальных базовых URL.") # [COHERENCE_CHECK_PASSED] Базовые URL успешно собраны. return list(unique_urls) except Exception as e: self.logger.error(f"{log_prefix} - [CRITICAL] Ошибка при парсинге каталога: {e}", exc_info=True) return [] def get_all_variant_urls(self, base_product_urls: List[str], run_id: str) -> List[str]: """[ACTION] Проходит по базовым URL и собирает URL всех их вариантов. @pre: `base_product_urls` - список доступных URL продуктов. @post: Возвращает список всех URL вариантов продуктов. """ all_variant_urls = [] total_base = len(base_product_urls) log_prefix = f"get_variant_urls(id={run_id})" self.logger.info(f"{log_prefix} - Начало сбора URL вариантов для {total_base} базовых продуктов.") for i, base_url in enumerate(base_product_urls): self.logger.info(f"{log_prefix} - Обработка базового URL {i+1}/{total_base}: {base_url.split('/')[-1]}") html = self._fetch_page(base_url, f"{log_prefix}-{i+1}") if not html: self.logger.warning(f"{log_prefix} - Пропуск базового URL из-за ошибки загрузки: {base_url}") continue try: soup = BeautifulSoup(html, 'html.parser') variant_items = soup.select(self.selectors.variant_list_item) if not variant_items: self.logger.debug(f"{log_prefix} - Товар не имеет явных вариантов, добавляю базовый URL как вариант: {base_url}") all_variant_urls.append(base_url) else: for item in variant_items: variant_id = item.get('data-id') if variant_id: variant_url = f"{base_url}?product={variant_id}" all_variant_urls.append(variant_url) else: self.logger.debug(f"{log_prefix} - Пропуск варианта без data-id: {item}") self.logger.debug(f"{log_prefix} - Найдено {len(variant_items)} вариантов для товара {base_url.split('/')[-1]}.") except Exception as e: self.logger.error(f"{log_prefix} - [ERROR] Ошибка при обработке вариантов для {base_url}: {e}") # Добавляем базовый URL как fallback all_variant_urls.append(base_url) time.sleep(0.5) # [ACTION] Задержка между запросами self.logger.info(f"{log_prefix} - [COHERENCE_CHECK_PASSED] Обнаружено всего {len(all_variant_urls)} URL вариантов для парсинга.") return all_variant_urls def scrape_variant_page(self, variant_url: str, run_id: str) -> Optional[ProductVariant]: """[ACTION] Парсит страницу одного варианта и возвращает Pydantic-модель. @pre: `variant_url` должен быть доступен и содержать ожидаемые элементы. @post: Возвращает `ProductVariant` или `None` в случае ошибки парсинга. """ log_prefix = f"scrape_variant(id={run_id}, url={variant_url.split('/')[-1]})" self.logger.info(f"{log_prefix} - Начало парсинга страницы варианта.") html = self._fetch_page(variant_url, log_prefix) if not html: self.logger.warning(f"{log_prefix} - Не удалось получить HTML страницы варианта, пропуск парсинга.") return None try: soup = BeautifulSoup(html, 'html.parser') # [ENHANCEMENT] Более детальная проверка элементов name_el = soup.select_one(self.selectors.product_page_name) price_el = soup.select_one(self.selectors.price_block) volume_el = soup.select_one(self.selectors.active_volume) # Optional, может отсутствовать # [PRECONDITION] Проверка наличия основных элементов if not name_el: self.logger.warning(f"{log_prefix} - [MISSING_ELEMENT] Не найден элемент имени продукта с селектором: {self.selectors.product_page_name}") return None if not price_el: self.logger.warning(f"{log_prefix} - [MISSING_ELEMENT] Не найден элемент цены с селектором: {self.selectors.price_block}") return None # [ACTION] Извлечение данных с дополнительной валидацией name = name_el.get_text(strip=True) if not name: self.logger.warning(f"{log_prefix} - [EMPTY_DATA] Пустое имя продукта") return None price_text = price_el.get_text(strip=True) if not price_text: self.logger.warning(f"{log_prefix} - [EMPTY_DATA] Пустая цена") return None price = self._clean_price(price_text) if price <= 0: self.logger.warning(f"{log_prefix} - [INVALID_PRICE] Некорректная цена: {price}") return None volume = volume_el.get_text(strip=True) if volume_el else "N/A" # [POSTCONDITION] Создаем экземпляр контракта данных. # [CONTRACT_VALIDATOR] Pydantic валидация при создании модели try: product = ProductVariant(name=name, volume=volume, price=price, url=variant_url) self.logger.debug(f"{log_prefix} - [COHERENCE_CHECK_PASSED] Успешно распарсен вариант: '{product.name}' | '{product.volume}' | '{product.price}'") return product except Exception as e: self.logger.error(f"{log_prefix} - [VALIDATION_ERROR] Ошибка валидации ProductVariant: {e}") return None except Exception as e: self.logger.error(f"{log_prefix} - [COHERENCE_CHECK_FAILED] Исключение при парсинге страницы {variant_url}: {e}", exc_info=True) return None # [REFACTORING_COMPLETE] Дублированные методы удалены, улучшена обработка ошибок