Пример расширения Чат на Python

Рассмотрим создание простого расширения Pyrus, реализующего интеграцию с Telegram через бота. Расширение позволит создавать задачи в Pyrus с внешним каналом для переписки с пользователем Telegram. Получаемые Telegram-ботом сообщения от разных пользователей будут приходить в отдельные задачи в Pyrus.

Для запуска примеров кода потребуется Python версии не ниже 3.8.

Подготовка окружения

Создадим папку проекта и виртуальное окружение для Python. Для этого выполните в консоли следующие команды:

• для Windows (PowerShell)

  mkdir pyrus-telegram-extension
  cd pyrus-telegram-extension
  python -m venv venv
  .\venv\scripts\activate
  

• для Linux/macOS

  mkdir pyrus-telegram-extension
  cd pyrus-telegram-extension
  python -m venv venv
  source .venv/bin/activate
  

Разработка

Будем использовать следующие библиотеки:

• FastAPI — для обработки вебхуков от Pyrus и Telegram.

• Pydantic — для описания запросов от Pyrus и их десериализации.

• Aiogram — для работы с чат-ботами Telegram.

• Uvicorn — реализация асинхронного веб-сервера для работы с FastAPI.

Для установки этих зависимостей создайте в текущей папке файл requirements.txt со следующим содержимым:

fastapi==0.110.1
pydantic==2.5.3
aiogram==3.4.1
uvicorn==0.29.0

Теперь установите зависимости командой:

pip install -r requirements.txt

Создайте файл main.py и добавьте следующие импорты:

from typing import List
from aiohttp import ClientSession
from fastapi import FastAPI, Request, status
from pydantic import BaseModel
from aiogram import Bot, Dispatcher
from aiogram.types import Update, Message

Создание расширения в Pyrus

Теперь создадим новое расширение в Pyrus, чтобы получить авторизационные параметры для доступа к Extensions API. Перейдите в Каталог расширений Pyrus и нажмите на вкладку Разработка расширений, затем Создать новое расширение.

Введите название расширения и нажмите Создать.

Затем перейдите на вкладку Общее созданного расширения.

Скопируйте Логин и Секретный ключ в файл main.py в переменные CLIENT_ID и SECRET_KEY соответственно:

CLIENT_ID = "<Логин>"
SECRET_KEY = "<Секретный ключ>"

Обработка вебхуков Telegram

Добавьте функцию получения токена (через запрос token) для авторизации запросов к Extensions API:

async def fetch_token():
    async with ClientSession() as session:
        async with session.post(
                "https://extensions.pyrus.com/v1/token",
                json={
                    "client_id": CLIENT_ID,
                    "secret": SECRET_KEY,
                },
                headers={
                    "Content-Type": "application/json"
                }
        ) as response:
            return (await response.json()).get("access_token")

Затем создайте обработчик, который будет вызываться при получении нового сообщения ботом Telegram. При получении нового сообщения будем получать новый токен доступа и отправлять запрос getmessage в Extensions API для регистрации входящего сообщения (в виде новой задачи либо комментария в существующую задачу):

telegram = Dispatcher()

@telegram.message()
async def message_with_text(message: Message):
    access_token = await fetch_token()
    async with ClientSession() as session:
        await session.post(
            "https://extensions.pyrus.com/v1/getmessage",
            json={
                "account_id": str(message.bot.id),
                "channel_id": str(message.chat.id),
                "sender_name": message.from_user.full_name,
                "message_text": message.text,
            },
            headers={
                "Content-Type": "application/json",
                "Authorization": f"Bearer {access_token}"
            }
        )

Так же добавим обработчик вебхуков от Telegram. Для этого создадим инстанс FastAPI, который нам ещё понадобится позднее.

app = FastAPI()

@app.post("/webhook/bot/{bot_token}")
async def bot_webhook(bot_token: str, update: dict):
    bot = Bot(token=bot_token)
    telegram_update = Update(**update)
    await telegram.feed_update(bot=bot, update=telegram_update)
    await bot.session.close()

Обработка вебхуков Pyrus

Для обработки запросов от Pyrus потребуется реализовать следующие эндпоинты:

• GET /pulse
• POST /authorize
• POST /toggle
• POST /sendmessage

Для запросов authorize, toggle и sendmessage создадим модели данных, которые позволят автоматически десериализовывать запросы от Pyrus, а также добавим обработчики для каждого запроса:

class Credential(BaseModel):
    code: str
    value: str

class ToggleRequest(BaseModel):
    credentials: List[Credential]
    account_id: str
    enabled: bool

