SavePearlHarbor

Как я научила Telegram-бота наводить порядок в чате с мемами: пересылка по хештегам в соответствующую тему

—

от автора

admin

Наверняка, у всех есть групповой чат со всякими приколами, но в котором периодически проскакивают нужные вещи, которые теряются в потоке мемов, флуда и всего прочего. У меня есть такой девчачий чат, в котором мы обсуждаем проблемы, скидываем рецепты, раздаем какие-то рекомендации друг другу что посмотреть, что почитать и т.д. Однажды я полчаса листала чат в поисках новой рекомендованной книги, которую скинули между фоткой с котиком и горением от работы, тогда мне в голову пришла гениальная мысль — создать бота, который будет пересылать сообщения в нужную тему.

Нам понадобится:

  • блокнот для кода

  • VPS для проксирования, т.к. в нынешние времена беда с телегой

  • сервер, на котором будет крутиться бот (у меня это Raspberry Pi)

  • перфекционизм, оно же желание навести порядок

Первое, что нужно сделать — разработать систему хештегов. Выделяем самое важное, что нужно обязательно сохранить, например:

  • #цитаты — смешные фразочки или фотки для важных переговоров

  • #рецепт — важный раздел, нужно знать, чем кормить мужа, кроме макарошек с котлеткой

  • #план — мы живем в разных городах и периодически встречаемся, поэтому ко встрече надо иметь план, чтобы все успеть И так далее, но сразу скажу, что слишком много придумывать не нужно, иначе в тегах можно запутаться/забыть/забить

Далее создаем нужные темы и сохраняем их ID в файл переменных .env. Чтобы найти ID темы, нужно перейти в нужную тему, кликнуть на название и сохранить цифры после последнего слэша Чтобы найти ID самого чата, берем цифры после /c/ и добавляем -100

Создаем сущность нашего бота в телеграм через @BotFather: пишем /start, и следующая команда /newbot, и затем отправляем ему уникальное имя бота Не пытайтесь придумать говорящее название — вероятнее всего оно уже занято, в одной статье про микросервисы читала, что лучше давать абстрактные имена, потому что роль может поменяться, неизвестно во что может в дальнейшем превратиться ваш бот. Название нашего бота, допустим, родилось из локального мема.

Теперь перейдем к “внутреннему миру” нашего бота.

Нам нужен telebot и его класс TeleBot для создания души (объекта) и сабмодуль apihelper для настройки проксирования:

from telebot import apihelper, TeleBot

и сразу же dotenv с load_dotenv и os для подгрузки, потому что хранить переменные отдельно — это хороший тон:

import osfrom dotenv import load_dotenvload_dotenv()

Подгружаем переменные:

TOKEN = os.getenv('TOKEN')MAIN = int(os.getenv('MAIN'))QUOTES = int(os.getenv('QUOTES'))PLAN = int(os.getenv('PLAN'))RECOMMEND = int(os.getenv('RECOMMEND'))

Составим таблицу соответствия:

