Бот для телеги, открывающий доступ в чатик по приглашениям.
You can not select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.

248 lines
8.0 KiB

"""Главный модуль программы."""
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)