app/main.py - master - artyuhovs/AIFeedBackTrainingBot

app/main.py
171 строка6 KB
Codex
Выпустить версию 0.8.0
19 июл 2026, 18:59
19 июл 2026, 18:59b788979
from __future__ import annotations

import asyncio
import logging
import os
import random

from aiogram import Bot, Dispatcher
from aiogram.client.default import DefaultBotProperties
from aiogram.client.session.aiohttp import AiohttpSession
from aiogram.exceptions import TelegramNetworkError
from aiogram.utils.backoff import BackoffConfig
from aiohttp import ClientError

from app.bot.access import guard_callback, guard_message
from app.bot.commands import setup_bot_menu
from app.bot.handlers import router
from app.config import Settings, load_settings
from app.db.repositories import Repositories
from app.logging_config import setup_logging
from app.services.asr import create_asr_provider
from app.services.audio_converter import AudioConverter
from app.services.course_scheduler import CourseScheduler
from app.services.gigachat_client import GigaChatClient
from app.services.pipeline import ProcessingPipeline

POLLING_BACKOFF = BackoffConfig(min_delay=1, max_delay=30, factor=2, jitter=0.3)
TRANSIENT_ERRORS = (TelegramNetworkError, ClientError, asyncio.TimeoutError, ConnectionError)
TELEGRAM_CONNECTIVITY_INTERVAL_SECONDS = 60.0
logger = logging.getLogger(__name__)


async def create_bot(settings: Settings) -> Bot:
    if not settings.telegram_bot_token:
        raise RuntimeError("Заполните TELEGRAM_BOT_TOKEN в .env")
    session = AiohttpSession(timeout=60)
    bot = Bot(token=settings.telegram_bot_token, session=session, default=DefaultBotProperties())
    try:
        await bot.delete_webhook(drop_pending_updates=settings.telegram_drop_pending_updates)
    except Exception:
        await bot.session.close()
        raise
    return bot


def create_dispatcher(pipeline: ProcessingPipeline, repository: Repositories, settings: Settings) -> Dispatcher:
    dispatcher = Dispatcher(pipeline=pipeline, repository=repository, settings=settings)
    dispatcher.message.middleware(guard_message)
    dispatcher.callback_query.middleware(guard_callback)
    dispatcher.include_router(router)
    return dispatcher


async def connect_with_retry(settings: Settings) -> Bot:
    attempt = 0
    while True:
        try:
            return await create_bot(settings)
        except TelegramNetworkError as error:
            attempt += 1
            delay = 30 + random.uniform(0, 3)
            logger.warning(
                "telegram.connect_retry",
                extra={
                    "operation": "telegram.connect_retry",
                    "attempt": attempt,
                    "retry_delay_seconds": round(delay, 1),
                    "error_code": type(error).__name__,
                },
            )
            await asyncio.sleep(delay)


async def run_polling_with_retry(bot: Bot, dispatcher: Dispatcher) -> None:
    attempt = 0
    while True:
        try:
            await dispatcher.start_polling(bot, backoff_config=POLLING_BACKOFF, polling_timeout=30)
            return
        except asyncio.CancelledError:
            raise
        except TRANSIENT_ERRORS as error:
            attempt += 1
            delay = min(30, 2 ** min(attempt, 5)) + random.uniform(0, 0.3)
            logger.warning(
                "telegram.polling_retry",
                extra={
                    "operation": "telegram.polling_retry",
                    "attempt": attempt,
                    "retry_delay_seconds": round(delay, 1),
                    "error_code": type(error).__name__,
                },
            )
            await asyncio.sleep(delay)


async def monitor_telegram_connectivity(
    bot: Bot,
    repository: Repositories,
    *,
    interval_seconds: float = TELEGRAM_CONNECTIVITY_INTERVAL_SECONDS,
) -> None:
    while True:
        repository.schema.record_runtime_heartbeat("telegram_connectivity", "running")
        try:
            await bot.get_me()
        except asyncio.CancelledError:
            raise
        except Exception as error:
            error_code = type(error).__name__
            repository.schema.record_runtime_heartbeat(
                "telegram_connectivity",
                "failed",
                error_code,
            )
            logger.warning(
                "telegram.connectivity_failed",
                extra={
                    "operation": "telegram.connectivity_failed",
                    "error_code": error_code,
                },
            )
        else:
            repository.schema.record_runtime_heartbeat("telegram_connectivity", "ok")
        await asyncio.sleep(interval_seconds)


async def async_main() -> None:
    settings = load_settings()
    setup_logging(settings)
    settings.workdir.mkdir(parents=True, exist_ok=True, mode=0o700)
    settings.workdir.chmod(0o700)
    settings.database_path.parent.mkdir(parents=True, exist_ok=True, mode=0o700)
    settings.database_path.parent.chmod(0o700)
    repository = Repositories(settings.database_path)
    repository.schema.require_schema_head()
    settings.database_path.chmod(0o600)
    pipeline = ProcessingPipeline(
        settings=settings,
        repository=repository,
        audio_converter=AudioConverter(settings),
        asr_provider=create_asr_provider(settings),
        gigachat_client=GigaChatClient(settings),
    )
    bot = await connect_with_retry(settings)
    await setup_bot_menu(bot, repository, settings)
    dispatcher = create_dispatcher(pipeline, repository, settings)
    logger.info(
        "application.started",
        extra={"operation": "application.started"},
    )
    background_tasks = (
        asyncio.create_task(CourseScheduler(repository).run_forever(bot)),
        asyncio.create_task(monitor_telegram_connectivity(bot, repository)),
    )
    try:
        await run_polling_with_retry(bot, dispatcher)
    finally:
        for task in background_tasks:
            task.cancel()
        await asyncio.gather(*background_tasks, return_exceptions=True)
        await pipeline.aclose()


def main() -> None:
    os.umask(0o077)
    asyncio.run(async_main())


if __name__ == "__main__":
    main()