Наверняка, у всех есть групповой чат со всякими приколами, но в котором периодически проскакивают нужные вещи, которые теряются в потоке мемов, флуда и всего прочего. У меня есть такой девчачий чат, в котором мы обсуждаем проблемы, скидываем рецепты, раздаем какие-то рекомендации друг другу что посмотреть, что почитать и т.д. Однажды я полчаса листала чат в поисках новой рекомендованной книги, которую скинули между фоткой с котиком и горением от работы, тогда мне в голову пришла гениальная мысль - создать бота, который будет пересылать сообщения в нужную тему.
Нам понадобится:
блокнот для кода
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 os from dotenv import load_dotenv 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'))
Составим таблицу соответствия:
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 logging def 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_logger logger = 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:1080 HTTPS=socks5://127.0.0.1:1080
Подгружаем переменные в основной файл:
HTTP = os.getenv('HTTP') HTTPS = os.getenv('HTTPS')
и настраиваем apihelper:
apihelper.proxy = { 'http': HTTP, 'https': HTTPS }
Собственно, это все: apihelper - тень, то есть работает, пока вы не видите, все запросы на взаимодействие с телеграмом проходят через него.
Итоговый скрипт выглядит так
import os from dotenv import load_dotenv from telebot import apihelper, TeleBot from logging_config import setup_logger logger = 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_id 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()
Теперь осталось его только запустить. Запускать будем в докере. О выборе способа запуска (контейнер, демон) и другие подробности вынесу в отдельную статью
Понадобится создать 3 файла:
requirements.txt
pyTelegramBotAPI==4.18.0 python-dotenv==1.0.0 equests[socks]
Dockerfile
FROM python:3.12-slim WORKDIR /app COPY requirements.txt . RUN pip install --no-cache-dir -r requirements.txt COPY 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
Осталось добавить бота в нужный чат и выдать ему права администратора.