Spaces:
Running on Zero
Running on Zero
File size: 2,680 Bytes
b37ebd1 7982d01 4e64e70 b37ebd1 a14f7f7 b37ebd1 7982d01 b37ebd1 a14f7f7 7982d01 d896946 4e64e70 d896946 7982d01 d896946 b37ebd1 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 | 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())
|