135 lines
5.2 KiB
Python
135 lines
5.2 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"
|
||
|
||
#: ClickHouse (аналитика событий). CLICKHOUSE_ENABLED=false — полностью выключить.
|
||
clickhouse_enabled: bool = True
|
||
clickhouse_host: str = "localhost"
|
||
clickhouse_port: int = 8123
|
||
clickhouse_db: str = "analytics"
|
||
clickhouse_user: str = "default"
|
||
clickhouse_password: str = ""
|
||
|
||
#: Батчинг записи событий: размер батча, интервал сброса и предел буфера.
|
||
analytics_batch_size: int = 200
|
||
analytics_flush_interval: float = 5.0
|
||
analytics_buffer_max: int = 5000
|
||
|
||
#: Сколько секунд бот держит шаблоны сообщений в памяти (правки из админки
|
||
#: применяются без перезапуска). 0 — читать БД при каждом обращении.
|
||
template_cache_ttl: float = 45.0
|
||
|
||
@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]
|
||
|
||
|
||
class AdminSettings(Settings):
|
||
"""Настройки сервиса админки.
|
||
|
||
Наследует общие поля (окружение, Postgres, таймзона, ClickHouse), но не требует
|
||
``BOT_TOKEN``: на этапе каркаса он не нужен, а появится в проверке подписи
|
||
Telegram-логина (Блок 4.2).
|
||
"""
|
||
|
||
bot_token: str = ""
|
||
|
||
#: Адрес и порт, которые слушает uvicorn внутри контейнера.
|
||
admin_host: str = "0.0.0.0"
|
||
admin_port: int = 8000
|
||
|
||
|
||
@lru_cache
|
||
def get_admin_settings() -> AdminSettings:
|
||
"""Вернуть singleton настроек админки."""
|
||
return AdminSettings() # type: ignore[call-arg]
|