Хотел отложить написание второй части трилогии в долгий ящик, но судя по просмотрам первого эпизода - тема создания Телеграм-ботов все еще актуальна на Хабр.

Во второй части сфокусируемся на разработке бизнес-логики бота. В нашем проекте, для взаимодействия с Telegram, будем использовать библиотеку Aiogram. Для Python написано достаточное количество библиотек для работы с ТГ, но Aiogram, наверное, самая популярная. Советую прочитать руководство по работе с Aiogram от Groosha - для меня это была основная теоретическая база. Кроме непосредственной работы с функционалом библиотеки, советую обратить внимание на раздел "Роутеры. Структура" - я буду следовать этой логике при создании бота.

Мы будем работать в асинхронном режиме. Это достаточно сложная тема для неподготовленного читателя и не могу сказать, что я освоил все аспекты работы Python в асинхронном режиме. Для поверхностного погружения рекомендую почитать серию статей от @vlakirна Хабр.

Не буду описывать процесс регистрации бота в ТГ - это достаточно простая процедура, чтобы тратить на нее время тут. В интернете очень много материала на эту тему. Будем считать, что у вас на руках уже есть Token вашего бота от BotFather. Также, в ТГ необходимо создать публичный канал и добавить бота в администраторы этого канала. Вопрос открытия комментариев к постам - на ваше усмотрение.

Бот должен стать промежуточным этапом между публикацией новости в канал. По расписанию или запросу скрипт должен пробегать по RSS-ленте сайта и проверять ее на наличие новых, ранее не опубликованных новостей. В случае, если появилось что-то новое, направлять сообщение, состоящее из заголовка, краткого содержания и ссылки на новость в ТГ-канал администратору с двумя inline-кнопками - "Удалить" и "Отправить в ChatGPT". В случае нажатия на кнопку "ChatGPT" скрипт направляет текст новости (после работы процедуры parse, созданной в Первой части материала) на сервер ChatGPT с заданным Prompt и удаляет исходную новость из ТГ-канала. Ответ от "ChatGPT" возвращается новым сообщением в бот с единственной кнопкой - "Отправить в канал", после нажатия на которую сообщение улетает в канал и удаляется из бота.

Сохраняемся

Для реализации подобного функционала понадобится, как минимум, где-то хранить информацию о ранее опубликованных новостях. Можно пойти по пути@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

Часть ссылок в ленте 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 в корне проекта:

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 процедуры:

1. async def create_table() - для создания БД в корневом каталоге. Причем благодаря "IF NOT EXISTS" мы проверяем, есть ли файл с БД, и если его нет - создаем базу и таблицу motor с полями id, news_id, title, link, status, date;

2. async def save_to_db(news_id, title, link) - для создания записи по полям news_id, title, link;

3. async def update_db(news_id, status, date) - для обновления записи - добавления данных в поля status, date по полю news_id;

4. 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 и вот это вот все.

Последние приготовления:

1. Установить aiogram в окружение venv: pip install aiogram

2. Определить номер id администратора/-ов канала. Это ваш ID в Телеграм. Получить его можно при помощи бота getmyid_bot, послав ему команду /start. Ваш ID будет "Your user ID";

3. Определить Id-канала. Тут чуть сложнее. Необходимо зайти в ваш Телеграм через web-версию мессенджера. Нажать на ваш канал (не бот!) и в адресной строке браузера скопировать номер после # с "-". Например, если в адресной строке браузера вы увидите что-то подобное: "....telegram.org/a/#-1001984511001", то id-канала будет "-1001984511001".

Создаем в корне проекта файл variables.py и добавляем туда наши константы:

BOT_TOKEN = 'Токен вашего ТГ канала, который вы получили от BotFather'
CHAT_ID = 'Id администратора бота из пункта 2 списка'
CHANNEL_ID = 'id канала из пункта 3'

В очередной раз грубо доработаем рашпилем файл 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())

Ура, у нас уже есть модные 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 дописываем:

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.

Напишем для кнопки 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")

В итоге сообщение в новостном канале должно выглядеть примерно так:

Считаю, что для второй части нашей трилогии материала достаточно.

Мы разобрали следующие темы:

1. Коснулись темы регулярных выражений

2. Научились основам работы с базами-данных

3. Написали 95% бота

4. Подключились к ChatGPT

В заключительной, третьей части, мы научимся запускать процедуру feed_reader() по расписанию и напишем еще один хандлер в handlers.py, который будет вызывать feed_reader() по команде. Кроме этого, соберем Docker Image бота и попробуем его запустить на удаленном сервере.