THREAD_MAP = {QUOTES: {'#цитат'}, #кто-то пишет "цитата", кто-то - "цитаты", поэтому оставляем часть слова    PLAN: {'#план'},    RECOMMEND: {'#рецепт', '#читать', '#смотреть'}, #сгруппировали рекомендации в одну тему}

Где есть возможность всегда использую множества вместо списков, т.к. они обрабатываются быстрее, понятное дело, что здесь это необязательно, но мало ли, будет решено добавить еще какой-то функционал.

Создаем душу:

bot = TeleBot(TOKEN)

И переходим к ее наполнению: у нас в чате демократия, поэтому пересылка происходит, когда кто угодно ответит на сообщение соответствующим тегом. Реализуем такой подход.

Будем использовать декоратор @bot.message_handler с лямбдой, который реагирует на входящие сообщения. В самом простом варианте будет выглядеть так:

@bot.message_handler(func=lambda message: message.reply_to_message is not None and '#цитат' in message.text.lower())

Сначала разберем, что у нас здесь вообще происходит

  • @bot.message_handler — реагирует на входящие сообщения

  • func=lambda message — анонимная функция для фильтрации входящих сообщений

  • message.reply_to_message is not None — проверка, что сообщение является ответом на какое-либо сообщение

  • ‘#цитат’ in message.text.lower() — проверка, что это не просто ответ, а наша метка, приводим к нижнему регистру, чтобы не держать в голове лишнюю информацию о формате тегов

Теперь нужна сама функция пересылки:

def reply_message_quotes(message):    bot.forward_message(chat_id=MAIN,            from_chat_id=message.chat.id,            message_id=message.reply_to_message.message_id,            message_thread_id=QUOTES)

  • bot.forward_message — вызываем метод пересылки сообщений с необходимыми параметрами

  • chat_id=MAIN — ID группы, куда пересылаем сообщение

  • from_chat_id=message.chat.id — ID, откуда пересылаем, получаем сразу же из сообщения, с которым работаем

  • message_id=message.reply_to_message.message_id — ID сообщения, которое пересылаем, получаем из нашего сообщения с тегом (message.reply_to_message), последний message_id говорит об id сообщения, которое мы будем пересылать (сообщение, на которое повесили тег)

  • message_thread_id=QUOTES — ID темы, в которую пересылаем сообщение

Но в нашем примере 3 группы тегов, а это значит, что мы будем копипастить 3 функции, но WET — плохая практика, поэтому выделим универсальные функции:

  • фильтр для хэндлера + получение ID темы для пересылки

  • функция пересылки

Объединим фильтр и получение ID, т.к. фильтр является часть логики получения ID.

Проверяем, что сообщение является ответом:

if message.reply_to_message is None:    return None

Чтобы получить ID темы, нам нужно пройтись по нашему словарю, и в зависимости от тега, вернуть необходимый ID. Т.к. в качестве значений у нас множества, чтобы не городить вложенные циклы, воспользуемся встроенной функцией any, которая возвращает True, если хотя бы один элемент множества проходит условие:

for thread_id, tags in THREAD_MAP.items():    if any(tag in message.text.lower() for tag in tags):        return thread_id

Таким образом, получаем первую готовую функцию:

def get_thread_id(message):    if message.reply_to_message is None:        return None    for thread_id, tags in THREAD_MAP.items():        if any(tag in message.text.lower() for tag in tags):            return thread_id

Далее приступаем к основной функции пересылки, по сути это будет та же функция reply_message_quotes(message), но в которую мы передаем thread_id, полученный из get_thread_id(message). Я предлагаю добавить сюда логирование, чтобы было видно, что наша пересылка работает + обработку исключений:

Настроим логирование, я это делаю в отдельном файле logging_config.py, кому-то это может показаться избыточным, но я частенько экспериментирую со своими ботами, поэтому настройки и переменные выношу в отдельные файлы, чтобы быстро найти нужную мне часть:

import loggingdef setup_logger():    logging.basicConfig(        level=logging.INFO,        format='%(asctime)s - %(levelname)s - %(message)s',        datefmt='%Y-%m-%d %H:%M:%S'    )    return logging.getLogger(__name__)

и не забываем импортировать в основной файл:

from logging_config import setup_loggerlogger = setup_logger()

Теперь возвращаемся к нашей функции пересылки. message.reply_to_message — укоротим, т.к. далее будем обращаться еще к атрибутам этого сообщения:

replied_message = message.reply_to_message

Мы можем пересылать разный контент, и мне было важно видеть, что именно пересылается:

        if replied_message.text:            logger.info(f"Пересылаем {replied_message.text} в {thread_id}")        else:            logger.info(f"Пересылаем {replied_message.content_type} в {thread_id}")

Затем идет наша функция:

bot.forward_message(chat_id=MAIN,        from_chat_id=message.chat.id,        message_id=message.reply_to_message.message_id,        message_thread_id=thread_id)

и логируем успешность:

logger.info("Сообщение успешно переслано")

Чтобы отследить что пошло не так, обернем это в try/except и получим в итоговом виде функцию:

def forward_to_thread(message, thread_id):    replied_message = message.reply_to_message    try:        if replied_message.text:            logger.info(f"Пересылаем {replied_message.text} в {thread_id}")        else:            logger.info(f"Пересылаем {replied_message.content_type} в {thread_id}")        bot.forward_message(chat_id=MAIN,                from_chat_id=message.chat.id,                message_id=message.reply_to_message.message_id,                message_thread_id=thread_id)        logger.info("Сообщение успешно переслано")    except Exception as e:        logger.error(f"Ошибка при пересылке: {e}", exc_info=True)

Теперь мы готовы собрать наш коротенький хэндлер в кучу:

@bot.message_handler(func=lambda message: get_thread_id(message) is not None)def send_message(message):    thread_id = get_thread_id(message)    forward_to_thread(message, thread_id)

Ну и финальная часть скрипта, настроить на постоянную прослушку чата:

if __name__ == '__main__':    logger.info("Погнали")    bot.infinity_polling()

Мы чуть не забыли про проксирование. Эта часть была делегирована (+ заслуживает отдельной статьи), поэтому предполагается, что у вас есть SSH-туннель до VPS Вносим адрес в файл с переменными:

HTTP=socks5://127.0.0.1:1080HTTPS=socks5://127.0.0.1:1080

Подгружаем переменные в основной файл:

HTTP = os.getenv('HTTP')HTTPS = os.getenv('HTTPS')

и настраиваем apihelper:

apihelper.proxy = {    'http': HTTP,    'https': HTTPS}

Собственно, это все: apihelper — тень, то есть работает, пока вы не видите, все запросы на взаимодействие с телеграмом проходят через него.

Итоговый скрипт выглядит так 
import osfrom dotenv import load_dotenvfrom telebot import apihelper, TeleBotfrom logging_config import setup_loggerlogger = setup_logger()load_dotenv()TOKEN = os.getenv('TOKEN')MAIN = int(os.getenv('MAIN'))QUOTES = int(os.getenv('QUOTES'))PLAN = int(os.getenv('PLAN'))RECOMMEND = int(os.getenv('RECOMMEND'))HTTP = os.getenv('HTTP')HTTPS = os.getenv('HTTPS')THREAD_MAP = {    QUOTES: {'#цитат'},    PLAN: {'#план'},    RECOMMEND: {'#рецепт', '#читать', '#смотреть'},}apihelper.proxy = {    'http': HTTP,    'https': HTTPS}bot = TeleBot(TOKEN)def get_thread_id(message):    if message.reply_to_message is None:        return None    for thread_id, tags in THREAD_MAP.items():        if any(tag in message.text.lower() for tag in tags):            return thread_iddef forward_to_thread(message, thread_id):    replied_message = message.reply_to_message    try:        if replied_message.text:            logger.info(f"Пересылаем {replied_message.text} в {thread_id}")        else:            logger.info(f"Пересылаем {replied_message.content_type} в {thread_id}")        bot.forward_message(chat_id=MAIN,                from_chat_id=message.chat.id,                message_id=message.reply_to_message.message_id,                message_thread_id=thread_id)        logger.info("Сообщение успешно переслано")    except Exception as e:        logger.error(f"Ошибка при пересылке: {e}", exc_info=True)    @bot.message_handler(func=lambda message: get_thread_id(message) is not None)def send_message(message):    thread_id = get_thread_id(message)    forward_to_thread(message, thread_id)if __name__ == '__main__':    logger.info("Погнали")    bot.infinity_polling()

Теперь осталось его только запустить. Запускать будем в докере. О выборе способа запуска (контейнер, демон) и другие подробности вынесу в отдельную статью

Понадобится создать 3 файла:

requirements.txt 
pyTelegramBotAPI==4.18.0python-dotenv==1.0.0equests[socks]
Dockerfile 
FROM python:3.12-slimWORKDIR /appCOPY requirements.txt .RUN pip install --no-cache-dir -r requirements.txtCOPY script.py .COPY logging_config.py .CMD ["python", "script.py"]
docker-compose.yml 
version: '3.12'services:  telegram-bot:    build: .    container_name: pizhma-bot    restart: unless-stopped    network_mode: host    env_file:      - .env    volumes:      - ./logs:/app/logs

Собираем образ, запускаем и наслаждаемся:

docker compose up -d --build

Осталось добавить бота в нужный чат и выдать ему права администратора.

ссылка на оригинал статьи https://habr.com/ru/articles/1039584/