initial

src/scraper/engine.py

@@ -0,0 +1,125 @@
+# [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 src.core.models import ProductVariant
+from src.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__)
+
+    def _clean_price(self, price_str: str) -> int:
+        """[HELPER] Очищает строку цены и возвращает целое число."""
+        digits = ''.join(filter(str.isdigit, price_str))
+        return int(digits) if digits else 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=20)
+            response.raise_for_status()  # Вызовет исключение для 4xx/5xx кодов.
+            self.logger.debug(f"{log_prefix} - [COHERENCE_CHECK_PASSED] Страница успешно получена, статус {response.status_code}.")
+            return response.text
+        except requests.RequestException as e:
+            self.logger.error(f"{log_prefix} - [COHERENCE_CHECK_FAILED] Сетевая ошибка: {e}")
+            return None
+
+    def get_base_product_urls(self, catalog_url: str, run_id: str) -> List[str]:
+        """[ACTION] Собирает 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:
+            return []
+        
+        soup = BeautifulSoup(html, 'html.parser')
+        links = soup.select(self.selectors.catalog_product_link)
+        unique_urls = {urljoin(self.base_url, link.get('href')) for link in links if link.get('href')}
+        
+        self.logger.info(f"{log_prefix} - Найдено {len(unique_urls)} уникальных базовых URL.")
+        return list(unique_urls)
+
+    def get_all_variant_urls(self, base_product_urls: List[str], run_id: str) -> List[str]:
+        """[ACTION] Проходит по базовым URL и собирает URL всех их вариантов."""
+        all_variant_urls = []
+        total_base = len(base_product_urls)
+        log_prefix = f"get_variant_urls(id={run_id})"
+
+        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:
+                continue
+
+            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)
+                self.logger.debug(f"{log_prefix} - Найдено {len(variant_items)} вариантов для товара.")
+            
+            time.sleep(0.5)
+        
+        self.logger.info(f"Обнаружено всего {len(all_variant_urls)} URL вариантов для парсинга.")
+        return all_variant_urls
+
+    def scrape_variant_page(self, variant_url: str, run_id: str) -> Optional[ProductVariant]:
+        """[ACTION] Парсит страницу одного варианта и возвращает Pydantic-модель."""
+        log_prefix = f"scrape_variant(id={run_id}, url={variant_url.split('/')[-1]})"
+        html = self._fetch_page(variant_url, log_prefix)
+        if not html:
+            return None
+        
+        soup = BeautifulSoup(html, 'html.parser')
+        
+        try:
+            name_el = soup.select_one(self.selectors.product_page_name)
+            price_el = soup.select_one(self.selectors.price_block)
+            
+            if not (name_el and price_el):
+                self.logger.warning(f"{log_prefix} - [COHERENCE_CHECK_FAILED] Не найдены базовые элементы (Имя или Цена). Пропуск URL.")
+                return None
+                
+            name = name_el.get_text(strip=True)
+            price = self._clean_price(price_el.get_text(strip=True))
+            
+            volume_el = soup.select_one(self.selectors.active_volume)
+            volume = volume_el.get_text(strip=True) if volume_el else "N/A"
+
+            # [POSTCONDITION] Создаем экземпляр контракта данных.
+            product = ProductVariant(name=name, volume=volume, price=price, url=variant_url)
+            self.logger.debug(f"{log_prefix} - Успешно: '{product.name}', '{product.volume}', '{product.price}'")
+            return product
+
+        except Exception as e:
+            self.logger.error(f"{log_prefix} - Исключение при парсинге страницы: {e}", exc_info=True)
+            return None
+
+# [REFACTORING_COMPLETE]