apply patch
This commit is contained in:
@@ -10,7 +10,7 @@ import requests
|
||||
from bs4 import BeautifulSoup
|
||||
from typing import List, Optional
|
||||
|
||||
from src.core.models import ProductVariant
|
||||
from src.core.models import ProductVariant # [FIX] Импорт ProductVariant
|
||||
from src.core.settings import ScraperSelectors
|
||||
|
||||
class Scraper:
|
||||
@@ -28,8 +28,11 @@ class Scraper:
|
||||
|
||||
def _clean_price(self, price_str: str) -> int:
|
||||
"""[HELPER] Очищает строку цены и возвращает целое число."""
|
||||
self.logger.debug(f"[DEBUG] Очистка цены: '{price_str}'")
|
||||
digits = ''.join(filter(str.isdigit, price_str))
|
||||
return int(digits) if digits else 0
|
||||
cleaned_price = int(digits) if digits else 0
|
||||
self.logger.debug(f"[DEBUG] Цена после очистки: {cleaned_price}")
|
||||
return cleaned_price
|
||||
|
||||
def _fetch_page(self, url: str, request_id: str) -> Optional[str]:
|
||||
"""[HELPER] Приватный метод для скачивания HTML-содержимого страницы."""
|
||||
@@ -41,7 +44,181 @@ class Scraper:
|
||||
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}")
|
||||
self.logger.error(f"{log_prefix} - [COHERENCE_CHECK_FAILED] Сетевая ошибка при запросе {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.warning(f"{log_prefix} - Не удалось получить 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.")
|
||||
# [COHERENCE_CHECK_PASSED] Базовые URL успешно собраны.
|
||||
return list(unique_urls)
|
||||
|
||||
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
|
||||
|
||||
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)} вариантов для товара {base_url.split('/')[-1]}.")
|
||||
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
|
||||
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)
|
||||
volume_el = soup.select_one(self.selectors.active_volume) # Optional, может отсутствовать
|
||||
|
||||
# [PRECONDITION] Проверка наличия основных элементов
|
||||
if not (name_el and price_el):
|
||||
self.logger.warning(f"{log_prefix} - [COHERENCE_CHECK_FAILED] Не найдены базовые элементы (Имя продукта или Блок цены). Пропуск URL: {variant_url}.")
|
||||
return None
|
||||
# [ACTION] Извлечение данных
|
||||
name = name_el.get_text(strip=True)
|
||||
price = self._clean_price(price_el.get_text(strip=True))
|
||||
volume = volume_el.get_text(strip=True) if volume_el else "N/A"
|
||||
|
||||
# [POSTCONDITION] Создаем экземпляр контракта данных.
|
||||
# [CONTRACT_VALIDATOR] Pydantic валидация при создании модели
|
||||
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} - [COHERENCE_CHECK_FAILED] Исключение при парсинге страницы {variant_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.warning(f"{log_prefix} - Не удалось получить 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.")
|
||||
# [COHERENCE_CHECK_PASSED] Базовые URL успешно собраны.
|
||||
return list(unique_urls)
|
||||
|
||||
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
|
||||
|
||||
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)} вариантов для товара {base_url.split('/')[-1]}.")
|
||||
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
|
||||
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)
|
||||
volume_el = soup.select_one(self.selectors.active_volume) # Optional, может отсутствовать
|
||||
|
||||
# [PRECONDITION] Проверка наличия основных элементов
|
||||
if not (name_el and price_el):
|
||||
self.logger.warning(f"{log_prefix} - [COHERENCE_CHECK_FAILED] Не найдены базовые элементы (Имя продукта или Блок цены). Пропуск URL: {variant_url}.")
|
||||
return None
|
||||
# [ACTION] Извлечение данных
|
||||
name = name_el.get_text(strip=True)
|
||||
price = self._clean_price(price_el.get_text(strip=True))
|
||||
volume = volume_el.get_text(strip=True) if volume_el else "N/A"
|
||||
|
||||
# [POSTCONDITION] Создаем экземпляр контракта данных.
|
||||
# [CONTRACT_VALIDATOR] Pydantic валидация при создании модели
|
||||
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} - [COHERENCE_CHECK_FAILED] Исключение при парсинге страницы {variant_url}: {e}", exc_info=True)
|
||||
return None
|
||||
|
||||
def get_base_product_urls(self, catalog_url: str, run_id: str) -> List[str]:
|
||||
@@ -51,11 +228,9 @@ class Scraper:
|
||||
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)
|
||||
|
||||
@@ -73,7 +248,6 @@ class Scraper:
|
||||
|
||||
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)
|
||||
@@ -84,9 +258,7 @@ class Scraper:
|
||||
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
|
||||
|
||||
@@ -96,20 +268,15 @@ class Scraper:
|
||||
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"
|
||||
|
||||
|
||||
Reference in New Issue
Block a user