admin/kandrusyak_bot
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

refactor: centralize settings and application services

Browse Source
This commit is contained in:
kandrusyak
2026-07-27 17:47:39 +03:00
parent 248e90faf5
commit ab029a4c67
10 changed files with 296 additions and 78 deletions
Download Patch File Download Diff File Expand all files Collapse all files

87
assistant_bot/application.py
View File

@@ -12,22 +12,7 @@ from telegram.ext import (
from .ai import AIClient
from .auth import authentication_guard
from .config import (
get_assistant_mode,
get_assistant_password,
get_bot_token,
get_db_path,
get_local_timezone,
get_ollama_base_url,
get_voice_max_duration_seconds,
get_whisper_compute_type,
get_whisper_device,
get_whisper_language,
get_whisper_model,
get_yandex_cloud_folder,
get_yandex_stt_language,
get_yandex_stt_model,
)
from .config import AppSettings, load_app_settings
from .handlers import (
ask_command,
help_command,
@@ -42,6 +27,7 @@ from .handlers import (
)
from .jobs import post_init, post_shutdown
from .ollama import OllamaClient
from .services import SERVICES_KEY, ApplicationServices
from .storage import AssistantStorage
from .speech import SpeechRecognizer, SpeechTranscriber
from .yandex_ai import YandexAIClient, YandexSpeechRecognizer
@@ -55,42 +41,51 @@ def configure_logging() -> None:
logging.getLogger("httpx").setLevel(logging.WARNING)
def create_application() -> Application:
assistant_password = get_assistant_password()
storage = AssistantStorage(get_db_path())
mode = get_assistant_mode()
def create_services(settings: AppSettings) -> ApplicationServices:
storage = AssistantStorage(settings.db_path)
ai_client: AIClient
speech_recognizer: SpeechTranscriber
if settings.mode == "local":
ai_client = OllamaClient(settings.ollama_base_url)
speech_recognizer = SpeechRecognizer(
model_name=settings.whisper_model,
device=settings.whisper_device,
compute_type=settings.whisper_compute_type,
language=settings.whisper_language,
)
else:
folder_id = settings.yandex_cloud_folder
if folder_id is None:
raise RuntimeError(
"YANDEX_CLOUD_FOLDER is required in yandex mode."
)
ai_client = YandexAIClient(folder_id=folder_id)
speech_recognizer = YandexSpeechRecognizer(
folder_id=folder_id,
language=settings.yandex_stt_language,
model=settings.yandex_stt_model,
)
return ApplicationServices(
storage=storage,
ai_client=ai_client,
timezone=settings.timezone,
speech_recognizer=speech_recognizer,
voice_max_duration_seconds=settings.voice_max_duration_seconds,
assistant_password=settings.assistant_password,
)
def create_application(settings: AppSettings | None = None) -> Application:
settings = settings or load_app_settings()
application = (
Application.builder()
.token(get_bot_token())
.token(settings.bot_token)
.post_init(post_init)
.post_shutdown(post_shutdown)
.build()
)
application.bot_data["storage"] = storage
application.bot_data["assistant_password"] = assistant_password
ai_client: AIClient
speech_recognizer: SpeechTranscriber
if mode == "local":
ai_client = OllamaClient(get_ollama_base_url())
speech_recognizer = SpeechRecognizer(
model_name=get_whisper_model(),
device=get_whisper_device(),
compute_type=get_whisper_compute_type(),
language=get_whisper_language(),
)
else:
folder_id = get_yandex_cloud_folder()
ai_client = YandexAIClient(folder_id=folder_id)
speech_recognizer = YandexSpeechRecognizer(
folder_id=folder_id,
language=get_yandex_stt_language(),
model=get_yandex_stt_model(),
)
application.bot_data["ai_client"] = ai_client
application.bot_data["timezone"] = get_local_timezone()
application.bot_data["speech_recognizer"] = speech_recognizer
application.bot_data["voice_max_duration"] = get_voice_max_duration_seconds()
application.bot_data[SERVICES_KEY] = create_services(settings)
application.add_handler(TypeHandler(Update, authentication_guard), group=-1)
application.add_handler(CommandHandler("start", start))

9
assistant_bot/auth.py
View File

@@ -4,7 +4,12 @@ from telegram import Update
from telegram.constants import ChatType
from telegram.ext import ApplicationHandlerStop, ContextTypes
from .telegram_utils import get_storage, reply_markdown, require_user_id
from .telegram_utils import (
get_services,
get_storage,
reply_markdown,
require_user_id,
)
def passwords_match(candidate: str, expected: str) -> bool:
@@ -37,7 +42,7 @@ async def authentication_guard(
raise ApplicationHandlerStop
candidate = message.text.strip() if message and message.text else ""
password = str(context.application.bot_data["assistant_password"])
password = get_services(context).assistant_password
if candidate and passwords_match(candidate, password):
storage.authorize_user(user_id)
await reply_markdown(

143
assistant_bot/config.py
View File

@@ -1,5 +1,6 @@
import logging
import os
from dataclasses import dataclass
from pathlib import Path
from zoneinfo import ZoneInfo, ZoneInfoNotFoundError
@@ -46,6 +47,26 @@ MAX_AGENT_STEPS = 5
CONVERSATION_HISTORY_LIMIT = 12
@dataclass(frozen=True)
class AppSettings:
bot_token: str
assistant_password: str
db_path: Path
mode: str
timezone: ZoneInfo
voice_max_duration_seconds: int
ollama_base_url: str
ollama_model: str
yandex_cloud_folder: str | None
yandex_cloud_model: str
yandex_stt_model: str
yandex_stt_language: str
whisper_model: str
whisper_device: str
whisper_compute_type: str
whisper_language: str | None
def load_env_file(path: Path = ENV_FILE) -> None:
if not path.exists():
return
@@ -68,6 +89,128 @@ def load_env_file(path: Path = ENV_FILE) -> None:
os.environ[key] = value
def load_app_settings() -> AppSettings:
"""Load and validate the complete startup configuration once."""
load_env_file()
token = os.getenv(TOKEN_ENV_NAME)
if not token:
raise RuntimeError(
f"Set {TOKEN_ENV_NAME} in environment or .env file."
)
password = os.getenv(PASSWORD_ENV_NAME, "").strip()
if not password:
raise RuntimeError(
f"Set {PASSWORD_ENV_NAME} in environment or .env file."
)
mode = os.getenv(MODE_ENV_NAME, DEFAULT_MODE).strip().lower()
if mode not in {"local", "yandex"}:
raise RuntimeError(
f"{MODE_ENV_NAME} must be either 'local' or 'yandex', got {mode!r}."
)
raw_db_path = os.getenv(DB_ENV_NAME)
if raw_db_path:
db_path = Path(raw_db_path).expanduser()
if not db_path.is_absolute():
db_path = PROJECT_ROOT / db_path
else:
db_path = DEFAULT_DB_FILE
timezone_name = os.getenv(TIMEZONE_ENV_NAME, DEFAULT_TIMEZONE)
try:
local_timezone = ZoneInfo(timezone_name)
except ZoneInfoNotFoundError:
logger.warning(
"Unknown timezone %s, falling back to UTC",
timezone_name,
)
local_timezone = ZoneInfo("UTC")
raw_voice_duration = os.getenv(
VOICE_MAX_DURATION_ENV_NAME,
str(DEFAULT_VOICE_MAX_DURATION_SECONDS),
)
try:
voice_duration = int(raw_voice_duration)
except ValueError:
voice_duration = 0
if voice_duration <= 0:
logger.warning(
"%s must be a positive integer, using %s",
VOICE_MAX_DURATION_ENV_NAME,
DEFAULT_VOICE_MAX_DURATION_SECONDS,
)
voice_duration = DEFAULT_VOICE_MAX_DURATION_SECONDS
yandex_folder = os.getenv(
YANDEX_CLOUD_FOLDER_ENV_NAME,
"",
).strip().strip("/")
if mode == "yandex" and not yandex_folder:
raise RuntimeError(
f"Set {YANDEX_CLOUD_FOLDER_ENV_NAME} in environment or .env file."
)
yandex_model = os.getenv(
YANDEX_CLOUD_MODEL_ENV_NAME,
DEFAULT_YANDEX_CLOUD_MODEL,
).strip().strip("/")
if not yandex_model:
yandex_model = DEFAULT_YANDEX_CLOUD_MODEL
whisper_language = os.getenv(
WHISPER_LANGUAGE_ENV_NAME,
DEFAULT_WHISPER_LANGUAGE,
).strip()
return AppSettings(
bot_token=token,
assistant_password=password,
db_path=db_path,
mode=mode,
timezone=local_timezone,
voice_max_duration_seconds=voice_duration,
ollama_base_url=os.getenv(
OLLAMA_BASE_URL_ENV_NAME,
DEFAULT_OLLAMA_BASE_URL,
).rstrip("/"),
ollama_model=os.getenv(
OLLAMA_MODEL_ENV_NAME,
DEFAULT_OLLAMA_MODEL,
),
yandex_cloud_folder=yandex_folder or None,
yandex_cloud_model=yandex_model,
yandex_stt_model=os.getenv(
YANDEX_STT_MODEL_ENV_NAME,
DEFAULT_YANDEX_STT_MODEL,
).strip(),
yandex_stt_language=os.getenv(
YANDEX_STT_LANGUAGE_ENV_NAME,
DEFAULT_YANDEX_STT_LANGUAGE,
).strip(),
whisper_model=os.getenv(
WHISPER_MODEL_ENV_NAME,
DEFAULT_WHISPER_MODEL,
).strip(),
whisper_device=os.getenv(
WHISPER_DEVICE_ENV_NAME,
DEFAULT_WHISPER_DEVICE,
).strip(),
whisper_compute_type=os.getenv(
WHISPER_COMPUTE_TYPE_ENV_NAME,
DEFAULT_WHISPER_COMPUTE_TYPE,
).strip(),
whisper_language=(
None
if whisper_language.lower() == "auto"
else whisper_language or None
),
)
def get_bot_token() -> str:
load_env_file()
token = os.getenv(TOKEN_ENV_NAME)

7
assistant_bot/jobs.py
View File

@@ -6,7 +6,7 @@ from zoneinfo import ZoneInfo
from telegram.ext import Application
from .config import REMINDER_POLL_SECONDS
from .storage import AssistantStorage
from .services import SERVICES_KEY, ApplicationServices
from .telegram_utils import escape_markdown_text, markdown_code, send_markdown
from .time_utils import format_local_dt, utc_now
@@ -15,8 +15,9 @@ logger = logging.getLogger(__name__)
async def reminder_loop(application: Application) -> None:
storage: AssistantStorage = application.bot_data["storage"]
tz: ZoneInfo = application.bot_data["timezone"]
services: ApplicationServices = application.bot_data[SERVICES_KEY]
storage = services.storage
tz: ZoneInfo = services.timezone
while True:
try:

19
assistant_bot/services.py Normal file
View File

@@ -0,0 +1,19 @@
from dataclasses import dataclass
from zoneinfo import ZoneInfo
from .ai import AIClient
from .speech import SpeechTranscriber
from .storage import AssistantStorage
SERVICES_KEY = "services"
@dataclass(frozen=True)
class ApplicationServices:
storage: AssistantStorage
ai_client: AIClient
timezone: ZoneInfo
speech_recognizer: SpeechTranscriber
voice_max_duration_seconds: int
assistant_password: str

17
assistant_bot/telegram_utils.py
View File

@@ -20,6 +20,7 @@ from telegramify_markdown.config import RenderConfig
from .ai import AIClient
from .config import MAX_TELEGRAM_MESSAGE_LENGTH
from .services import SERVICES_KEY, ApplicationServices
from .speech import SpeechTranscriber
from .storage import AssistantStorage
@@ -262,24 +263,30 @@ def build_inline_results(query: str) -> list[InlineQueryResultArticle]:
]
def get_services(
context: ContextTypes.DEFAULT_TYPE,
) -> ApplicationServices:
return context.application.bot_data[SERVICES_KEY]
def get_storage(context: ContextTypes.DEFAULT_TYPE) -> AssistantStorage:
return context.application.bot_data["storage"]
return get_services(context).storage
def get_ai_client(context: ContextTypes.DEFAULT_TYPE) -> AIClient:
return context.application.bot_data["ai_client"]
return get_services(context).ai_client
def get_tz(context: ContextTypes.DEFAULT_TYPE) -> ZoneInfo:
return context.application.bot_data["timezone"]
return get_services(context).timezone
def get_speech_recognizer(context: ContextTypes.DEFAULT_TYPE) -> SpeechTranscriber:
return context.application.bot_data["speech_recognizer"]
return get_services(context).speech_recognizer
def get_voice_max_duration(context: ContextTypes.DEFAULT_TYPE) -> int:
return int(context.application.bot_data["voice_max_duration"])
return get_services(context).voice_max_duration_seconds
def command_text(context: ContextTypes.DEFAULT_TYPE) -> str: