import os import logging from typing import Any from dotenv import load_dotenv logger = logging.getLogger(__name__) class ConfigError(Exception): """Ошибка валидации конфигурации""" pass class Config: def __init__(self): self._config_data = self._load() logger.info("Конфигурация успешно загружена.") def _load_replies(self, file_path: str = "storage/replies.txt"): default_replies = [["✓"]] try: reply_lines = [] with open(file_path, "r", encoding="utf-8") as replies: for line in replies: line = line.rstrip("\n") if not line: continue if " || " in line: options = [o.strip() for o in line.split(" || ")] else: options = [line] reply_lines.append(options) return reply_lines if reply_lines else default_replies except FileNotFoundError: return default_replies def _load_captcha(self, file_path: str = "storage/captcha.txt"): default_captcha = ( f"[CENTER][B]Напишите ответ:[/B]\n" f"сколько будет 20 + 20? [I][COLOR=rgb(80, 80, 78)](Капча меняется раз в месяц)[/COLOR][/I][/CENTER]" ).strip() try: captcha = "" with open(file_path, "r", encoding="utf-8") as replies: captcha = replies.read() return captcha.strip() if captcha else default_captcha except FileNotFoundError: return default_captcha def _load_captcha_answer(self): """Load captcha answer from environment variable (supports live reload)""" load_dotenv(override=True) # Reload .env file return os.getenv(key="CAPTCHA_ANSWER", default=None) def _load(self): load_dotenv() try: # Pyrogram api_id = int(os.getenv(key="API_ID", default=2040)) api_hash = os.getenv( key="API_HASH", default="b18441a1ff607e10a989891a5462e627" ) # Notification bot (aiogram) bot_token = os.getenv(key="BOT_TOKEN", default=None) admin_id_str = os.getenv(key="ADMIN_ID", default=None) admin_id = int(admin_id_str) if admin_id_str else None notify_admin = ( os.getenv(key="NOTIFY_ADMIN", default="true") ).lower() == "true" # Lolz - required fields lolz_token = os.getenv(key="LOLZ_TOKEN", default=None) if not lolz_token: raise ConfigError("LOLZ_TOKEN обязателен, но не установлен") lolz_thread_id_str = os.getenv(key="LOLZ_THREAD_ID", default=None) if not lolz_thread_id_str: raise ConfigError("LOLZ_THREAD_ID обязателен, но не установлен") try: lolz_thread_id = int(lolz_thread_id_str) except (ValueError, TypeError) as e: raise ConfigError( f"LOLZ_THREAD_ID должен быть целым числом: {lolz_thread_id_str}" ) from e # Stars stars_count = int(os.getenv(key="STARS_COUNT", default=1)) # Misc start_page = max(int(os.getenv(key="START_PAGE", default=1)), 1) check_interval = max( float(os.getenv(key="CHECK_INTERVAL", default=30)), 0.2 ) # 300r/60s для GET запросов, 60/300 = минимум api_delay = max( float(os.getenv(key="API_DELAY", default=5)), 2 ) # 30r/60s для non-GET запросов, 60/30 = минимум api_timeout = float(os.getenv(key="API_TIMEOUT", default=15)) max_retries = int(os.getenv(key="MAX_RETRIES", default=3)) enable_reply = ( os.getenv(key="ENABLE_REPLY", default="true") ).lower() == "true" skip_commented = ( os.getenv(key="SKIP_COMMENTED", default="true") ).lower() == "true" enable_auto_bump = ( os.getenv(key="ENABLE_AUTO_BUMP", default="true") ).lower() == "true" bump_check_interval = max( float(os.getenv(key="BUMP_CHECK_INTERVAL", default=3600)), 3600 ) # 1 поднятие/3600с (1ч) = минимум enable_captcha = ( os.getenv(key="ENABLE_CAPTCHA", default="true") ).lower() == "true" captcha_answer = os.getenv(key="CAPTCHA_ANSWER", default=None) captcha_check_interval = max( float(os.getenv(key="CAPTCHA_CHECK_INTERVAL", default=1800)), 300 ) # проверка обновления капчи, минимум 300с (5 минут) # Validate captcha configuration if enable_captcha and not captcha_answer: raise ConfigError( "ENABLE_CAPTCHA установлен, но CAPTCHA_ANSWER не задан" ) proxy_telegram = ( os.getenv(key="PROXY_TELEGRAM", default="false") ).lower() == "true" proxy_lolz = ( os.getenv(key="PROXY_LOLZ", default="false") ).lower() == "true" proxy_protocol = None proxy_host = None proxy_port = None proxy_port_str = None proxy_username = None proxy_password = None if proxy_telegram or proxy_lolz: proxy = os.getenv(key="PROXY", default=None) proxy_protocol = os.getenv(key="PROXY_PROTOCOL", default="socks5") proxy_host = os.getenv(key="PROXY_HOST", default=None) proxy_port_str = os.getenv(key="PROXY_PORT", default=None) proxy_username = os.getenv(key="PROXY_USERNAME", default=None) proxy_password = os.getenv(key="PROXY_PASSWORD", default=None) if proxy: logger.info("Парсинг прокси из переменной PROXY...") try: parts = proxy.split("://") if len(parts) != 2: raise ConfigError( "Неверный формат PROXY. Ожидается: protocol://[username:password@]host:port" ) proxy_protocol = parts[0] main = parts[1] # Проверяем наличие авторизации (username:password@host:port) if "@" in main: main_split = main.split("@") if len(main_split) != 2: raise ConfigError("Неверный формат авторизации в PROXY") authorization = main_split[0] server = main_split[1] # Парсим username:password (используем split с limit для поддержки : в пароле) auth_parts = authorization.split(":", 1) if len(auth_parts) != 2: raise ConfigError( "Неверный формат авторизации в PROXY. Ожидается: username:password" ) proxy_username = auth_parts[0] proxy_password = auth_parts[1] else: server = main # Парсим host:port (используем rsplit для поддержки IPv6) server_parts = server.rsplit(":", 1) if len(server_parts) != 2: raise ConfigError( "Неверный формат сервера в PROXY. Ожидается: host:port" ) proxy_host = server_parts[0] try: proxy_port = int(server_parts[1]) except ValueError as e: raise ConfigError( f"PROXY_PORT должен быть целым числом: {server_parts[1]}" ) from e logger.info( "Прокси успешно распарсен: protocol=%s, host=%s, port=%s, auth=%s", proxy_protocol, proxy_host, proxy_port, "да" if proxy_username else "нет", ) except ConfigError: raise except Exception as e: raise ConfigError( f"Произошла ошибка при парсе PROXY: {e}" ) from e else: logger.info("Загрузка прокси из отдельных переменных...") # Используем отдельные поля if not proxy_host or not proxy_port_str: raise ConfigError( "PROXY не указан, но PROXY_HOST или PROXY_PORT также отсутствуют" ) try: proxy_port = int(proxy_port_str) except (ValueError, TypeError) as e: raise ConfigError( f"PROXY_PORT должен быть целым числом: {proxy_port_str}" ) from e logger.info( "Прокси загружен: protocol=%s, host=%s, port=%s, auth=%s", proxy_protocol, proxy_host, proxy_port, "да" if proxy_username else "нет", ) # Валидация if proxy_protocol not in ["socks5", "https", "http"]: raise ConfigError( f"Формат прокси '{proxy_protocol}' не поддерживается. Укажите socks5 или https" ) if (proxy_username and not proxy_password) or ( proxy_password and not proxy_username ): raise ConfigError( "PROXY_USERNAME и PROXY_PASSWORD должны быть указаны оба или не указаны вовсе" ) # Финальная проверка if not (proxy_protocol and proxy_host and proxy_port): raise ConfigError("Прокси не полностью настроены") if proxy_telegram: logger.info("Прокси для Telegram включен") if proxy_lolz: logger.info("Прокси для Lolz API включен") return { # Pyrogram "api_id": api_id, "api_hash": api_hash, # Notification bot (aiogram) "bot_token": bot_token, "admin_id": admin_id, "notify_admin": notify_admin, # Lolz "lolz_token": lolz_token, "lolz_thread_id": lolz_thread_id, # Stars "stars_count": stars_count, # Posts "start_page": start_page, "enable_reply": enable_reply, "check_interval": check_interval, "skip_commented": skip_commented, # Misc "api_delay": api_delay, "api_timeout": api_timeout, "max_retries": max_retries, # Bump "enable_auto_bump": enable_auto_bump, "bump_check_interval": bump_check_interval, # Captcha "enable_captcha": enable_captcha, "captcha_answer": captcha_answer, "captcha_check_interval": captcha_check_interval, # Proxy "proxy_telegram": proxy_telegram, "proxy_lolz": proxy_lolz, "proxy_protocol": proxy_protocol, "proxy_host": proxy_host, "proxy_port": proxy_port, "proxy_username": proxy_username, "proxy_password": proxy_password, } except ConfigError: raise except (ValueError, TypeError) as e: logger.error(f"Invalid configuration value: {e}") raise ConfigError(f"Invalid configuration value: {e}") from e except Exception as e: logger.error(f"Unexpected error loading configuration: {e}") raise ConfigError(f"Failed to load configuration: {e}") from e def __getattr__(self, name: str) -> Any: return self._config_data.get( name, True if name == "skip_posts_with_comments" else None )