A python library for interacting with eXpress BotX API
Project description
pybotx
Библиотека для создания чат-ботов и SmartApps для мессенджера eXpress
Особенности
- Простая для использования
- Поддерживает коллбэки BotX
- Легко интегрируется с асинхронными веб-фреймворками
- Полное покрытие тестами
- Полное покрытие аннотациями типов
Установка
Используя poetry:
poetry add pybotx
Предупреждение: Данный проект находится в активной разработке (0.y.z) и
его API может быть изменён при повышении минорной версии.
Информация о мессенджере eXpress и платформе BotX
Документацию по мессенджеру (включая руководство пользователя и администратора) можно найти на официальном сайте.
Перед тем, как продолжать знакомство с библиотекой pybotx,
советуем прочитать данные статьи: Что такое чат-боты и SmartApp
и Взаимодействие с Bot API и BotX API
.
В этих статьях находятся исчерпывающие примеры работы с платформой, которые
легко повторить, используя pybotx.
Также не будет лишним ознакомиться с документацией по плаформе BotX .
Примеры готовых проектов на базе pybotx
- Next Feature Bot - бот, используемый для тестирования функционала платформы BotX.
- ToDo Bot - бот для ведения списка дел.
- Weather SmartApp - приложение для просмотра погоды.
Минимальный пример бота (интеграция с FastAPI)
from http import HTTPStatus
from uuid import UUID
from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse
# В этом и последующих примерах импорт из `pybotx` будет производиться
# через звёздочку для краткости. Однако, это не является хорошей практикой.
from pybotx import *
collector = HandlerCollector()
@collector.command("/echo", description="Send back the received message body")
async def echo_handler(message: IncomingMessage, bot: Bot) -> None:
await bot.answer_message(message.body)
# Сюда можно добавлять свои обработчики команд
# или копировать примеры кода, расположенные ниже.
bot = Bot(
collectors=[collector],
bot_accounts=[
BotAccountWithSecret(
# Не забудьте заменить эти учётные данные на настоящие,
# когда создадите бота в панели администратора.
id=UUID("123e4567-e89b-12d3-a456-426655440000"),
cts_url="https://cts.example.com",
secret_key="e29b417773f2feab9dac143ee3da20c5",
),
],
)
app = FastAPI()
app.add_event_handler("startup", bot.startup)
app.add_event_handler("shutdown", bot.shutdown)
# На этот эндпоинт приходят команды BotX
# (сообщения и системные события).
@app.post("/command")
async def command_handler(request: Request) -> JSONResponse:
bot.async_execute_raw_bot_command(
await request.json(),
request_headers=request.headers,
)
return JSONResponse(
build_command_accepted_response(),
status_code=HTTPStatus.ACCEPTED,
)
# К этому эндпоинту BotX обращается, чтобы узнать
# доступность бота и его список команд.
@app.get("/status")
async def status_handler(request: Request) -> JSONResponse:
status = await bot.raw_get_status(
dict(request.query_params),
request_headers=request.headers,
)
return JSONResponse(status)
# На этот эндпоинт приходят коллбэки с результатами
# выполнения асинхронных методов в BotX.
@app.post("/notification/callback")
async def callback_handler(request: Request) -> JSONResponse:
await bot.set_raw_botx_method_result(
await request.json(),
verify_request=False,
)
return JSONResponse(
build_command_accepted_response(),
status_code=HTTPStatus.ACCEPTED,
)
Примеры
Получение сообщений
from uuid import UUID
from pybotx import *
ADMIN_HUIDS = (UUID("123e4567-e89b-12d3-a456-426614174000"),)
collector = HandlerCollector()
@collector.command("/visible", description="Visible command")
async def visible_handler(_: IncomingMessage, bot: Bot) -> None:
# Обработчик команды бота. Команда видимая, поэтому описание
# является обязательным.
print("Hello from `/visible` handler")
@collector.command("/_invisible", visible=False)
async def invisible_handler(_: IncomingMessage, bot: Bot) -> None:
# Невидимая команда - не отображается в списке команд бота
# и не нуждается в описании.
print("Hello from `/invisible` handler")
async def is_admin(status_recipient: StatusRecipient, bot: Bot) -> bool:
return status_recipient.huid in ADMIN_HUIDS
@collector.command("/admin-command", visible=is_admin)
async def admin_command_handler(_: IncomingMessage, bot: Bot) -> None:
# Команда показывается только если пользователь является админом.
# Список команд запрашивается при открытии чата в приложении.
print("Hello from `/admin-command` handler")
@collector.default_message_handler
async def default_handler(_: IncomingMessage, bot: Bot) -> None:
# Если команда не была найдена, вызывается `default_message_handler`,
# если он определён. Такой обработчик может быть только один.
print("Hello from default handler")
Получение системных событий
from pybotx import *
collector = HandlerCollector()
@collector.chat_created
async def chat_created_handler(event: ChatCreatedEvent, bot: Bot) -> None:
# Работа с событиями производится с помощью специальных обработчиков.
# На каждое событие можно объявить только один такой обработчик.
print(f"Got `chat_created` event: {event}")
@collector.smartapp_event
async def smartapp_event_handler(event: SmartAppEvent, bot: Bot) -> None:
print(f"Got `smartapp_event` event: {event}")
Middlewares
(Этот функционал относится исключительно к pybotx)
from httpx import AsyncClient
from pybotx import *
collector = HandlerCollector()
async def custom_api_client_middleware(
message: IncomingMessage,
bot: Bot,
call_next: IncomingMessageHandlerFunc,
) -> None:
# До вызова `call_next` (обязателен в каждой миддлвари) располагается
# код, который выполняется до того, как сообщение дойдёт до
# своего обработчика.
async_client = AsyncClient()
# У сообщения есть объект состояния, в который миддлвари могут добавлять
# необходимые данные.
message.state.async_client = async_client
await call_next(message, bot)
# После вызова `call_next` выполняется код, когда обработчик уже
# завершил свою работу.
await async_client.aclose()
@collector.command(
"/fetch-resource",
description="Fetch resource from passed URL",
middlewares=[custom_api_client_middleware],
)
async def fetch_resource_handler(message: IncomingMessage, bot: Bot) -> None:
async_client = message.state.async_client
response = await async_client.get(message.argument)
print(response.status_code)
Сборщики обработчиков
(Этот функционал относится исключительно к pybotx)
from uuid import UUID, uuid4
from pybotx import *
ADMIN_HUIDS = (UUID("123e4567-e89b-12d3-a456-426614174000"),)
async def request_id_middleware(
message: IncomingMessage,
bot: Bot,
call_next: IncomingMessageHandlerFunc,
) -> None:
message.state.request_id = uuid4()
await call_next(message, bot)
async def ensure_admin_middleware(
message: IncomingMessage,
bot: Bot,
call_next: IncomingMessageHandlerFunc,
) -> None:
if message.sender.huid not in ADMIN_HUIDS:
await bot.answer_message("You are not admin")
return
await call_next(message, bot)
# Для того чтобы добавить новый обработчик команды,
# необходимо создать экземпляр класса `HandlerCollector`.
# Позже этот сборщик будет использован при создании бота.
main_collector = HandlerCollector(middlewares=[request_id_middleware])
# У сборщиков (как у обработчиков), могут быть собственные миддлвари.
# Они автоматически применяются ко всем обработчикам данного сборщика.
admin_collector = HandlerCollector(middlewares=[ensure_admin_middleware])
# Сборщики можно включать друг в друга. В данном примере у
# `admin_collector` будут две миддлвари. Первая - его собственная,
# вторая - полученная при включении в `main_collector`.
main_collector.include(admin_collector)
Отправка сообщения
from uuid import UUID
from pybotx import *
collector = HandlerCollector()
@collector.command("/answer", description="Answer to sender")
async def answer_to_sender_handler(message: IncomingMessage, bot: Bot) -> None:
# Т.к. нам известно, откуда пришло сообщение, у `pybotx` есть необходимый
# контекст для отправки ответа.
await bot.answer_message("Text")
@collector.command("/send", description="Send message to specified chat")
async def send_message_handler(message: IncomingMessage, bot: Bot) -> None:
try:
chat_id = UUID(message.argument)
except ValueError:
await bot.answer_message("Invalid chat id")
return
# В данном случае нас интересует не ответ, а отправка сообщения
# в другой чат. Чат должен существовать и бот должен быть в нём.
try:
await bot.send_message(
bot_id=message.bot.id,
chat_id=chat_id,
body="Text",
)
except Exception as exc:
await bot.answer_message(f"Error: {exc}")
return
await bot.answer_message("Message was send")
@collector.command("/prebuild-answer", description="Answer with prebuild message")
async def prebuild_answer_handler(message: IncomingMessage, bot: Bot) -> None:
# С помощью OutgoingMessage можно выносить логику
# формирования ответов в другие модули.
answer = OutgoingMessage(
bot_id=message.bot.id,
chat_id=message.chat.id,
body="Text",
)
await bot.send(message=answer)
Отправка сообщения с кнопками
from pybotx import *
collector = HandlerCollector()
@collector.command("/bubbles", description="Send buttons")
async def bubbles_handler(message: IncomingMessage, bot: Bot) -> None:
# Если вам нужна клавиатура под полем для ввода сообщения,
# используйте `KeyboardMarkup`. Этот класс имеет те же методы,
# что и `BubbleMarkup`.
bubbles = BubbleMarkup()
bubbles.add_button(
command="/choose",
label="Red",
data={"pill": "red"},
background_color="#FF0000",
)
bubbles.add_button(
command="/choose",
label="Blue",
data={"pill": "blue"},
background_color="#0000FF",
new_row=False,
)
# В кнопку можно добавит ссылку на ресурс,
# для этого нужно добавить url в аргумент `link`, а `command` оставить пустым,
# `alert` добавляется в окно подтверждения при переходе по ссылке.
bubbles.add_button(
label="Bubble with link",
alert="alert text",
link="https://example.com",
)
await bot.answer_message(
"The time has come to make a choice, Mr. Anderson:",
bubbles=bubbles,
)
Упоминание пользователя
from pybotx import *
collector = HandlerCollector()
@collector.command("/send-contact", description="Send author's contact")
async def send_contact_handler(message: IncomingMessage, bot: Bot) -> None:
contact = MentionBuilder.contact(message.sender.huid)
await bot.answer_message(f"Author is {contact}")
@collector.command("/echo-contacts", description="Send back recieved contacts")
async def echo_contact_handler(message: IncomingMessage, bot: Bot) -> None:
if not (contacts := message.mentions.contacts):
await bot.answer_message("Please send at least one contact")
return
answer = ", ".join(map(str, contacts))
await bot.answer_message(answer)
Отправка файла в сообщении
from aiofiles.tempfile import NamedTemporaryFile
from pybotx import *
collector = HandlerCollector()
@collector.command("/send-file", description="Send file")
async def send_file_handler(message: IncomingMessage, bot: Bot) -> None:
# Для создания файла используется file-like object
# с поддержкой асинхронных операций.
async with NamedTemporaryFile("wb+") as async_buffer:
await async_buffer.write(b"Hello, world!\n")
await async_buffer.seek(0)
file = await OutgoingAttachment.from_async_buffer(async_buffer, "test.txt")
await bot.answer_message("Attached file", file=file)
@collector.command("/echo-file", description="Echo file")
async def echo_file_handler(message: IncomingMessage, bot: Bot) -> None:
if not (attached_file := message.file):
await bot.answer_message("Attached file is required")
return
await bot.answer_message("", file=attached_file)
Редактирование сообщения
from pybotx import *
collector = HandlerCollector()
@collector.command("/increment", description="Self-updating widget")
async def increment_handler(message: IncomingMessage, bot: Bot) -> None:
if message.source_sync_id: # ID сообщения, в котором была нажата кнопка.
current_value = message.data["current_value"]
next_value = current_value + 1
else:
current_value = 0
next_value = 1
answer_text = f"Counter: {current_value}"
bubbles = BubbleMarkup()
bubbles.add_button(
command="/increment",
label="+",
data={"current_value": next_value},
)
if message.source_sync_id:
await bot.edit_message(
bot_id=message.bot.id,
sync_id=message.source_sync_id,
body=answer_text,
bubbles=bubbles,
)
else:
await bot.answer_message(answer_text, bubbles=bubbles)
Удаление сообщения
from pybotx import *
collector = HandlerCollector()
@collector.command("/deleted-message", description="Self-deleted message")
async def deleted_message_handler(message: IncomingMessage, bot: Bot) -> None:
if message.source_sync_id: # ID сообщения, в котором была нажата кнопка.
await bot.delete_message(
bot_id=message.bot.id,
sync_id=message.source_sync_id,
)
return
bubbles = BubbleMarkup()
bubbles.add_button(
command="/deleted-message",
label="Delete",
)
await bot.answer_message("Self-deleted message", bubbles=bubbles)
Обработчики ошибок
(Этот функционал относится исключительно к pybotx)
from loguru import logger
from pybotx import *
async def internal_error_handler(
message: IncomingMessage,
bot: Bot,
exc: Exception,
) -> None:
logger.exception("Internal error:")
await bot.answer_message(
"**Error:** internal error, please contact your system administrator",
)
# Для перехвата исключений существуют специальные обработчики.
# Бот принимает словарь из типов исключений и их обработчиков.
bot = Bot(
collectors=[],
bot_accounts=[],
exception_handlers={Exception: internal_error_handler},
)
Создание чата
from pybotx import *
collector = HandlerCollector()
@collector.command("/create-group-chat", description="Create group chat")
async def create_group_chat_handler(message: IncomingMessage, bot: Bot) -> None:
if not (contacts := message.mentions.contacts):
await bot.answer_message("Please send at least one contact")
return
try:
chat_id = await bot.create_chat(
bot_id=message.bot.id,
name="New group chat",
chat_type=ChatTypes.GROUP_CHAT,
huids=[contact.entity_id for contact in contacts],
)
except (ChatCreationProhibitedError, ChatCreationError) as exc:
await bot.answer_message(str(exc))
return
chat_mention = MentionBuilder.chat(chat_id)
await bot.answer_message(f"Chat created: {chat_mention}")
Поиск пользователей
import dataclasses
from pybotx import *
collector = HandlerCollector()
@collector.command("/my-info", description="Get info of current user")
async def search_user_handler(message: IncomingMessage, bot: Bot) -> None:
try:
user_info = await bot.search_user_by_huid(
bot_id=message.bot.id,
huid=message.sender.huid,
)
except UserNotFoundError: # Если пользователь и бот находятся на разных CTS
await bot.answer_message("User not found. Maybe you are on a different cts.")
return
await bot.answer_message(f"Your info:\n{dataclasses.asdict(user_info)}\n")
Получение списка пользователей
from pybotx import *
collector = HandlerCollector()
@collector.command("/get_users_list", description="Get a list of users")
async def users_list_handler(message: IncomingMessage, bot: Bot) -> None:
async with bot.users_as_csv(
bot_id=message.bot.id,
cts_user=True,
unregistered=False,
botx=False,
) as users:
async for user in users:
print(user)
Release history Release notifications | RSS feed
Download files
Download the file for your platform. If you're not sure which to choose, learn more about installing packages.
Source Distribution
pybotx-0.66.0.tar.gz
(70.4 kB
view hashes)
Built Distribution
pybotx-0.66.0-py3-none-any.whl
(120.0 kB
view hashes)
