cit-is-bot-backend/bot/main.py

import os

from aiogram import Bot, Dispatcher
from aiogram.client.default import DefaultBotProperties
from aiogram.enums import ParseMode
from aiogram.filters import CommandStart
from aiogram.types import User, Message, PollAnswer, InputPollOption
from apscheduler.schedulers.asyncio import AsyncIOScheduler

import config
import database
import models


dp = Dispatcher()


@dp.message(CommandStart())
async def command_start_handler(
        message: Message,
):
    config.Redis.set('chat_id', message.chat.id)
    await message.answer('Чат успешно зарегистрирован!')


bot = Bot(
    token=config.Telegram.token,
    default=DefaultBotProperties(
        parse_mode=ParseMode.HTML,
    ),
)

scheduler = AsyncIOScheduler()


async def create_or_update_local_user_data(
        telegram_user: User,
) -> models.User:
    database_user = await database.Users.insert_or_update_user(
        telegram_id=telegram_user.id,
        first_name=telegram_user.first_name,
        last_name=telegram_user.last_name,
        username=telegram_user.username,
    )
    user_profile_photos = await telegram_user.get_profile_photos(
        limit=1,
    )
    try:
        photo_size = user_profile_photos.photos[0][0]
        file = await bot.get_file(photo_size.file_id)
        dir_path = os.path.join(
            config.Main.cwd,
            'static',
            'images',
            'users',
        )
        file_path = os.path.join(
            dir_path,
            '%s%s' % (
                database_user.id,
                os.path.splitext(file.file_path)[1],
            ),
        )
        os.makedirs(
            name=dir_path,
            exist_ok=True,
        )
        await bot.download_file(
            file_path=file.file_path,
            destination=file_path,
        )
    except Exception:
        pass
    return database_user


async def send_mood_poll():
    poll_schema = await database.PollSchemas.get_poll_schema_by_name(
        name='mood',
    )
    poll_options = await database.PollOptions.get_poll_options(
        poll_schema=poll_schema,
    )
    message = await bot.send_poll(
        chat_id=config.Redis.get('chat_id'),
        question=poll_schema.question,
        options=[
            InputPollOption(
                text=poll_option.name,
            )
            for poll_option in poll_options
        ],
        is_anonymous=False,
        allows_multiple_answers=False,
        is_closed=False,
        disable_notification=True,
    )
    await database.Polls.insert_poll(
        telegram_message_id=message.message_id,
        telegram_poll_id=message.poll.id,
        poll_schema=poll_schema,
    )


# async def send_lunch_poll():
#     poll_schema = await database.PollSchemas.get_poll_schema_by_name(
#         name='lunch',
#     )
#     message = await bot.send_poll(
#         chat_id=config.Redis.get('chat_id'),
#         question=poll_schema.question,
#         options=get_poll_options(poll_schema),
#         is_anonymous=False,
#         allows_multiple_answers=False,
#         is_closed=False,
#         disable_notification=True,
#     )
#     await database.Polls.insert_poll(
#         telegram_message_id=message.message_id,
#         telegram_poll_id=message.poll.id,
#         poll_schema=poll_schema,
#     )


# async def send_lunch_delivery_poll() -> Poll:
#     message = await bot.send_poll(
#         chat_id=config.Redis.get('chat_id'),
#         question='Где будем заказывать?',
#         options=[
#             InputPollOption(
#                 text='Dark Side',
#             ),
#             InputPollOption(
#                 text='Самокат',
#             ),
#             InputPollOption(
#                 text='...',
#             ),
#             InputPollOption(
#                 text='...',
#             ),
#         ],
#         is_anonymous=False,
#         allows_multiple_answers=False,
#         is_closed=False,
#         disable_notification=True,
#     )
#     return message.poll


@dp.poll_answer()
async def get_poll_answer(
        poll_answer: PollAnswer,
):
    user = await create_or_update_local_user_data(poll_answer.user)
    for poll in await database.Polls.get_polls():
        if poll.telegram_poll_id == poll_answer.poll_id:
            if poll.is_complete:
                await bot.delete_message(
                    chat_id=config.Redis.get('chat_id'),
                    message_id=poll.telegram_message_id,
                )
                break
            poll_options = await database.PollOptions.get_poll_options(
                poll_schema=poll.poll_schema,
                ordinals=poll_answer.option_ids,
            )
            await database.PollAnswers.insert_or_update_poll_answer(
                poll=poll,
                user=user,
                poll_options=poll_options,
            )
            break


async def on_startup(
        dispatcher: Dispatcher,
):
    # ### TEST
    await send_mood_poll()
    # TEST ###
    # scheduler.add_job(
    #     func=send_mood_poll,
    #     trigger='cron',
    #     day_of_week='mon-fri',
    #     hour=22,
    #     minute=35,
    # )
    # scheduler.add_job(
    #     func=send_lunch_poll,
    #     trigger='cron',
    #     day_of_week='mon-fri',
    #     hour=23,
    #     minute=55,
    # )
    # scheduler.add_job(
    #     func=get_lunch_poll_result,
    #     trigger='cron',
    #     day_of_week='mon-fri',
    #     hour=23,
    #     minute=19,
    # )
    scheduler.start()


dp.startup.register(on_startup)