Не так давно в Telegram вышло большое обновление – «Telegram для бизнеса». В данный момент оно доступно для Premium-пользователей, а в будущем, вероятно, станет отдельным режимом.
«Telegram для бизнеса» предоставляет собой новый способ взаимодействия с клиентами через Telegram, вводя для этого новые функции:
- Адрес – позволяет указать адрес и геопозицию в профиле.
- Часы работы – позволяет указать график работы бизнеса.
- Быстрые ответы – позволяет создать набор «шаблонных» ответов.
- Приветствия – позволяет установить автоматическое приветствие для новых клиентов.
- «Нет на месте» – позволяет отправлять автоматические ответы, в нерабочее время.
- Ссылки на чат – позволяет кастомизировать ссылки на чат с вами.
- Вид нового чата – позволяет кастомизировать вид чата для клиента, который открыл чат с вами, но ещё не написал сообщение.
- Чат-боты – позволяет подключить к учётной записи бота для взаимодействия с клиентами в личных чатах.
Из всего этого набора нас интересует только два пункта: Чат-боты и Часы работы.
Что мы с вами сделаем?
В этой статье мы создадим Telegram-бота, который будет принимать личные сообщения только в нерабочее время и для ответа использовать ChatGPT от OpenAI.
Поскольку OpenAI недоступен на территории РФ, вместо него будем использовать сервис NeuroAPI. Он предоставляет доступ к OpenAI из России и СНГ по более низким ценам.
Как это можно использовать?
Описанный в статье бот можно будет использовать как частному лицу, сделав личного ассистента на время отсутствия в сети, так и бизнесу для взаимодействия с клиентами в нерабочее время.
Главная сложность будет заключаться в составлении грамотного «системного промта», покрывающего ваши потребности.
Подключение бота в профиле
Для проекта вам нужен бот. Как его создать рассказано в статье AIOgram3 1.5. Регистрация бота
После создания бота и получения токена, в интерфейсе BotFather, выполните команду /mybots для вывода списка всех ботов.Выберите нужного бота.
Затем в открывшемся меню выберите пункт Bot Settings.
В следующем меню выберите пункт Business Mode.
Включите бизнес режим.
После того, как включили бизнес режим для бота, откройте настройки Telegram и выберите пункт Telegram для бизнеса, а в нём пункт Чат-боты.
В открывшемся окне в первое поле пропишите ссылку на бота t.me/mybot или его имя @mybot.
Готово.
Подготовка проекта
Создайте новый проект в удобной для вас IDE и активируйте виртуальное окружение.
Если вы пользуетесь PyCharm, то виртуальное окружение создаст IDE для нового проекта. Если вы пользуетесь VSCode, то его придётся создать вручную, выполнив следующие команды:
python -m venv .venv
# для Windows
venv\Scripts\activate.ps1 или venv\Scripts\activate.bat
# для *NIX-систем
source venv/bin/activate
В проекте используются следующие библиотеки:
aiogram– Фреймворк для бота.pydantic-settings– Библиотека для создания классов конфигураций.openai– Официальная библиотека OpenAI для Python.pytz– Библиотека для работы с часовыми поясами.httpx– Современная библиотека для создания синхронных/асинхронных запросов.redis– Библиотека для подключения к Redis.
Установите их, выполнив команду:
pip install -U aiogram pydantic-settings openai pytz httpx redis
Создайте файл requirements.txt и внесите в него установленные библиотеки:
aiogram==3.6.0
pydantic-settings==2.2.1
openai==1.29.0
pytz==2024.1
httpx==0.27.0
redis==5.0.4
Далее создайте файл .env для хранения переменных окружения.Необходимы следующие переменные:
token– Токен бота, полученный от BotFather.admin_id– Telegram-id администратора.openai_key– API-ключ полученный на сайте NeuroAPI или OpenAI.openai_base_url– Адрес прокси-сервера для OpenAI.redis_host– Хост для подключения к Redis. В нашем случае используется Docker compose, поэтому прописываем имя сервиса –redis.delay– Задержка между ответами в минутах. Об этом ниже.
Пример:
token=12345:abcd
admin_id=123456789
openai_key=sk-abcd
openai_base_url=https://lk.neuroapi.host/v1
redis_host=redis
delay=10
Также создайте файл main.py и пакет (Python package) app.
Файл конфигурации
В пакете app создайте файл settings.py. В нём будем получать данные из .env-файла и определим инстанс бота и Redis.
Создайте класс Secrets, унаследованный от BaseSettings. Этот класс будет получать из .env-файла данные и преобразовывать их в Python-объекты. Для этого используется библиотека pydantic-settings.
В теле класса пропишите шесть полей с указанием типа данных:
token: str
admin_id: int
openai_key: str
openai_base_url: str
redis_host: str
delay: int
После полей, внутри класса напишите внутренний класс Config, в котором укажите из какого файла брать данные и его кодировку:
class Config:
env_file = ".env"
env_file_encoding = "utf-8"
Под классом создадим переменную secrets и объявим её экземпляром класса Secrets.
Далее создайте переменную redis_conn, это будет экземпляр класса Redis, в который передаём адрес хоста. Будьте внимательны во время импорта класса! Нам нужен асинхронный Redis.
redis_conn = Redis(host=secrets.redis_host)
Последней будет переменная bot. Объявите её экземпляром класса Bot, передав в него токен и режим форматирования сообщений.
bot = Bot(token=secrets.token, parse_mode="Markdown")
Про parse_mode: Поскольку в ответе ChatGPT может находиться блок кода или другое форматирование, для корректного отображения его необходимо «распарсить». Передав параметр parse_mode="Markdown", мы сообщаем боту, что все сообщения будут с Markdown-форматированием.
Полный код файла
from aiogram import Bot
from pydantic_settings import BaseSettings
from redis.asyncio import Redis
class Secrets(BaseSettings):
token: str
admin_id: int
openai_key: str
openai_base_url: str
redis_host: str
delay: int
class Config:
env_file = ".env"
env_file_encoding = "utf-8"
secrets = Secrets()
redis_conn = Redis(host=secrets.redis_host)
bot = Bot(token=secrets.token, parse_mode="Markdown")
Хранилище строк
Для хранения текстовых строк в одном месте в пакете app создайте файл views.py.
Этого можно и не делать. Кроме того, вариант с функциями можно заменить на получение текста из файла или иной способ.
Создайте три простые функции, которые ничего не принимают и возвращают текстровую строку:
start_bot_message– Сообщение о запуске бота для администратора.stop_bot_message– Сообщение об остановке бота для администратора.system_prompt– Системный промт, описывающий поведение ChatGPT.
Код
def start_bot_message():
return "Бот запущен"
def stop_bot_message():
return "Бот остановлен"
def system_prompt():
return """Ты бот помощник и ты должен помогать людям."""
Проверка рабочего времени
В Telegram часы работы указываются по дням с понедельника по воскресенье. В коде же это выглядит как список объектов класса BusinessOpeningHoursInterval.
В объекте класса BusinessOpeningHoursInterval есть два поля: opening_minute и closing_minute, представленные в виде количества минут прошедших с 00:00 ближайшего понедельника, с учётом указанной временной зоны.
Необходимо получить текущее количество минут, прошедших с понедельника, и пройтись по списку, проверяя, входит ли текущее число в один из диапазонов. Если входит, то бот будет игнорировать сообщения. Если не входит, бот будет отвечать на сообщения.
В пакете app, создайте новый пакет utils. В этом пакете создайте файл opening_hours.py.
Создайте функцию check_opening_hours, принимающую opening_hours – объект класса BusinessOpeningHours.
Класс BusinessOpeningHours содержит два поля:
time_zone_name– Название временной зоны. Определяется в профиле Telegram при заполнении графика работы.opening_hours– Упомянутый выше список с объектами классаBusinessOpeningHoursInterval.
Далее создайте четыре переменные:
tz– В ней при помощи библиотекиpytzполучаем информацию об указанной временной зоне.now– В ней получаем текущее время с учётом временной зоны.monday_start– В ней высчитываем время до начала понедельника.minutes_since_monday– В ней высчитываем сколько прошло минут с начала недели.
tz = pytz.timezone(opening_hours.time_zone_name)
now = datetime.datetime.now(tz)
monday_start = now - datetime.timedelta(
days=now.weekday(),
hours=now.hour,
minutes=now.minute,
seconds=now.second,
microseconds=now.microsecond,
)
minutes_since_monday = (now - monday_start).total_seconds() / 60
Далее создайте цикл, в котором будем итерироваться по списку интервалов и проверять, входит ли текущее время в этот список.
for day in opening_hours.opening_hours:
if day.opening_minute <= minutes_since_monday <= day.closing_minute:
return False
return True
Полный код
import datetime
import pytz
from aiogram.types import BusinessOpeningHours
def check_opening_hours(opening_hours: BusinessOpeningHours):
tz = pytz.timezone(opening_hours.time_zone_name)
now = datetime.datetime.now(tz)
monday_start = now - datetime.timedelta(
days=now.weekday(),
hours=now.hour,
minutes=now.minute,
seconds=now.second,
microseconds=now.microsecond,
)
minutes_since_monday = (now - monday_start).total_seconds() / 60
for day in opening_hours.opening_hours:
if day.opening_minute <= minutes_since_monday <= day.closing_minute:
return False
return True
Проверка входящих сообщений
При получении входящего сообщения необходимо проверить актуальный режим работы и либо передать сообщение дальше в обработчик, либо «сбросить» его, тем самым никак не реагируя.
Для этого будем использовать миддлвари (middleware) – это так называемые «посредники», срабатывающие до передачи сообщения в обработчик и, в зависимости от логики, выполняющие различные действия, например, запись в БД, проверку аутентификации и многое другое.
В пакете app создайте пакет middlewares. В нём создайте файл business_middleware.py.
В этом файле создайте класс BusinessMiddleware, унаследованный от BaseMiddleware.
В нём нам нужно переопределить dunder-метод __call__, принимающий self, handler, event, data.
Далее нам необходимо получить из текущего чата объект класса BusinessOpeningHours.
Лирическое отступление
В актуальной на момент написания поста версии aiogram 3.6.0, заявлена полная поддержка Bot API 7.3. Если обратиться к объекту чата, то там будет параметр business_opening_hours, однако вместо желаемого объекта BusinessOpeningHours там находится None.
В этой статье мы применим небольшой «костыль» для решения этой проблемы.
Разработчикам aiogram был отправлен баг-репорт. Если в будущих версиях ситуация будет исправлена, пост будет обновлён.
Конец лирического отступления
Для получения актуального графика работы мы обратимся к API Telegram.
Используя асинхронный менеджер контекста и библиотеку httpx, откройте асинхронный клиент для работы.
В переменную response получаем результат GET-запроса на сервер Telegram.
В переменной chat получаем JSON-объект из переменной response.
Затем в переменной full_chat создаём экземпляр класса ChatFullInfo, распаковав в него содержимое chat по ключу result. Таким образом мы преобразуем чистые JSON-данные в Python-объекты.
async with httpx.AsyncClient() as client:
response = await client.get(
f"https://api.telegram.org/bot{secrets.token}/getChat?chat_id={secrets.admin_id}"
)
chat = response.json()
full_chat = ChatFullInfo(**chat["result"])
Далее в блоке if вызываем ранее написанную функцию check_opening_hours, передав в неё full_chat.business_opening_hours.
Если возвращается True, мы продолжаем.
Внутри условия создаём переменную context, в которую присваиваем значение ключа event_context из переменной data.
Дальше ещё одно условие if, в котором проверяем, что сообщение содержит business_connection_id, т. е. является личным и что отправитель сообщения не админ, иначе бот будет реагировать и на ваши сообщения тоже. Если условия соблюдаются, передаём сообщение дальше в обработчик.
if check_opening_hours(full_chat.business_opening_hours):
context: EventContext = data.get("event_context")
if (
context.user.id != secrets.admin_id
and context.business_connection_id
):
return await handler(event, data)
Полный код файла
from typing import Callable, Dict, Any, Awaitable
import httpx
from aiogram import BaseMiddleware
from aiogram.dispatcher.middlewares.user_context import EventContext
from aiogram.types import TelegramObject, ChatFullInfo
from app.settings import secrets
from app.utils.opening_hours import check_opening_hours
class BusinessMiddleware(BaseMiddleware):
async def __call__(
self,
handler: CallableTelegramObject, Dict[str, Any, Awaitable[Any]],
event: TelegramObject,
data: Dict[str, Any],
) -> Any:
async with httpx.AsyncClient() as client:
response = await client.get(
f"https://api.telegram.org/bot{secrets.token}/getChat?chat_id={secrets.admin_id}"
)
chat = response.json()
full_chat = ChatFullInfo(**chat["result"])
if check_opening_hours(full_chat.business_opening_hours):
context: EventContext = data.get("event_context")
if (
context.user.id != secrets.admin_id
and context.business_connection_id
):
return await handler(event, data)
Подключение ChatGPT
В этой функции будем отправлять запрос к ChatGPT и возвращать полученный ответ.
В пакете utils, создайте файл openai_actions.py.
Создайте асинхронную функцию get_chat_completion, принимающую message – объект класса Message.
В переменной http_client определите объект класса httpx.AsyncClient. Это объект HTTP-клиента, используя который будет произведён запрос.
В переменной client определите объект класса AsyncOpenAI, передав в него аргументы: api_key, http_client и base_url. Это объект клиента для OpenAI.
http_client = httpx.AsyncClient(
limits=httpx.Limits(max_connections=100, max_keepalive_connections=20)
)
client = AsyncOpenAI(
api_key=secrets.openai_key,
http_client=http_client,
base_url=secrets.openai_base_url,
)
Далее в переменной messages создайте список словарей, где первый словарь – это системный промт, а второй – сообщение от пользователя:
messages = [
{"role": "system", "content": system_prompt()},
{"role": "user", "content": message.text},
]
В переменную response создайте запрос, передав в него:
model– Выбранная модель ChatGPT, например,gpt-3.5-turbo,gpt-4-turbo,gpt-4oили любую другую поддерживаемую OpenAI.messages– Список словарей с сообщениями.max_tokens– Ограничение на максимальное количество токенов в ответе.temperature– Температура в диапазоне от 0 до 1. Определяет уровень «фантазии» бота. Чем ближе число к нулю, тем более предсказуемы будут ответы и наоборот, чем ближе к единице, тем более случайными будут ответы.
И возвращаем результат запроса в обработчик:
response = await client.chat.completions.create(
model="gpt-3.5-turbo", messages=messages, max_tokens=1000, temperature=0.8
)
return response.choices[0].message.content
Полный код
import httpx
from aiogram.types import Message
from openai import AsyncOpenAI
from app.settings import secrets
from app.views import system_prompt
async def get_chat_completion(message: Message):
http_client = httpx.AsyncClient(
limits=httpx.Limits(max_connections=100, max_keepalive_connections=20)
)
client = AsyncOpenAI(
api_key=secrets.openai_key,
http_client=http_client,
base_url=secrets.openai_base_url,
)
messages = [
{"role": "system", "content": system_prompt()},
{"role": "user", "content": message.text},
]
response = await client.chat.completions.create(
model="gpt-3.5-turbo", messages=messages, max_tokens=1000, temperature=0.8
)
return response.choices[0].message.content
Задержка обработки сообщений
Для того, чтобы пользователи не спамили и не использовали личные сообщения как «бесплатный GPT», добавим задержку в обработке сообщений.
В вашей реализации логики она может быть не нужна.
В пакете utils создайте файл check_delay.py, а в нём асинхронную функцию check_user_delay, принимающую user_id.
Тут-то нам и понадобится Redis для хранения пользовательских ID и времени последнего сообщения. Вы можете использовать для этого другую БД или вовсе словарь в коде, это не принципиально.
В переменную last_message_time получаем из Redis по user_id время последнего сообщения, если оно есть. Если его нет – вернётся None.
В блоке if проверяем, что last_message_time True (проще говоря, не None).Внутри блока в переменную time_since_last_message получаем разницу между текущим временем и полученным из хранилища.Ниже проверяем, если оно меньше указанной в .env допустимой задержки, то возвращаем False.
Во всех остальных случаях возвращаем True.
Полный код
import asyncio
from app.settings import redis_conn, secrets
async def check_user_delay(user_id: int):
last_message_time = await redis_conn.get(f"users:{user_id}")
if last_message_time:
time_since_last_message = asyncio.get_event_loop().time() - float(
last_message_time
)
if time_since_last_message < secrets.delay * 60:
return False
return True
Обработчик бизнес сообщений
Осталось написать обработчик, в который middleware будет передавать сообщение.
В пакете app создайте пакет handlers, а в нём файл business_handler.py.
В этом файле создайте асинхронную функцию handle_business_message, принимающую message – объект класса Message.
В самом начале создайте блок if, проверяющий задержку и наличие текста в сообщении (отправить могут картинку или видео, а это другая логика работы с ChatGPT).
Если условие не выполняется, то сообщение просто игнорируется.
Если условие выполнено, переходим к обработке.
В переменной answer вызываем функцию get_chat_completion, передав в неё message.
Затем отвечаем пользователю полученным сообщением.
Сохраняем в Redis время текущего сообщения.
Полный код
import asyncio
from aiogram.types import Message
from app.settings import redis_conn
from app.utils.check_delay import check_user_delay
from app.utils.openai_actions import get_chat_completion
async def handle_business_message(message: Message):
if await check_user_delay(message.from_user.id) and message.text:
answer = await get_chat_completion(message)
await message.reply(answer)
await redis_conn.set(
f"users:{message.from_user.id}", asyncio.get_event_loop().time()
)
Обработка уведомлений о запуске/остановке бота
Небольшое, но удобное дополнение.
В пакете handlers создайте файл events.py.
В нём создайте две асинхронные функции: start_bot и stop_bot.
В функциях отправляем сообщение администратору.
from app.settings import bot, secrets
from app import views
async def start_bot():
await bot.send_message(secrets.admin_id, views.start_bot_message())
async def stop_bot():
await bot.send_message(secrets.admin_id, views.stop_bot_message())
Основной файл
Логику написали. Теперь осталось соединить всё вместе.
Откройте созданный ранее файл main.py. Он должен находиться в корне проекта рядом с файлом .env.
В нём создайте асинхронную функцию start.
В переменной dp объявите экземпляр класса Dispatcher.
Далее в несколько строк зарегистрируйте middleware и обработчики:
dp = Dispatcher()
dp.update.middleware(BusinessMiddleware())
dp.startup.register(start_bot)
dp.shutdown.register(stop_bot)
dp.business_message.register(handle_business_message)
Обратите внимание на dp.business_message.register. Регистрируется обработка business_message, а не обычного message.
Далее в блоке try вызывается очистка сообщений, отправленных, когда бот был офлайн, и запуск пуллинга, а в блоке finally выполняется остановка бота.
Вне функции в блоке if __name__ "__main__" запускаем функцию старт.
Полный код
import asyncio
from aiogram import Dispatcher
from aiogram.methods import DeleteWebhook
from app.handlers.business_handler import handle_business_message
from app.handlers.events import start_bot, stop_bot
from app.middlewares.business_middleware import BusinessMiddleware
from app.settings import bot
async def start():
dp = Dispatcher()
dp.update.middleware(BusinessMiddleware())
dp.startup.register(start_bot)
dp.shutdown.register(stop_bot)
dp.business_message.register(handle_business_message)
try:
await bot(DeleteWebhook(drop_pending_updates=True))
await dp.start_polling(bot)
finally:
await bot.session.close()
if __name__ "__main__":
asyncio.run(start())
Запуск бота
Для запуска бота и Redis будем использовать Docker compose.
Сперва необходимо создать образ с ботом, для этого создайте файл Dockerfile со следующим содержимым:
FROM python:3.11-slim
WORKDIR /code
COPY requirements.txt /code
RUN pip install --upgrade pip && pip install -r requirements.txt
COPY . /code
CMD [ "python", "./main.py" ]
В нём создаётся Docker-образ, в котором устанавливаются все зависимости из файла requirements.txt. Затем копируются файлы проекта и выполняется команда запуска бота.
Затем создайте файл docker-compose.yaml со следующим содержимым:
services:
bot:
build: .
restart: always
env_file:
- .env
volumes:
- .:/code
redis:
image: redis
restart: always
volumes:
- ./redis_data:/data
В нём описываются два сервиса:
Первый bot. Указываем, что необходимо создать образ из Dockerfile, передать в него .env-файл и подключить текущую папку внутри контейнера.
Второй redis. Указываем, что будет использоваться официальный образ redis последней версии, и подключаем папку redis_data внутри контейнера, чтобы не потерять данные.
Готово.
Запустить бота можно командой:
docker compose up -d
Пост написан для Telegram-канала Код на салфетке. У нас также есть сайт.
Комментарии