class AuthorizeRequest(BaseModel):
    credentials: List[Credential]

class SendMessageRequest(BaseModel):
    credentials: List[Credential]
    account_id: str
    channel_id: str
    message_text: str

@app.post("/authorize", status_code=status.HTTP_200_OK)
async def authorize(request: AuthorizeRequest):
    token = next(c.value for c in request.credentials 
                 if c.code == "bot_token")
    bot = Bot(token)
    try:
        response = await bot.get_me()
        return {
            "account_name": response.full_name, 
            "account_id": response.id
        }
    finally:
        await bot.session.close()

@app.post("/sendmessage", status_code=status.HTTP_200_OK)
async def send_message(request: SendMessageRequest):
    token = next(c.value for c in request.credentials 
                 if c.code == "bot_token")
    bot = Bot(token)
    try:
        await bot.send_message(
            chat_id=request.channel_id, 
            text=request.message_text
        )
    finally:
        await bot.session.close()

@app.post("/toggle", status_code=status.HTTP_200_OK)
async def toggle(request: ToggleRequest, raw_request: Request):
    token = next(c.value for c in request.credentials 
                 if c.code == "bot_token")
    bot = Bot(token=token)
    hostname = raw_request.url.hostname
    bot_webhook = f"https://{hostname}/webhook/bot/{token}"
    try:
        if request.enabled:
            await bot.set_webhook(url=bot_webhook)
        else:
            await bot.delete_webhook(drop_pending_updates=True)
    finally:
        await bot.session.close()

@app.get("/pulse", status_code=status.HTTP_200_OK)
async def pulse(): return {}

Тестирование

Запустите сервер с приложением локально, выполнив команду:

uvicorn main:app

Сервер запустится на порту 8000 и будет доступен по адресу http://127.0.0.1:8000:

INFO:     Started server process [14876]
INFO:     Waiting for application startup.
INFO:     Application startup complete.
INFO:     Uvicorn running on http://127.0.0.1:8000 (Press CTRL+C to quit)

Проверьте, что сервер вернёт статус код 200 на запрос pulse, выполнив команду:

curl http://127.0.0.1:8000/pulse

Получение временного публичного адреса

Для того чтобы к приложению могли идти запросы из Pyrus или Telegram, нужно предоставить ему публичный адрес с протоколом https. При первичном тестировании можно запустить приложение локально и создать туннель с временным адресом с помощью сервиса serveo.net:

ssh -R 80:127.0.0.1:8000 serveo.net

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

Forwarding HTTP traffic from https://<random-unique-name>.serveo.net

Конфигурация расширения Pyrus в личном кабинете

Перейдите в настройки созданного расширения в Pyrus.

В качестве адреса веб-сервиса укажите публичный адрес, полученный на предыдущем шаге:

Выберите авторизацию по параметрам и добавьте параметр bot_token:

На следующем шаге включите дополнительную функцию Онлайн-чат:

На шаге Информация о расширении введите локализованное название для параметра bot_token (например, Токен бота), а также заполните остальные обязательные поля.

После этого нажмите Продолжить для сохранения настроек — теперь расширение можно установить в форму.

Создание бота в Telegram

Для создания нового бота в Telegram перейдите в бота @BotFather и выполните команду /newbot. Далее, следуя инструкциям BotFather, введите название бота и сохраните его токен доступа — он понадобится при подключении расширения в форму.

Установка расширения в форму

Нажмите кнопку Установить под иконкой расширения и выберите из списка любую доступную вам форму.

После этого вы попадете на страницу подключения расширения. Заполните поле Токен бота и нажмите Подключить аккаунт.

Активируйте расширение переведя флаг в поле Статус в активное положение.

Отправка и получение сообщений

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

Отправленные из Pyrus во внешний канал комментарии появятся в приложении Telegram:

Деплой сервиса расширений

Развернём сервис расширения в облаке Yandex.Cloud. Будем использовать виртуальную машину с Ubuntu и запуск приложения в docker-контейнере. Для хранения docker-образов воспользуемся Yandex Container Registry. Также потребуется доменное имя и ssl-сертификат для него.

Подготовка сервера

Создайте виртуальную машину с Ubuntu в yandex cloud.

После выбора параметров виртуальной машины нажмите Создать ВМ.

После создания зайдите на сервер через ssh и установите docker и docker-compose:

