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
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)
|
|
|