97 lines
3.6 KiB
Python
97 lines
3.6 KiB
Python
"""Конфигурация приложения.
|
||
|
||
Значения читаются из переменных окружения / файла .env (см. .env.example).
|
||
Реальных секретов в коде быть не должно.
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
from functools import lru_cache
|
||
from typing import Annotated
|
||
from urllib.parse import quote_plus
|
||
from zoneinfo import ZoneInfo
|
||
|
||
from pydantic import Field, field_validator
|
||
from pydantic_settings import BaseSettings, NoDecode, SettingsConfigDict
|
||
|
||
|
||
class Settings(BaseSettings):
|
||
"""Настройки, общие для бота и админки."""
|
||
|
||
model_config = SettingsConfigDict(
|
||
env_file=".env",
|
||
env_file_encoding="utf-8",
|
||
extra="ignore",
|
||
case_sensitive=False,
|
||
)
|
||
|
||
app_env: str = "dev"
|
||
|
||
bot_token: str = Field(..., description="Токен бота от @BotFather")
|
||
|
||
db_host: str = "localhost"
|
||
db_port: int = 5432
|
||
db_name: str = "dispel_evil"
|
||
db_user: str = "de_admin"
|
||
db_password: str = ""
|
||
db_pool_min: int = 1
|
||
db_pool_max: int = 10
|
||
db_statement_timeout_ms: int = 15000
|
||
|
||
#: Основной чат клана (единственный; контуры разделяются отдельными .env).
|
||
chat_id: int = 0
|
||
#: Доп. чат для служебных уведомлений/логов (опционально).
|
||
alert_chat_id: int = 0
|
||
|
||
#: Telegram-ID, которым доступ разрешён в обход проверки членства.
|
||
#: NoDecode — чтобы pydantic-settings не пытался JSON-декодировать значение
|
||
#: (поддерживаем формат "1,2,3" и пустое значение через валидатор ниже).
|
||
admin_ids: Annotated[list[int], NoDecode] = Field(default_factory=list)
|
||
|
||
timezone: str = "Europe/Moscow"
|
||
|
||
#: Файл-heartbeat для Docker healthcheck (обновляется ботом периодически).
|
||
heartbeat_path: str = "/tmp/dispelevilbot.heartbeat"
|
||
|
||
@field_validator("chat_id", "alert_chat_id", mode="before")
|
||
@classmethod
|
||
def _empty_int_to_zero(cls, value: object) -> object:
|
||
"""Пустое значение в .env трактуем как 0 (а не ошибку валидации)."""
|
||
if value is None or value == "":
|
||
return 0
|
||
return value
|
||
|
||
@field_validator("admin_ids", mode="before")
|
||
@classmethod
|
||
def _parse_admin_ids(cls, value: object) -> object:
|
||
"""Поддержать формат 'ADMIN_IDS=1,2,3' в дополнение к JSON-списку."""
|
||
if value is None or value == "":
|
||
return []
|
||
if isinstance(value, str):
|
||
cleaned = value.strip().strip("[]")
|
||
if not cleaned:
|
||
return []
|
||
return [int(item.strip()) for item in cleaned.split(",") if item.strip()]
|
||
return value
|
||
|
||
@property
|
||
def is_prod(self) -> bool:
|
||
return self.app_env.strip().lower() in {"prod", "production"}
|
||
|
||
@property
|
||
def tzinfo(self) -> ZoneInfo:
|
||
return ZoneInfo(self.timezone)
|
||
|
||
@property
|
||
def dsn(self) -> str:
|
||
"""Строка подключения psycopg для async-пула (креды URL-экранируются)."""
|
||
user = quote_plus(self.db_user)
|
||
password = quote_plus(self.db_password)
|
||
return f"postgresql://{user}:{password}" f"@{self.db_host}:{self.db_port}/{self.db_name}"
|
||
|
||
|
||
@lru_cache
|
||
def get_settings() -> Settings:
|
||
"""Вернуть singleton настроек."""
|
||
return Settings() # type: ignore[call-arg]
|