Merge pull request 'upd bot init' (#1) from main into prepack
Reviewed-on: BotForge/qbot#1
This commit is contained in:
74
lifespan.py
74
lifespan.py
@@ -1,74 +0,0 @@
|
||||
from aiogram.types import BotCommand
|
||||
from contextlib import asynccontextmanager
|
||||
from .main import QBotApp
|
||||
from logging import getLogger
|
||||
|
||||
logger = getLogger(__name__)
|
||||
|
||||
|
||||
@asynccontextmanager
|
||||
async def default_lifespan(app: QBotApp):
|
||||
logger.debug("starting qbot app")
|
||||
|
||||
if app.config.USE_NGROK:
|
||||
try:
|
||||
from pyngrok import ngrok
|
||||
from pyngrok.conf import PyngrokConfig
|
||||
|
||||
except ImportError:
|
||||
logger.error("pyngrok is not installed")
|
||||
raise
|
||||
|
||||
tunnel = ngrok.connect(
|
||||
app.config.API_PORT,
|
||||
pyngrok_config=PyngrokConfig(auth_token=app.config.NGROK_AUTH_TOKEN),
|
||||
)
|
||||
app.config.NGROK_URL = tunnel.public_url
|
||||
|
||||
commands_captions = dict[str, list[tuple[str, str]]]()
|
||||
|
||||
for command_name, command in app.bot_commands.items():
|
||||
if command.show_in_bot_commands:
|
||||
if isinstance(command.caption, str) or command.caption is None:
|
||||
if "default" not in commands_captions:
|
||||
commands_captions["default"] = []
|
||||
commands_captions["default"].append(
|
||||
(command_name, command.caption or command_name)
|
||||
)
|
||||
else:
|
||||
for locale, description in command.caption.items():
|
||||
if locale not in commands_captions:
|
||||
commands_captions[locale] = []
|
||||
commands_captions[locale].append((command_name, description))
|
||||
|
||||
for locale, commands in commands_captions.items():
|
||||
await app.bot.set_my_commands(
|
||||
[
|
||||
BotCommand(command=command[0], description=command[1])
|
||||
for command in commands
|
||||
],
|
||||
language_code=None if locale == "default" else locale,
|
||||
)
|
||||
|
||||
await app.bot.set_webhook(
|
||||
url=f"{app.config.API_URL}/api/telegram/webhook",
|
||||
drop_pending_updates=True,
|
||||
allowed_updates=["message", "callback_query", "pre_checkout_query"],
|
||||
secret_token=app.bot_auth_token,
|
||||
)
|
||||
|
||||
logger.info("qbot app started")
|
||||
|
||||
if app.lifespan:
|
||||
async with app.lifespan(app):
|
||||
yield
|
||||
else:
|
||||
yield
|
||||
|
||||
logger.info("stopping qbot app")
|
||||
|
||||
await app.bot.delete_webhook()
|
||||
if app.config.USE_NGROK:
|
||||
ngrok.disconnect(app.config.NGROK_URL)
|
||||
ngrok.kill()
|
||||
logger.info("qbot app stopped")
|
||||
112
main.py
112
main.py
@@ -1,13 +1,15 @@
|
||||
from contextlib import asynccontextmanager
|
||||
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.types import Message, BotCommand as AiogramBotCommand
|
||||
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 logging import getLogger
|
||||
|
||||
from .config import Config
|
||||
from .fsm.db_storage import DbStorage
|
||||
@@ -18,6 +20,38 @@ from .model.descriptors import BotCommand
|
||||
from .router import Router
|
||||
|
||||
|
||||
logger = getLogger(__name__)
|
||||
|
||||
@asynccontextmanager
|
||||
async def default_lifespan(app: "QBotApp"):
|
||||
logger.debug("starting qbot app")
|
||||
|
||||
if app.lifespan_bot_init:
|
||||
|
||||
if app.config.USE_NGROK:
|
||||
app.ngrok_init()
|
||||
|
||||
await app.bot_init()
|
||||
|
||||
logger.info("qbot app started")
|
||||
|
||||
if app.lifespan:
|
||||
async with app.lifespan(app):
|
||||
yield
|
||||
else:
|
||||
yield
|
||||
|
||||
logger.info("stopping qbot app")
|
||||
|
||||
if app.lifespan_bot_init:
|
||||
await app.bot_close()
|
||||
|
||||
if app.config.USE_NGROK:
|
||||
app.ngrok_stop()
|
||||
|
||||
logger.info("qbot app stopped")
|
||||
|
||||
|
||||
class QBotApp(FastAPI):
|
||||
"""
|
||||
Main class for the QBot application
|
||||
@@ -50,6 +84,8 @@ class QBotApp(FastAPI):
|
||||
| None
|
||||
) = None,
|
||||
lifespan: Lifespan[AppType] | None = None,
|
||||
lifespan_bot_init: bool = True,
|
||||
allowed_updates: list[str] | None = None,
|
||||
*args,
|
||||
**kwargs,
|
||||
):
|
||||
@@ -61,6 +97,8 @@ class QBotApp(FastAPI):
|
||||
|
||||
user_class = DefaultUser
|
||||
|
||||
self.allowed_updates = allowed_updates or ["message", "callback_query"]
|
||||
|
||||
self.user_class = user_class
|
||||
self.entity_metadata: EntityMetadata = user_class.entity_metadata
|
||||
self.config = config
|
||||
@@ -95,7 +133,7 @@ class QBotApp(FastAPI):
|
||||
self.start_handler = bot_start
|
||||
self.bot_commands = dict[str, BotCommand]()
|
||||
|
||||
from .lifespan import default_lifespan
|
||||
self.lifespan_bot_init = lifespan_bot_init
|
||||
|
||||
super().__init__(lifespan=default_lifespan, *args, **kwargs)
|
||||
|
||||
@@ -106,7 +144,77 @@ class QBotApp(FastAPI):
|
||||
self.root_router._commands = self.bot_commands
|
||||
self.command = self.root_router.command
|
||||
|
||||
|
||||
def register_routers(self, *routers: Router):
|
||||
for router in routers:
|
||||
for command_name, command in router._commands.items():
|
||||
self.bot_commands[command_name] = command
|
||||
|
||||
|
||||
def ngrok_init(self):
|
||||
try:
|
||||
from pyngrok import ngrok
|
||||
from pyngrok.conf import PyngrokConfig
|
||||
|
||||
except ImportError:
|
||||
logger.error("pyngrok is not installed")
|
||||
raise
|
||||
|
||||
tunnel = ngrok.connect(
|
||||
self.config.API_PORT,
|
||||
pyngrok_config=PyngrokConfig(auth_token=self.config.NGROK_AUTH_TOKEN),
|
||||
)
|
||||
self.config.NGROK_URL = tunnel.public_url
|
||||
|
||||
|
||||
def ngrok_stop(self):
|
||||
try:
|
||||
from pyngrok import ngrok
|
||||
|
||||
except ImportError:
|
||||
logger.error("pyngrok is not installed")
|
||||
raise
|
||||
|
||||
ngrok.disconnect(self.config.NGROK_URL)
|
||||
ngrok.kill()
|
||||
|
||||
|
||||
|
||||
async def bot_init(self):
|
||||
|
||||
commands_captions = dict[str, list[tuple[str, str]]]()
|
||||
|
||||
for command_name, command in self.bot_commands.items():
|
||||
if command.show_in_bot_commands:
|
||||
if isinstance(command.caption, str) or command.caption is None:
|
||||
if "default" not in commands_captions:
|
||||
commands_captions["default"] = []
|
||||
commands_captions["default"].append(
|
||||
(command_name, command.caption or command_name)
|
||||
)
|
||||
else:
|
||||
for locale, description in command.caption.items():
|
||||
if locale not in commands_captions:
|
||||
commands_captions[locale] = []
|
||||
commands_captions[locale].append((command_name, description))
|
||||
|
||||
for locale, commands in commands_captions.items():
|
||||
await self.bot.set_my_commands(
|
||||
[
|
||||
AiogramBotCommand(command=command[0], description=command[1])
|
||||
for command in commands
|
||||
],
|
||||
language_code=None if locale == "default" else locale,
|
||||
)
|
||||
|
||||
await self.bot.set_webhook(
|
||||
url=f"{self.config.API_URL}/api/telegram/webhook",
|
||||
drop_pending_updates=True,
|
||||
allowed_updates=self.allowed_updates,
|
||||
secret_token=self.bot_auth_token,
|
||||
)
|
||||
|
||||
|
||||
async def bot_close(self):
|
||||
await self.bot.delete_webhook()
|
||||
|
||||
Reference in New Issue
Block a user