152 lines
8.1 KiB
Python
152 lines
8.1 KiB
Python
# <MODULE name="core.settings" semantics="application_configuration" />
|
||
# <DESIGN_NOTE>
|
||
# Этот модуль является единственным источником истины для всех
|
||
# конфигурационных параметров приложения. Использует Pydantic для типизации,
|
||
# валидации и загрузки настроек из переменных окружения.
|
||
# Заменяет устаревший `src/config.py`.
|
||
# </DESIGN_NOTE>
|
||
|
||
# <IMPORTS>
|
||
import os
|
||
from pathlib import Path
|
||
from pydantic import BaseModel, Field, validator
|
||
from dotenv import load_dotenv
|
||
from typing import List
|
||
# </IMPORTS>
|
||
|
||
# <ACTION name="load_environment_variables">
|
||
# Загрузка переменных окружения из .env файла, если он существует.
|
||
load_dotenv()
|
||
# </ACTION>
|
||
|
||
# <CONSTANTS>
|
||
# <CONSTANT name="BASE_DIR" value="Path(__file__).parent.parent.parent" />
|
||
BASE_DIR = Path(__file__).resolve().parent.parent.parent
|
||
# </CONSTANTS>
|
||
|
||
# <MAIN_CONTRACT for="ScraperSelectors">
|
||
# description: "Определяет CSS-селекторы для парсинга как строгий, типизированный контракт."
|
||
# invariant: "Все поля являются обязательными непустыми строками."
|
||
# </MAIN_CONTRACT>
|
||
class ScraperSelectors(BaseModel):
|
||
# <CONFIG name="selectors_config">
|
||
catalog_product_link: str = Field(..., alias='CATALOG_PRODUCT_LINK')
|
||
variant_list_item: str = Field(..., alias='VARIANT_LIST_ITEM')
|
||
product_page_name: str = Field(..., alias='PRODUCT_PAGE_NAME')
|
||
active_volume: str = Field(..., alias='ACTIVE_VOLUME')
|
||
price_block: str = Field(..., alias='PRICE_BLOCK')
|
||
product_unavailable: str = Field(..., alias='PRODUCT_UNAVAILABLE')
|
||
# </CONFIG>
|
||
|
||
# <CONTRACT for="validate_selectors">
|
||
# description: "Валидатор, проверяющий, что селекторы не являются пустыми строками."
|
||
# </CONTRACT>
|
||
# <HELPER name="validate_selectors">
|
||
@validator('*', pre=True, allow_reuse=True)
|
||
def validate_selectors(cls, v):
|
||
if not v or not v.strip():
|
||
raise ValueError('Селектор не может быть пустым')
|
||
return v.strip()
|
||
# </HELPER>
|
||
|
||
# <MAIN_CONTRACT for="Settings">
|
||
# description: "Главный класс конфигурации приложения. Собирает все настройки в одном месте, используя переменные окружения."
|
||
# </MAIN_CONTRACT>
|
||
class Settings(BaseModel):
|
||
# <CONFIG name="parser_settings">
|
||
base_url: str = Field(default=os.getenv('PARSER_BASE_URL', 'https://elixirpeptide.ru'), description="Базовый URL сайта")
|
||
catalog_url: str = Field(default=os.getenv('PARSER_CATALOG_URL', 'https://elixirpeptide.ru/catalog/'), description="URL каталога товаров")
|
||
headers: dict = Field(default={'User-Agent': os.getenv('PARSER_USER_AGENT', 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36')})
|
||
# </CONFIG>
|
||
|
||
# <CONFIG name="output_settings">
|
||
output_dir: Path = Field(default=BASE_DIR / "price_data_final", description="Директория для сохранения результатов")
|
||
save_to_csv: bool = Field(default=os.getenv('PARSER_SAVE_TO_CSV', 'true').lower() == 'true')
|
||
save_to_db: bool = Field(default=os.getenv('PARSER_SAVE_TO_DB', 'true').lower() == 'true')
|
||
# </CONFIG>
|
||
|
||
# <CONFIG name="logging_settings">
|
||
log_to_db: bool = Field(default=os.getenv('PARSER_LOG_TO_DB', 'true').lower() == 'true')
|
||
log_dir: Path = Field(default=BASE_DIR / "logs", description="Директория для сохранения логов")
|
||
# </CONFIG>
|
||
|
||
# <CONFIG name="performance_settings">
|
||
request_timeout: int = Field(default=int(os.getenv('PARSER_TIMEOUT', 30)))
|
||
delay_between_requests: float = Field(default=float(os.getenv('PARSER_DELAY', 1.0)))
|
||
max_retries: int = Field(default=int(os.getenv('PARSER_RETRIES', 3)))
|
||
num_parser_threads: int = Field(default=int(os.getenv('PARSER_THREADS', 5)), description="Количество потоков для парсинга")
|
||
# </CONFIG>
|
||
|
||
# <CONFIG name="telegram_settings">
|
||
telegram_bot_token: str = Field(default=os.getenv('TELEGRAM_BOT_TOKEN', ''), description="Токен для Telegram бота")
|
||
telegram_chat_id: str = Field(default=os.getenv('TELEGRAM_CHAT_ID', ''), description="ID чата для отправки уведомлений")
|
||
# </CONFIG>
|
||
|
||
# <CONFIG name="rabbitmq_settings">
|
||
rabbitmq_host: str = Field(default=os.getenv('RABBITMQ_HOST', 'localhost'))
|
||
rabbitmq_port: int = Field(default=int(os.getenv('RABBITMQ_PORT', 5672)))
|
||
rabbitmq_user: str = Field(default=os.getenv('RABBITMQ_USERNAME', 'guest'))
|
||
rabbitmq_password: str = Field(default=os.getenv('RABBITMQ_PASSWORD', 'guest'))
|
||
rabbitmq_vhost: str = Field(default=os.getenv('RABBITMQ_VIRTUAL_HOST', '/'))
|
||
rabbitmq_products_queue: str = Field(default=os.getenv('RABBITMQ_PRODUCTS_QUEUE', 'price_parser.products'))
|
||
rabbitmq_logs_queue: str = Field(default=os.getenv('RABBITMQ_LOGS_QUEUE', 'price_parser.logs'))
|
||
rabbitmq_exchange: str = Field(default=os.getenv('RABBITMQ_EXCHANGE', 'price_parser.exchange'))
|
||
# </CONFIG>
|
||
|
||
# <CONFIG name="selectors_config_instance">
|
||
selectors: ScraperSelectors = ScraperSelectors(
|
||
CATALOG_PRODUCT_LINK='.product-card h4 a.product-link',
|
||
VARIANT_LIST_ITEM='.product-version-select li, .product-variants-list .variant-item',
|
||
PRODUCT_PAGE_NAME='h1.product-h1',
|
||
ACTIVE_VOLUME='.product-version-select li.active, .variant-item.active',
|
||
PRICE_BLOCK='.price-value, .product-price .price, .product-sale-box .price span',
|
||
PRODUCT_UNAVAILABLE='.product-unavailable, .out-of-stock, .unavailable, .stock.status-0',
|
||
)
|
||
# </CONFIG>
|
||
|
||
# <CONTRACT for="db_path">
|
||
# description: "Вычисляемое свойство для получения полного пути к файлу базы данных."
|
||
# </CONTRACT>
|
||
# <HELPER name="db_path" type="property">
|
||
@property
|
||
def db_path(self) -> Path:
|
||
return self.output_dir / 'parser_data.db'
|
||
# </HELPER>
|
||
|
||
# <CONTRACT for="validate_configuration">
|
||
# description: "Про<D180><D0BE>еряет ключевые параметры конфигурации на доступность и корректность."
|
||
# postconditions: "Возвращает список строк с описанием ошибок."
|
||
# </CONTRACT>
|
||
# <ACTION name="validate_configuration">
|
||
def validate_configuration(self) -> List[str]:
|
||
errors: List[str] = []
|
||
try:
|
||
self.output_dir.mkdir(parents=True, exist_ok=True)
|
||
except Exception as e:
|
||
errors.append(f"Не удается создать директорию {self.output_dir}: {e}")
|
||
|
||
try:
|
||
import requests
|
||
response = requests.head(self.base_url, timeout=self.request_timeout)
|
||
if response.status_code >= 400:
|
||
errors.append(f"Базовый URL недоступен: {self.base_url} (статус: {response.status_code})")
|
||
except Exception as e:
|
||
errors.append(f"Не удается подключиться к базовому URL {self.base_url}: {e}")
|
||
|
||
return errors
|
||
# </ACTION>
|
||
|
||
# <SINGLETON name="settings">
|
||
# Создаем единственный экземпляр настроек, который будет импортиров<D0BE><D0B2>ться
|
||
# и использоваться во всем приложении.
|
||
settings = Settings()
|
||
# </SINGLETON>
|
||
|
||
# <CONSTANTS name="export_flags">
|
||
ENABLE_RABBITMQ_EXPORT = os.getenv("ENABLE_RABBITMQ_EXPORT", "false").lower() == "true"
|
||
ENABLE_CSV_EXPORT = settings.save_to_csv
|
||
ENABLE_DATABASE_EXPORT = settings.save_to_db
|
||
# </CONSTANTS>
|
||
|
||
# <COHERENCE_CHECK status="PASSED" description="Модуль настроек полностью отрефакторен и является единственным источником конфигурации." />
|