Хотел отложить написание второй части трилогии в долгий ящик, но судя по просмотрам первого эпизода - тема создания Телеграм-ботов все еще актуальна на Хабр.
Во второй части сфокусируемся на разработке бизнес-логики бота. В нашем проекте, для взаимодействия с Telegram, будем использовать библиотеку Aiogram. Для Python написано достаточное количество библиотек для работы с ТГ, но Aiogram, наверное, самая популярная. Советую прочитать руководство по работе с Aiogram от Groosha - для меня это была основная теоретическая база. Кроме непосредственной работы с функционалом библиотеки, советую обратить внимание на раздел "Роутеры. Структура" - я буду следовать этой логике при создании бота.
Мы будем работать в асинхронном режиме. Это достаточно сложная тема для неподготовленного читателя и не могу сказать, что я освоил все аспекты работы Python в асинхронном режиме. Для поверхностного погружения рекомендую почитать серию статей от @vlakirна Хабр.
Не буду описывать процесс регистрации бота в ТГ - это достаточно простая процедура, чтобы тратить на нее время тут. В интернете очень много материала на эту тему. Будем считать, что у вас на руках уже есть Token вашего бота от BotFather. Также, в ТГ необходимо создать публичный канал и добавить бота в администраторы этого канала. Вопрос открытия комментариев к постам - на ваше усмотрение.
Бот должен стать промежуточным этапом между публикацией новости в канал. По расписанию или запросу скрипт должен пробегать по RSS-ленте сайта и проверять ее на наличие новых, ранее не опубликованных новостей. В случае, если появилось что-то новое, направлять сообщение, состоящее из заголовка, краткого содержания и ссылки на новость в ТГ-канал администратору с двумя inline-кнопками - "Удалить" и "Отправить в ChatGPT". В случае нажатия на кнопку "ChatGPT" скрипт направляет текст новости (после работы процедуры parse, созданной в Первой части материала) на сервер ChatGPT с заданным Prompt и удаляет исходную новость из ТГ-канала. Ответ от "ChatGPT" возвращается новым сообщением в бот с единственной кнопкой - "Отправить в канал", после нажатия на которую сообщение улетает в канал и удаляется из бота.
Дополнение для вредных
Мне хотелось на этапе "Отправить в канал" реализовать возможность редактирования материала после ChatGPT, но к сожалению, это невозможно на уровне логики работы Telegram — администратор не может править сообщения от бота. Текст сообщения можно будет отредактировать уже в канале стандартным инструментарием работы с сообщениями Telegram. Конечно, это не очень удобно, но если делать быстро и канал небольшой — вполне нормально:) Возможно вы сможете найти более изящное решение этой проблемы.
Сохраняемся
Для реализации подобного функционала понадобится, как минимум, где-то хранить информацию о ранее опубликованных новостях. Можно пойти по пути@DimaFromMaiс его агрегатором новостей и хранить информацию в оперативной памяти. Но нужно понимать, что при любом сбое/перезагрузке сервера вы теряете всю историю и заново получаете всю ленту новостей, включая ранее опубликованное. Для нашего примера я решил чуть-чуть усложнить задачу и хранить все в базе данных. Выбор пал на SQLite, поскольку это наиболее простая и легковесная база для небольших проектов.
В первой части мы написали парсер RSS-ленты Мотор'a. При запуске скрипта он может пройтись по списку всех новостей и вывести на печать заголовок, ссылку на новость и другую информацию из ленты. Сосредоточимся на ссылках на новости - нам нужно определить уникальный ключ для хранения в нашей базе данных, который можно получить из ссылок, при этом не сохраняя всю ссылку как ключ (это очень жестоко по отношению к базе):
def main(): rss_link = 'https://motor.ru/exports/rss' rss_text=requests.get(rss_link).text #Загружаем RSS-ленту rss = feedparser.parse(rss_text)#Парсим RSS ленту for news in rss.entries[::-1]: print(news['link'])
Результат будет выглядеть примерно так:
... https://motor.ru/news/wey-25-10-2023.htm https://motor.ru/news/renault-volvo-25-10-2023.htm https://motor.ru/news/buhanka-parts-25-10-2023.htm https://motor.ru/news/geely-monjaro-phev-25-10-2023.htm ...
К сожалению, Мотор не присваивает цифровой код своим новостям, но очевидно, что в качестве уникального ключа можно использовать все, что идет после "news/" и до расширения ".html". Чтобы вычленить из ссылки ключ воспользуемся регулярными выражениями и библиотекой re (она уже идет вместе с Python и не требует установки, достаточно ее испортировать - import re):
Создадим новую процедуру parselink(link) в файле parse.py, которая будет вырезать из полученной строки (ссылки) нужный нам ключ:
def parselink(link): # парсим ИД из ссылки try: return re.search(r'/news/([^/]+)\.htm$', link).group(1) except: return None
Абракадабра
Понимаю, что конструкция "r'/news/([^/]+)\.htm$'" выглядит жутко для неподготовленного пользователя.
Если попробовать объяснить просто, то логика в следующем:
"/news/" и "\.htm$" означают начало и конец нашего "вхождения". Символ "\" служит для экранирования точки, т.к. она является спецсимволом, а символ "$" говорит о конце строки;
([^/]+) - внутри группирующих скобок "()" то, что мы должны проверить. "[^/]+" - любой символ кроме "/", а "+" обозначает одно или более повторений.
Маленький лайфхак - очень удобно формировать шаблоны к re при помощи ChatGPT:)) Достаточно сформулировать запрос в стиле "Как при помощи регулярных выражений в Python из строки "а" получить подстроку "б"".
Часть ссылок в ��енте RSS на Мотор может вести не только на новостную ленту (с префиксом "/news/"), но и на другие страницы сайта, где другая разметка и структура (например - 'https://motor.ru/selector/nemolodo-zeleno-elektricheskie-restomody.htm'). Поскольку нам нужно будет парсить текст страницы в будущем, сфокусируемся только на новостных страницах, для чего используем конструкцию try...except - при попытке обработать ссылку не содержащую "/news" re выдает ошибку, которую мы и отлавливаем.
Доработаем bot.py следующим образом:
import requests import feedparser from parse import parselink def main(): rss_link = 'https://motor.ru/exports/rss' rss_text=requests.get(rss_link).text #Загружаем RSS-ленту rss = feedparser.parse(rss_text)#Парсим RSS ленту for news in rss.entries[::-1]: print(parselink(news['link']))
При запуске получаем либо ключ новости, либо None - если ссылка ведет на иную, чем новостная, часть сайте.
Все подготовительные процедуры, предшествующие созданию базы данных выполнены. Приступим.
Для работы с БД будем использовать библиотеку aiosqlite ($pip install aiosqlite). Принципы работы с SQL в Python можно почитать тут, а работу с aiosqlite можно посмотреть во второй части материала от @vlakirна Хабр. Я сразу приведу код с комментариями в спойлере, который необходимо поместить в файл sql.py в корне проекта:
sql.py
import aiosqlite #Создаем БД, если ее нет в каталоге async def create_table(): async with aiosqlite.connect('storage.db') as db: await db.execute('CREATE TABLE IF NOT EXISTS motor ' '(id integer, news_id text, title text, link text, status text, date text, PRIMARY KEY(id AUTOINCREMENT))') await db.commit() #Создаем запись с новостью, заголовком и ссылкой async def save_to_db(news_id, title, link): async with aiosqlite.connect('storage.db') as db: await db.execute('INSERT INTO motor (news_id, title, link) VALUES (?, ?, ?)', (news_id, title, link)) await db.commit() #Обновляем запись в БД по ИД async def update_db(news_id, status, date): async with aiosqlite.connect('storage.db') as db: await db.execute('UPDATE motor SET status = ?, date = ? WHERE news_id = ?', (status, date, news_id)) await db.commit() #Запрашиваем данные async def select_for_db(news_id, column): async with aiosqlite.connect('storage.db') as db: cursor = await db.cursor() await cursor.execute(f'SELECT {column} FROM motor WHERE news_id = ?',(news_id,)) return await cursor.fetchone()
Для нашего проекта нам достаточно создать 4 процедуры:
async def create_table()- для создания БД в корневом каталоге. Причем благодаря "IF NOT EXISTS" мы проверяем, есть ли файл с БД, и если его нет - создаем базу и таблицу motor с полями id, news_id, title, link, status, date;async def save_to_db(news_id, title, link)- для создания записи по полям news_id, title, link;async def update_db(news_id, status, date)- для обновления записи - добавления данных в поля status, date по полю news_id;async def select_for_db(news_id, column)- для создания SELECT'ов к базе.
Поскольку библиотека aiosqlite асинхронная, то весь модуль работы с базой данных у нас уже написан в async. Допилим файл bot.py следующим образом:
from httpx import AsyncClient import feedparser import asyncio #Тут подгружаем процедуры из наших модулей from parse import parselink from sql import create_table, save_to_db, select_for_db async def main(): rss_link = 'https://motor.ru/exports/rss' #создаем БД, есои ее нет await create_table() #создаем экземпляр класса AsyncClient() и посылаем асинхронный get httpx_client = AsyncClient() rss_text = await httpx_client.get(rss_link) #Парсим RSS ленту rss = feedparser.parse(rss_text.text) for news in rss.entries[::-1]: news_id = parselink(news['link']) #Если ссылка именно на новость, а не на другой раздел сайта if news_id is not None: #Если записи нет в базе данных: if await select_for_db(news_id, 'news_id') is None: #Добавляем запись в базу данных: await save_to_db(news_id, news['title'], news['link']) print (f"Новость ID {news_id} добавлена в БД") # - спим 4 секунды, можно больше после каждой итерации с новой записью. await asyncio.sleep(4) else: #Если запись уже есть - пропускаем continue if __name__ == '__main__': asyncio.run(main())
Переделываем все под асинхронную работу, вместо библиотеки requests используем класс AsyncClient из httpx ($ pip install httpx). Внутри цикла добавляем два ветвления:
первое проверяет, что ссылка ведет именно на новость (результат работу процедуры parselink);
второе ветвление проверяет через
select_for_db, что запись ранее в базу не попадала. Если запись новая - она добавляется в базу черезsave_to_db. Далее здесь будет код отправки сообщения администратору ТГ-бота. В случае, если запись уже есть в базе (старая новость) - переходим к следующему шагу цикла через операторcontinue.
Начало ботостроения
Вот теперь самое интересное, ради чего все это делалось - разработка бота, подключение aiogram и вот это вот все.
Последние приготовления:
Установить aiogram в окружение venv:
pip install aiogramОпределить номер id администратора/-ов канала. Это ваш ID в Телеграм. Получить его можно при помощи бота getmyid_bot, послав ему команду /start. Ваш ID будет "Your user ID";
Определить Id-канала. Тут чуть сложнее. Необходимо зайти в ваш Телеграм через web-версию мессенджера. Нажать на ваш канал (не бот!) и в адресной строке браузера скопировать номер после # с "-". Например, если в адресной строке браузера вы увидите что-то подобное: "....telegram.org/a/#-1001984511001", то id-канала будет "-1001984511001".
Создаем в корне проекта файл variables.py и добавляем туда наши константы:
BOT_TOKEN = 'Токен вашего ТГ канала, который вы получили от BotFather' CHAT_ID = 'Id администратора бота из пункта 2 списка' CHANNEL_ID = 'id канала из пункта 3'
Для продвинутых любителей сисурности
Хранить переменные с чувствительными данными в коде не самая лучшая идея. Есть очень хорошее решение с использованием библиотеки pydantic и хранением данных в .env файле. Прочитать подробный гайд можно в том же мануале по Aiogram от Groosha - параграф "Файлы конфигурации".
В очередной раз грубо доработаем рашпилем файл bot.py:
bot.py
from httpx import AsyncClient import feedparser import asyncio from aiogram import Bot, Dispatcher #Тут подгружаем процедуры из наших модулей from parse import parselink from sql import create_table, save_to_db, select_for_db #Подгружаем переменные from variables import BOT_TOKEN, CHAT_ID #Подключаем бота bot = Bot(token=BOT_TOKEN) async def feed_reader(): rss_link = 'https://motor.ru/exports/rss' #создаем экземпляр класса AsyncClient() и посылаем асинхронный get httpx_client = AsyncClient() rss_text = await httpx_client.get(rss_link) #Парсим RSS ленту rss = feedparser.parse(rss_text.text) for news in rss.entries[::-1]: news_id = parselink(news['link']) #Если ссылка именно на новость, а не на другой раздел сайта if news_id is not None: #Если записи нет в базе данных: if await select_for_db(news_id, 'news_id') is None: #Сохраняем запись в БД: await save_to_db(news_id, news['title'], news['link']) print (f"Новость ID {news_id} добавлена в БД") # - спим 4 секунды, можно больше после каждой итерации с новой записью. await asyncio.sleep(4) else: #Если запись уже есть - пропускаем continue async def main(): #Не забываем про диспетчера и удаляем webhook на всякий случай dp = Dispatcher() await bot.delete_webhook(drop_pending_updates=True) #создаем БД, есои ее нет await create_table() #вызываем чтение ленты await feed_reader() #пуллим бот - чтобы наш скрипт постоянно запрашивал сервер ТГ о состоянии нашего бота await dp.start_polling(bot) if __name__ == '__main__': asyncio.run(main())
Переименовали процедуру main() в feed_reader(), перенесли процедуру создания БД в новую main(), скормили BOT_TOKEN aiogram.
Напишем процедуру отправки сообщений в бота в файл bot.py. Пока без inline-кнопок, их добавим на следующем шаге:
async def send_message_to_bot(text): await bot.send_message(chat_id=CHAT_ID, text=text, parse_mode='HTML')
А вместо print (f"Новость ID {news_id} добавлена в БД") в процедуре feed_reader() сделаем отправку сообщений в бота в виде заголовка статьи и ссылки на нее:
async def feed_reader(): rss_link = 'https://motor.ru/exports/rss' #создаем экземпляр класса AsyncClient() и посылаем асинхронный get httpx_client = AsyncClient() rss_text = await httpx_client.get(rss_link) #Парсим RSS ленту rss = feedparser.parse(rss_text.text) for news in rss.entries[::-1]: news_id = parselink(news['link']) #Если ссылка именно на новость, а не на другой раздел сайта if news_id is not None: #Если записи нет в базе данных: if await select_for_db(news_id, 'news_id') is None: #Сохраняем запись в БД: await save_to_db(news_id, news['title'], news['link']) #отправляем сообщение в бот - чуть магии с HTML для эстетики await send_message_to_bot(f'<b>{news["title"]}</b>\n{news["summary"]}\n<a href="{news["link"]}">Link</a>') # - спим 4 секунды, можно больше после каждой итерации с новой записью. await asyncio.sleep(4) else: #Если запись уже есть - пропускаем continue
Если все сделали правильно, то после запуска скрипта в бот придут первые сообщения. По сути мы уже сделали собственный простой feed-reader для Телеграм.
Теперь добавим этой штуке стероидов. Начнем с кнопок. У нас будет 2 группы inline-кнопок: первая группа (first_collection) - две кнопки для отправки в Chat-GPT и/или удаления сообщения, вторая группа (second_collection) - одна кнопка для отправки сообщения в новостной канал.
Следуя логике из мануала по aiogram от Groosha cоздадим отдельную папку "entrails" в корне проекта и в ней файл keyboards.py со следующим кодом:
#keyboards collection from aiogram.types import InlineKeyboardMarkup from aiogram.utils.keyboard import InlineKeyboardBuilder def first_collection() -> InlineKeyboardMarkup: #Отправка в ChatGPT buttons = InlineKeyboardBuilder() buttons.button(text='Send to Chat-GPT', callback_data='gpt_button_pressed') buttons.button(text='Delete', callback_data='gpt_button_delete') buttons.adjust(2) return buttons.as_markup() def second_collection() -> InlineKeyboardMarkup: #Отправка в канал buttons = InlineKeyboardBuilder() buttons.button(text='Send to Channel', callback_data='channel_button_pressed') buttons.adjust(1) return buttons.as_markup()
После этого добавим в процедуру send_message_to_bot в файле bot.py нашу первую коллекцию кнопок:
#подключаем первую коллекцию from entrails.keyboard import first_collection async def send_message_to_bot(text): await bot.send_message(chat_id=CHAT_ID, text=text, parse_mode='HTML', reply_markup=first_collection())
Маленький совет
Чтобы было удобно отлаживать наш скрипт и каждый раз не удалять базу данных и не смотреть на 80+ сообщений в канале, советую поставить какой-нибудь редактор для SQLite баз и перед каждым запускам удалять пару строчек из базы для "отладочного" прогона. Например, я использую DB Browser for SQLite - бесплатный, есть почти на все платформы.
Ура, у нас уже есть модные inline-кнопки. Все как у взрослых:)) Одна проблема, нажимать на них бесполезно - скрипт пока не умеет с ними работать.
Попробуем исправить эту оплошность. Нам нужно написать обработчики событий, который будет отлавливать нажатия на кнопки и совершать нужные действия.
Задача сводится к тому, чтобы отслеживать callback_data (см. код кнопок выше) от кнопки. Создадим в папке entrails файл handlers.py.
Начнем с обработки события по нажатию кнопки Delete. Чтобы остаться в разумных пределах статьи Хабра и не допиливать этот обработчик несколько раз, сразу добавим функцию, которая позволит по Id-статьи добавлять в уже имеющуюся в базе-данных запись информацию об удалении новости и дате и времени этого действия. Мы создавали поля status и date в таблице motor и процедуру update_db - настало их время. В поле статус будем ставить статус "Del" или "Can't del - more 48H", в случае если мы захотим удалить сообщение по истечение 48 часов с момента отправки (Это особенность работы Telegram - сообщения можно удалить только в течение 48 часов после отправки). В поле date - добавляем текущую дату и время в системе при помощи библиотеки datetime.
from aiogram import Router, F from aiogram.types import CallbackQuery import datetime #импортируем переменные from variables import CHANNEL_ID, CHAT_ID #импортируем процедуру работы с БД from sql import update_db #импортируем парсер ссылки на новость from parse import parselink #подключаем роутер router = Router() #если новость не нужна - удаляем @router.callback_query(F.data=='gpt_button_delete') async def delete_message(callback: CallbackQuery): try: await callback._bot.delete_message(chat_id=CHAT_ID, message_id=callback.message.message_id) for item in callback.message.entities: if item.type == 'text_link': post = item.url id = parselink(post) await update_db(id,'Del', datetime.datetime.now()) except: for item in callback.message.entities: if item.type == 'text_link': post = item.url id = parselink(post) await update_db(id,"Can't del - more 48H", datetime.datetime.now())
Обратите внимание на то, как мы вытаскиваем из сообщения ссылку из которой уже при помощи знакомой процедуры parselink() тянем news_id. У объекта message есть свои entities по коллекции которых мы проходим циклом до появления объекта с типом text_link.
Если же смотреть на принцип работы обработки нажатия кнопки, то через декоратор роутера мы отслеживаем коллбэки, и если появляется событие "gpt_button_delete" запускаем процедуру delete_message(). Внимательным читателям гайда по Aiogram должно быть все понятно.
Теперь у нас полностью рабочая кнопка Delete - сообщение удаляется, а в базу данных заносится статус удаления и время этого удаления.
Куда же без AI?
Если вы еще не забыли, то в первой коллекции кнопок у нас есть кнопка - "Send to Chat-GPT".
Для нашего примера будем использовать ChatGPT и их python-библиотеку openai, которую необходимо установить в окру��ение проекта стандартным способом через утилиту pip.
Понимаю, что в России сейчас есть определенные трудности с доступом, регистрацией и оплатой сервисов OpenAI. В сети достаточно примеров того, как решить эту проблему. Будем считать, что вы удачно зарегестрировались, внесли оплату и получили токен.
В примере я буду использовать модель "gpt-4" - на мой взгляд, результат ее работы лучше, чем у "gpt-3.5 turbo". Но стоит каждый такой запрос дороже и работает она существенно медленнее.
Формирование вопроса (Prompt) к Chat-GPT отдельная и сложная тема, можно сказать - целое искусство. Советую почитать этот материал. В любом случае, вам советую запастись терпением и бюджетом на эксперименты - тут придется экспериментировать какое-то время. Допустим, наш Prompt будет выглядеть следующим образом (позже вставим в код процедуры):
Как администратор новостного канала напиши короткое изложение новости (не больше 100 слов)
Создадим в корне проекта файл gptai.py и добавим в него код:
#библиотека для работы с ChatGPT import openai #токен добавляем в variables.py, а сюда подгружаем from variables import GPT_TOKEN #процедура, которая получает на вход чистый текст новости и возвращает рерайт-версию def get_GPT(text): openai.api_key = GPT_TOKEN response = openai.ChatCompletion.create( model="gpt-4", temperature=0, messages=[ { "role": "system", "content": "Как администратор новостного канала напиши короткое изложение новости (не больше 100 слов)" }, { "role": "user", "content": text } ]) return response['choices'][0]['message']['content']
Все достаточно просто - процедура get_GPT() получает текст статьи и дальше скармливает его openai - направляя запрос на две роли: пользователя с текстом статьи и "системы" с нашим prompt'ом. Процедура в итоге возвращает ответ ChatGPT.
Возвращаемся в мир aiogram
Осталось написать новый хендлер, отлавливающий нажатие кнопки Send to Chat-GPT и отправляющий текст в get_GPT(). В файл handlers.py дописываем:
handlers.py
from aiogram import Router, F from aiogram.types import CallbackQuery import datetime #импортируем переменные from variables import CHANNEL_ID, CHAT_ID #импортируем процедуру работы с БД from sql import update_db, select_for_db #импортируем парсер новости и парсер ссылки from parse import parse, parselink #импортируем вторую коллекцию кнопок from entrails.keyboard import second_collection #импортируем процедуру работы с ChatGPT from gptai import get_GPT #подключаем роутер router = Router() #Отправляем новость на съедение ChatGPT @router.callback_query(F.data=='gpt_button_pressed') async def get_link(callback: CallbackQuery): for item in callback.message.entities: if item.type == 'text_link': post = item.url #url на новость id = parselink(post) #получаем из url id-новости text = parse(post) #чистый текст новости title = await select_for_db(id, 'title') #берем заголовок из базы brief = get_GPT(text) #получает текст от ChatGpt #формируем текст сообщения format_text = f'<b>{title[0]}</b>\n{brief}\n<a href="{post}">Link</a>' #добавляем в базу статус и время отправки сообщения await update_db(id,'Send', datetime.datetime.now()) #отправляем сообщение await callback.message.answer(format_text, parse_mode='HTML', disable_web_page_preview=True, reply_markup=second_collection()) #пробуем удалить исходное сообщение try: await callback._bot.delete_message(chat_id=CHAT_ID, message_id=callback.message.message_id) except: print(f"{callback.message.message_id} can't be deleted")
Я снабдил код под спойлером подробными комментариями, по этому не буду повторяться. Сообщение формируется из ранее занесенных в базу title, link и ответа Chat-GPT. Внимательный читатель заметит, что в callback.message.answer мы добавили disable_web_page_preview=Tre чтобы отключить предпросмотр страницы по ссылке, а также вторую коллекцию с одной единственной кнопкой - Send to Channel.
Важное замечание по работе
Обратите внимание, что модель gpt-4 работает очень медленно. После нажатия на кнопку Send to Chat-GPT придется подождать секунд 30-40 прежде, чем вам придет обратное сообщение в бот. Крайне не советую жать на кнопки сразу в нескольких сообщениях, не дожидаясь ответа - это может вызвать сбой.
Напишем для кнопки Send to Channel еще один маленький хендлер, который будет отправлять копию сообщения (уже без кнопок) в новостной канал:
@router.callback_query(F.data=='channel_button_pressed') async def send_tochat(callback: CallbackQuery): await callback._bot.copy_message(chat_id=CHANNEL_ID, from_chat_id=CHAT_ID, message_id=callback.message.message_id) try: await callback._bot.delete_message(chat_id=CHAT_ID, message_id=callback.message.message_id) except: print(f"{callback.message.message_id} can't be deleted")
В итоге сообщение в новостном канале должно выглядеть примерно так:
Считаю, что для второй части нашей трилогии материала достаточно.
Мы разобрали следующие темы:
Коснулись темы регулярных выражений
Научились основам работы с базами-данных
Написали 95% бота
Подключились к ChatGPT
В заключительной, третьей части, мы научимся запускать процедуру feed_reader() по расписанию и напишем еще один хандлер в handlers.py, который будет вызывать feed_reader() по команде. Кроме этого, соберем Docker Image бота и попробуем его запустить на удаленном сервере.
