331 lines
14 KiB
Python
331 lines
14 KiB
Python
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"):
|
|
script_ad_reply = [
|
|
o.strip()
|
|
for o in str(
|
|
"Скрипт для авто-раздачи :da: [ad][UNFURL]https://lolz.team/threads/10187517/[/UNFURL][/ad] || Сиськи, сливы 18+ [ad][UNFURL]https://lolz.team/threads/10187517/[/UNFURL][/ad] || Твой помощник для авто-раздачи :animeyessir: [ad][UNFURL]https://lolz.team/threads/10187517/[/UNFURL][/ad]"
|
|
).split(" || ")
|
|
]
|
|
default_replies = [["✓"], script_ad_reply]
|
|
|
|
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)
|
|
reply_lines.append(script_ad_reply)
|
|
|
|
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"]:
|
|
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
|
|
)
|