117 lines
4.4 KiB
Python
117 lines
4.4 KiB
Python
import os
|
|
import sys
|
|
from typing import Optional
|
|
|
|
from dotenv import load_dotenv
|
|
|
|
from database.async_db import AsyncBotDB
|
|
from helper_bot.utils.s3_storage import S3StorageService
|
|
|
|
|
|
class BaseDependencyFactory:
|
|
def __init__(self):
|
|
project_dir = os.path.dirname(
|
|
os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
|
|
)
|
|
env_path = os.path.join(project_dir, ".env")
|
|
if os.path.exists(env_path):
|
|
load_dotenv(env_path)
|
|
|
|
self.settings = {}
|
|
|
|
database_path = os.getenv("DATABASE_PATH", "database/tg-bot-database.db")
|
|
if not os.path.isabs(database_path):
|
|
database_path = os.path.join(project_dir, database_path)
|
|
|
|
self.database = AsyncBotDB(database_path)
|
|
|
|
self._load_settings_from_env()
|
|
self._init_s3_storage()
|
|
|
|
def _load_settings_from_env(self):
|
|
"""Загружает настройки из переменных окружения."""
|
|
self.settings["Telegram"] = {
|
|
"bot_token": os.getenv("BOT_TOKEN", ""),
|
|
"listen_bot_token": os.getenv("LISTEN_BOT_TOKEN", ""),
|
|
"test_bot_token": os.getenv("TEST_BOT_TOKEN", ""),
|
|
"preview_link": self._parse_bool(os.getenv("PREVIEW_LINK", "false")),
|
|
"main_public": os.getenv("MAIN_PUBLIC", ""),
|
|
"group_for_posts": self._parse_int(os.getenv("GROUP_FOR_POSTS", "0")),
|
|
"group_for_message": self._parse_int(os.getenv("GROUP_FOR_MESSAGE", "0")),
|
|
"group_for_logs": self._parse_int(os.getenv("GROUP_FOR_LOGS", "0")),
|
|
"important_logs": self._parse_int(os.getenv("IMPORTANT_LOGS", "0")),
|
|
"archive": self._parse_int(os.getenv("ARCHIVE", "0")),
|
|
"test_group": self._parse_int(os.getenv("TEST_GROUP", "0")),
|
|
}
|
|
|
|
self.settings["Settings"] = {
|
|
"logs": self._parse_bool(os.getenv("LOGS", "false")),
|
|
"test": self._parse_bool(os.getenv("TEST", "false")),
|
|
}
|
|
|
|
self.settings["Metrics"] = {
|
|
"host": os.getenv("METRICS_HOST", "0.0.0.0"),
|
|
"port": self._parse_int(os.getenv("METRICS_PORT", "8080")),
|
|
}
|
|
|
|
self.settings["S3"] = {
|
|
"enabled": self._parse_bool(os.getenv("S3_ENABLED", "false")),
|
|
"endpoint_url": os.getenv("S3_ENDPOINT_URL", ""),
|
|
"access_key": os.getenv("S3_ACCESS_KEY", ""),
|
|
"secret_key": os.getenv("S3_SECRET_KEY", ""),
|
|
"bucket_name": os.getenv("S3_BUCKET_NAME", ""),
|
|
"region": os.getenv("S3_REGION", "us-east-1"),
|
|
}
|
|
|
|
def _init_s3_storage(self):
|
|
"""Инициализирует S3StorageService если S3 включен."""
|
|
self.s3_storage = None
|
|
if self.settings["S3"]["enabled"]:
|
|
s3_config = self.settings["S3"]
|
|
if (
|
|
s3_config["endpoint_url"]
|
|
and s3_config["access_key"]
|
|
and s3_config["secret_key"]
|
|
and s3_config["bucket_name"]
|
|
):
|
|
self.s3_storage = S3StorageService(
|
|
endpoint_url=s3_config["endpoint_url"],
|
|
access_key=s3_config["access_key"],
|
|
secret_key=s3_config["secret_key"],
|
|
bucket_name=s3_config["bucket_name"],
|
|
region=s3_config["region"],
|
|
)
|
|
|
|
def _parse_bool(self, value: str) -> bool:
|
|
"""Парсит строковое значение в boolean."""
|
|
return value.lower() in ("true", "1", "yes", "on")
|
|
|
|
def _parse_int(self, value: str) -> int:
|
|
"""Парсит строковое значение в integer."""
|
|
try:
|
|
return int(value)
|
|
except (ValueError, TypeError):
|
|
return 0
|
|
|
|
def get_settings(self):
|
|
return self.settings
|
|
|
|
def get_db(self) -> AsyncBotDB:
|
|
"""Возвращает подключение к базе данных."""
|
|
return self.database
|
|
|
|
def get_s3_storage(self) -> Optional[S3StorageService]:
|
|
"""Возвращает S3StorageService если S3 включен, иначе None."""
|
|
return self.s3_storage
|
|
|
|
|
|
_global_instance = None
|
|
|
|
|
|
def get_global_instance():
|
|
"""Возвращает глобальный экземпляр BaseDependencyFactory."""
|
|
global _global_instance
|
|
if _global_instance is None:
|
|
_global_instance = BaseDependencyFactory()
|
|
return _global_instance
|