110 lines
3.4 KiB
Python
110 lines
3.4 KiB
Python
from typing import Annotated, Callable, Any
|
|
from typing_extensions import Doc
|
|
from aiogram import Bot, Dispatcher
|
|
from aiogram.client.default import DefaultBotProperties
|
|
from aiogram.types import Message
|
|
from aiogram.utils.callback_answer import CallbackAnswerMiddleware
|
|
from aiogram.utils.i18n import I18n
|
|
from fastapi import FastAPI
|
|
from fastapi.applications import Lifespan, AppType
|
|
from secrets import token_hex
|
|
|
|
from .config import Config
|
|
from .fsm.db_storage import DbStorage
|
|
from .middleware.telegram import AuthMiddleware, I18nMiddleware
|
|
from .model.user import UserBase
|
|
from .model.entity_metadata import EntityMetadata
|
|
from .model.descriptors import BotCommand
|
|
from .router import Router
|
|
|
|
|
|
class QBotApp(FastAPI):
|
|
"""
|
|
Main class for the QBot application
|
|
"""
|
|
|
|
def __init__[UserType: UserBase](
|
|
self,
|
|
user_class: (
|
|
Annotated[
|
|
type[UserType], Doc("User class that will be used in the application")
|
|
]
|
|
| None
|
|
) = None,
|
|
config: Config | None = None,
|
|
bot_start: (
|
|
Annotated[
|
|
Callable[
|
|
[
|
|
Annotated[
|
|
Callable[[Message, Any], None],
|
|
Doc("Default handler for the start command"),
|
|
],
|
|
Message,
|
|
Any,
|
|
],
|
|
None,
|
|
],
|
|
Doc("Handler for the start command"),
|
|
]
|
|
| None
|
|
) = None,
|
|
lifespan: Lifespan[AppType] | None = None,
|
|
*args,
|
|
**kwargs,
|
|
):
|
|
if config is None:
|
|
config = Config()
|
|
|
|
if user_class is None:
|
|
from .model.default_user import DefaultUser
|
|
|
|
user_class = DefaultUser
|
|
|
|
self.user_class = user_class
|
|
self.entity_metadata: EntityMetadata = user_class.entity_metadata
|
|
self.config = config
|
|
self.lifespan = lifespan
|
|
self.bot = Bot(
|
|
token=self.config.TELEGRAM_BOT_TOKEN,
|
|
default=DefaultBotProperties(parse_mode="HTML"),
|
|
)
|
|
|
|
dp = Dispatcher(storage=DbStorage())
|
|
|
|
i18n = I18n(path="locales", default_locale="en", domain="messages")
|
|
i18n_middleware = I18nMiddleware(user_class=user_class, i18n=i18n)
|
|
i18n_middleware.setup(dp)
|
|
dp.callback_query.middleware(CallbackAnswerMiddleware())
|
|
|
|
from .bot.handlers.start import router as start_router
|
|
|
|
dp.include_router(start_router)
|
|
|
|
from .bot.handlers.menu.main import router as main_menu_router
|
|
|
|
auth = AuthMiddleware(user_class=user_class)
|
|
main_menu_router.message.middleware.register(auth)
|
|
main_menu_router.callback_query.middleware.register(auth)
|
|
dp.include_router(main_menu_router)
|
|
|
|
self.dp = dp
|
|
|
|
self.bot_auth_token = token_hex(128)
|
|
|
|
self.start_handler = bot_start
|
|
self.bot_commands = dict[str, BotCommand]()
|
|
|
|
from .lifespan import default_lifespan
|
|
|
|
super().__init__(lifespan=default_lifespan, *args, **kwargs)
|
|
|
|
from .api_route.telegram import router as telegram_router
|
|
|
|
self.include_router(telegram_router, prefix="/api/telegram", tags=["telegram"])
|
|
|
|
def register_routers(self, *routers: Router):
|
|
for router in routers:
|
|
for command_name, command in router._commands.items():
|
|
self.bot_commands[command_name] = command
|