curl -fsSL https://get.docker.com -o get-docker.sh
sh get-docker.sh
curl -L "https://github.com/docker/compose/releases/download/1.28.0/docker-compose-$(uname -s)-$(uname -m)" -o /usr/local/bin/docker-compose
chmod +x /usr/local/bin/docker-compose

Создание Docker-образа приложения

Добавьте файл Dockerfile в папку с кодом приложения:

# /Dockerfile

FROM python:3.9

WORKDIR /app

ENV PYTHONDONTWRITEBYTECODE 1
ENV PYTHONUNBUFFERED 1

RUN apt-get update && pip install --upgrade pip

COPY . .

RUN pip install --no-cache-dir --upgrade -r requirements.txt

CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "8000"]

EXPOSE 8000

Соберите образ локально:

docker build . -t pyrus-ext

Публикация в yandex registry

Создайте новый реестр Docker-образов в yandex cloud.

Сохраните идентификатор реестра, он понадобится далее.

Установите Yandex Cloud CLI и создайте профиль по инструкции.

Авторизуйтесь в Container Registry:

yc container registry configure-docker

Переименуем образ, созданный на предыдущем шаге. В качестве <ID реестра> используется идентификатор, полученный при создании реестра.

docker image tag pyrus-ext:latest cr.yandex/<ID реестра>/pyrus-ext:1.0

Загрузите необходимый Docker-образ в реестр:

docker push cr.yandex/<ID реестра>/pyrus-ext:1.0

Настройка доменного имени

Зарегистрируйте свой домен на любом регистраторе доменных имён, например, Рег.ру.

Добавьте ресурсную запись к домену и укажите публичный IP-адрес виртуальной машины.

Получение ssl сертификата Let's Encrypt

Установите утилиту Certbot для генерации SSL сертификатов:

sudo apt-get update 
sudo apt-get install certbot

Запустите Certbot для создания SSL сертификата:

sudo certbot certonly --manual --preferred-challenges dns -d example.com -d www.example.com

Замените example.com на ваш домен.

Certbot попросит вас подтвердить, что вы владеете доменом. Он также предоставит вам TXT запись, которую нужно добавить в DNS вашего домена.

Войдите в панель управления хостингом и добавьте TXT запись с указанными данными. Обновление DNS может занять некоторое время.

После того как запись будет добавлена и успешно обновлена, нажмите Enter для продолжения.

После этого сертификаты будут сохранены в /etc/letsencrypt/live/example.com/.

Настройка nginx

Создайте на севере файл nginx.conf.

Замените example.com на ваш домен.

server {
    listen 80;
    listen [::]:80;

	server_name example.com;

    location / {
        return 301 https://$host$request_uri;
    }
}

server {
    listen 443 ssl;
    listen [::]:443 ssl http2;

    server_name example.com;

	ssl_certificate /etc/nginx/certs/fullchain.pem;
    ssl_certificate_key /etc/nginx/certs/privkey.pem;

    location / {
			proxy_pass http://web:8000;
			proxy_set_header Host $host;
			proxy_set_header X-Real-IP $remote_addr;
			proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
			proxy_set_header X-Forwarded-Proto $scheme;
    }
}

Настройка Yandex Cloud CLI на сервере

Авторизация в Container Registry Yandex.Cloud

Установите инструмент командной строки yc cli:

curl -sSL https://storage.yandexcloud.net/yandexcloud-yc/install.sh | bash

Выполните команду для аутентификации в yandex cloud:

yc init

Следуйте инструкциям, чтобы войти в свою учетную запись Yandex.Cloud.

После успешной аутентификации выполните команду для авторизации в Container Registry:

yc container registry configure-docker

Эта команда настроит ваш Docker CLI для доступа к Container Registry.

Запуск с docker-compose

Создайте файл docker-compose.yaml

version: "3.7"

services:
  nginx:
	image: nginx:latest
    container_name: nginx
    restart: always
    ports:
      - 80:80
      - 443:443
    volumes:
	  - ./nginx.conf:/etc/nginx/conf.d/default.conf
      - /etc/letsencrypt/live/example.com/fullchain.pem:/etc/nginx/certs/fullchain.pem
      - /etc/letsencrypt/live/example.com/privkey.pem:/etc/nginx/certs/privkey.pem

  web:
    image: cr.yandex/<ID реестра>/pyrus-telegram-integration
	container_name: pyrus-telegram-integration
    restart: always
	ports:
	   - 8000:8000

Выполните команду docker ps и убедитесь, что контейнеры nginx и web запущены.

Теперь замените адрес сервиса в настройках своего расширения Pyrus на постоянный.