Spaces:
Running on Zero
Running on Zero
| import asyncio | |
| import logging | |
| from aiogram import Bot | |
| from aiogram.client.default import DefaultBotProperties | |
| from aiogram.enums import ParseMode | |
| from app.bot.loader import build_dispatcher | |
| from app.config import get_settings | |
| from app.core.logging import setup_logging | |
| from app.db.session import init_db, keepalive, warm_up | |
| from app.tasks.jobs import close_previous_week, notify_pending_submissions, periodic, qualify_referrals | |
| log = logging.getLogger(__name__) | |
| async def set_command_menus(bot: Bot) -> None: | |
| """Telegram's "/" suggestions: short for students, complete for admins.""" | |
| from aiogram.types import ( | |
| BotCommand, BotCommandScopeAllGroupChats, BotCommandScopeAllPrivateChats, BotCommandScopeChat, | |
| ) | |
| from app.bot import manual | |
| def commands(names: set[str]) -> list[BotCommand]: | |
| return [BotCommand(command=n, description=d) for n, d in manual.menu_commands(names)] | |
| try: | |
| await bot.set_my_commands(commands({"start", "help", "rules"}), scope=BotCommandScopeAllPrivateChats()) | |
| await bot.set_my_commands(commands({"find"}), scope=BotCommandScopeAllGroupChats()) | |
| everything = commands(manual.all_command_names() - {"admin"}) | |
| for admin_id in get_settings().admin_id_set: | |
| await bot.set_my_commands(everything, scope=BotCommandScopeChat(chat_id=admin_id)) | |
| except Exception: | |
| log.exception("Could not set the command menus (not fatal)") | |
| async def run_bot_forever() -> None: | |
| setup_logging() | |
| await init_db() | |
| await warm_up() | |
| bot = Bot( | |
| token=get_settings().bot_token, | |
| default=DefaultBotProperties(parse_mode=ParseMode.HTML), | |
| ) | |
| await set_command_menus(bot) | |
| await bot.me() # cache bot info (used by the invite link) before the first user arrives | |
| background = [ # keep references so the tasks aren't garbage-collected | |
| asyncio.create_task(keepalive()), | |
| asyncio.create_task(periodic("referrals", 1800, qualify_referrals, bot)), | |
| asyncio.create_task(periodic("season", 600, close_previous_week, bot)), | |
| asyncio.create_task(periodic("submissions", 120, notify_pending_submissions, bot, first_delay=90)), | |
| ] | |
| try: | |
| while True: | |
| try: | |
| # handle_signals=False because we run in a non-main thread | |
| await build_dispatcher().start_polling(bot, handle_signals=False) | |
| break | |
| except Exception: | |
| log.exception("Polling crashed; restarting in 5s") | |
| await asyncio.sleep(5) | |
| finally: | |
| for task in background: | |
| task.cancel() | |
| if __name__ == "__main__": | |
| asyncio.run(run_bot_forever()) | |