tg-gorgorod-bot/main.py

"""Главный модуль программы."""
import logging
from aiogram import Bot, Dispatcher, executor, types
import asyncio
from datetime import timedelta
from models import Base
from aiogram.utils.markdown import escape_md
import settings
import services

logging.basicConfig(level=logging.INFO)


bot = Bot(token=settings.API_TOKEN)
dp = Dispatcher(bot)


@dp.message_handler(commands=["start", "help"])
@services.auth
async def send_welcome(message: types.Message) -> None:
    r"""Справка по командам.

    Команды бота: `/start`, `/help'.
    Возвращает справку по командам. Текст берет из файла
    `static/help_msg.txt`. Доступна только пользователям из белого
    списка.

    Parameters
    ----------
    message: types.Message
        Сообщение пользователя. Библиотека aiogram.
    """
    await message.reply(
        services.get_static("static/help_msg.txt"),
        reply=False,
        parse_mode="Markdown",
    )


@dp.message_handler(commands=["channel_id"])
@services.only_admins
async def get_channel_id(message: types.Message) -> None:
    """Возвращает id канала по его имени."""
    c_id = message.text[12:]

    result = await bot.send_message(
        chat_id=c_id,
        text="ping",
        disable_notification=True,
    )

    await message.reply(
        f"Ид чата {c_id}: *{result.sender_chat.id}*", parse_mode="Markdown"
    )


@dp.message_handler(commands=["m"])
@services.only_admins
async def test_mes(message: types.Message) -> None:
    """Для тестирования сообщений."""
    q = services.get_text_from_command(message)

    await services.send_message_to_chat(message.date, 1, q, bot)


@dp.message_handler(commands=["q"])
@services.auth
async def send_question(message: types.Message) -> None:
    """Отправить вопрос через бота.

    Через бота можно отправить вопрос, который будет удален через
    DELAY_TIME_Q минут.

    """
    question = services.get_text_from_command(message)

    # если прислали только команду, без текста, показываем как надо
    # пользоваться командой
    if len(question) == 0:
        await message.reply(
            services.get_static("static/short_question_error.txt"),
            reply=False,
            parse_mode="Markdown",
        )
        return

    text = "*Внимание, вопрос!*\n\n"
    text += question

    await services.send_message_to_chat(message.date, settings.DELAY_TIME_Q, text, bot)


@dp.message_handler(commands=["p"])
@services.auth
async def send_post(message: types.Message) -> None:
    """Отправить в чат пост.

    Через бота можно отправить вопрос, который будет удален через
    DELAY_TIME_POST минут.
    """
    text = services.get_text_from_command(message)

    await services.send_message_to_chat(
        message.date, settings.DELAY_TIME_POST, text, bot
    )


@dp.message_handler(commands=["get_invite"])
@services.only_admins
async def get_invite(message: types.Message) -> None:
    """Получить токен для нового пользователя."""
    bot_info = await bot.get_me()
    bot_link = f"@{bot_info.username}"

    token = services.get_new_token()

    msg = f"Новый токен: *{token}*\n"
    msg += "Будет действителен 5 минут\n\n"
    msg += f"Чтобы вступить в чат напиши боту {escape_md(bot_link)} \n"
    msg += f"`/add_me {token}`"

    await message.reply(
        msg,
        reply=False,
        parse_mode="Markdown",
    )


@dp.message_handler(commands=["add_me"])
async def add_me(message: types.Message) -> None:
    """Добавление юзера в чат.

    Когда бот получает токен, он проверяет его на валидность (есть ли
    такой вообще, не истек ли) и высылает разовую ссылку на вступление
    в группу.
    """
    user_token = services.get_text_from_command(message)

    if not services.check_token(user_token):
        return

    services.add_new_user_to_allowed_list(message["from"]["id"])

    # генерируем ссылку, истечет через 5 минут, может вступить 1 человек
    link = await bot.create_chat_invite_link(
        chat_id=settings.CHAT_ID, expire_date=timedelta(minutes=5), member_limit=1
    )

    msg = services.get_static("static/welcome_msg_short.txt")
    await message.reply(
        msg.format(invite=link["invite_link"]), reply=False, parse_mode="Markdown"
    )

    await asyncio.sleep(10)

    msg = services.get_static("static/welcome_msg.txt")
    await message.reply(
        msg, reply=False, parse_mode="Markdown", disable_web_page_preview=True
    )


@dp.message_handler(commands=["stop"])
@services.auth
async def stop_and_panic(message: types.Message) -> None:
    """Panic-button для бота.

    Команда `/stop`
    Доступна белому списку.
    При нажатии:
        - кидает юзера в чс чата и удаляет его из чата
        - удаляет из белого списка.
    """
    u_id = message["from"]["id"]
    services.delete_user_from_allowed_list(u_id)
    await bot.kick_chat_member(
        chat_id=settings.CHAT_ID, user_id=u_id, revoke_messages=True
    )


@dp.message_handler(commands=["wlc"])
@services.only_admins
async def wlc(message: types.Message) -> None:
    """Тестирование приветственного сообщения.

    Команда `/wlc`
    Доступна только админам.
    Команда для тестирования привественного сообщения.
    """
    await message.reply(
        services.get_static("static/welcome_msg.txt"),
        reply=False,
        parse_mode="Markdown",
        disable_web_page_preview=True,
    )


@dp.message_handler(commands=["info", "version"])
@services.only_admins
async def get_info(message: types.Message) -> None:
    """Выводит информацию о боте и окружении.

    Команды: `/info`, `/version`
    Доступна только админам.
    Бот присылает свою версию, время жизни постов и вопросов.
    """
    msg = f"Версия бота: *{settings.VERSION}*\n"
    msg += f"Время жизни постов: *{settings.DELAY_TIME_POST} минут*\n"
    msg += f"Время жизни вопросов: *{settings.DELAY_TIME_Q} минут*\n"
    await message.reply(
        msg,
        reply=False,
        parse_mode="Markdown",
    )


@dp.message_handler(commands=["test"])
@services.only_admins
async def test_admins(message: types.Message):
    """Справка по командам."""
    await message.reply(
        "Справка для админов",
        reply=False,
    )


async def messages_cleanup() -> None:
    """Удаление "забытых" сообщений.

    При запуске вызывает функцию удаления старых сообщений. Постоянно
    висит в фоне и вызывает функцию удаления каждые 600 мс.

    See Also
    --------
    https://stackoverflow.com/questions/67637631/create-a-background-process-using-aiogram
    """
    while True:
        await services.delete_old_messages(bot)
        await asyncio.sleep(600)


async def on_startup(dispatcher: Dispatcher) -> None:
    """Задачи для запуска при старте бота."""
    asyncio.create_task(messages_cleanup())


if __name__ == "__main__":
    Base.metadata.create_all(services.engine)
    executor.start_polling(dp, skip_updates=True, on_startup=on_startup)