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)

fromhttpimportHTTPStatusfromuuidimportUUIDfromfastapiimportFastAPI, Requestfromfastapi.responsesimportJSONResponse# В этом и последующих примерах импорт из `pybotx` будет производиться# через звёздочку для краткости. Однако, это не является хорошей практикой.frompybotximport*collector=HandlerCollector()
@collector.command("/echo", description="Send back the received message body")asyncdefecho_handler(message: IncomingMessage, bot: Bot) ->None:
awaitbot.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")asyncdefcommand_handler(request: Request) ->JSONResponse:
bot.async_execute_raw_bot_command(
awaitrequest.json(),
request_headers=request.headers,
)
returnJSONResponse(
build_command_accepted_response(),
status_code=HTTPStatus.ACCEPTED,
)
# На этот эндпоинт приходят события BotX для SmartApps, обрабатываемые синхронно.@app.post("/smartapps/request")asyncdefsync_smartapp_event_handler(request: Request) ->JSONResponse:
response=awaitbot.sync_execute_raw_smartapp_event(
awaitrequest.json(),
request_headers=request.headers,
)
returnJSONResponse(response.jsonable_dict(), status_code=HTTPStatus.OK)
# К этому эндпоинту BotX обращается, чтобы узнать# доступность бота и его список команд.@app.get("/status")asyncdefstatus_handler(request: Request) ->JSONResponse:
status=awaitbot.raw_get_status(
dict(request.query_params),
request_headers=request.headers,
)
returnJSONResponse(status)
# На этот эндпоинт приходят коллбэки с результатами# выполнения асинхронных методов в BotX.@app.post("/notification/callback")asyncdefcallback_handler(request: Request) ->JSONResponse:
awaitbot.set_raw_botx_method_result(
awaitrequest.json(),
verify_request=False,
)
returnJSONResponse(
build_command_accepted_response(),
status_code=HTTPStatus.ACCEPTED,
)

Примеры

Получение сообщений

(подробное описание функции)

fromuuidimportUUIDfrompybotximport*ADMIN_HUIDS= (UUID("123e4567-e89b-12d3-a456-426614174000"),)
collector=HandlerCollector()
@collector.command("/visible", description="Visible command")asyncdefvisible_handler(_: IncomingMessage, bot: Bot) ->None:
# Обработчик команды бота. Команда видимая, поэтому описание# является обязательным.print("Hello from `/visible` handler")
@collector.command("/_invisible", visible=False)asyncdefinvisible_handler(_: IncomingMessage, bot: Bot) ->None:
# Невидимая команда - не отображается в списке команд бота# и не нуждается в описании.print("Hello from `/invisible` handler")
asyncdefis_admin(status_recipient: StatusRecipient, bot: Bot) ->bool:
returnstatus_recipient.huidinADMIN_HUIDS@collector.command("/admin-command", visible=is_admin)asyncdefadmin_command_handler(_: IncomingMessage, bot: Bot) ->None:
# Команда показывается только если пользователь является админом.# Список команд запрашивается при открытии чата в приложении.print("Hello from `/admin-command` handler")
@collector.default_message_handlerasyncdefdefault_handler(_: IncomingMessage, bot: Bot) ->None:
# Если команда не была найдена, вызывается `default_message_handler`,# если он определён. Такой обработчик может быть только один.print("Hello from default handler")

Получение системных событий

(подробное описание функции)

frompybotximport*collector=HandlerCollector()
@collector.chat_createdasyncdefchat_created_handler(event: ChatCreatedEvent, bot: Bot) ->None:
# Работа с событиями производится с помощью специальных обработчиков.# На каждое событие можно объявить только один такой обработчик.print(f"Got `chat_created` event: {event}")
@collector.smartapp_eventasyncdefsmartapp_event_handler(event: SmartAppEvent, bot: Bot) ->None:
print(f"Got `smartapp_event` event: {event}")

Получение синхронных SmartApp событий

frompybotximport*collector=HandlerCollector()
# Обработчик синхронных Smartapp событий, приходящих на эндпоинт `/smartapps/request`@collector.sync_smartapp_eventasyncdefhandle_sync_smartapp_event(
event: SmartAppEvent, bot: Bot,
) ->BotAPISyncSmartAppEventResultResponse:
print(f"Got sync smartapp event: {event}")
returnBotAPISyncSmartAppEventResultResponse.from_domain(
data={},
files=[],
)

Middlewares

(Этот функционал относится исключительно к pybotx)

fromhttpximportAsyncClientfrompybotximport*collector=HandlerCollector()
asyncdefcustom_api_client_middleware(
message: IncomingMessage,
bot: Bot,
call_next: IncomingMessageHandlerFunc,
) ->None:
# До вызова `call_next` (обязателен в каждой миддлвари) располагается# код, который выполняется до того, как сообщение дойдёт до# своего обработчика.async_client=AsyncClient()
# У сообщения есть объект состояния, в который миддлвари могут добавлять# необходимые данные.message.state.async_client=async_clientawaitcall_next(message, bot)
# После вызова `call_next` выполняется код, когда обработчик уже# завершил свою работу.awaitasync_client.aclose()
@collector.command("/fetch-resource",description="Fetch resource from passed URL",middlewares=[custom_api_client_middleware],)asyncdeffetch_resource_handler(message: IncomingMessage, bot: Bot) ->None:
async_client=message.state.async_clientresponse=awaitasync_client.get(message.argument)
print(response.status_code)

Сборщики обработчиков

(Этот функционал относится исключительно к pybotx)

fromuuidimportUUID, uuid4frompybotximport*ADMIN_HUIDS= (UUID("123e4567-e89b-12d3-a456-426614174000"),)
asyncdefrequest_id_middleware(
message: IncomingMessage,
bot: Bot,
call_next: IncomingMessageHandlerFunc,
) ->None:
message.state.request_id=uuid4()
awaitcall_next(message, bot)
asyncdefensure_admin_middleware(
message: IncomingMessage,
bot: Bot,
call_next: IncomingMessageHandlerFunc,
) ->None:
ifmessage.sender.huidnotinADMIN_HUIDS:
awaitbot.answer_message("You are not admin")
returnawaitcall_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)

Отправка сообщения

(подробное описание функции)

fromuuidimportUUIDfrompybotximport*collector=HandlerCollector()
@collector.command("/answer", description="Answer to sender")asyncdefanswer_to_sender_handler(message: IncomingMessage, bot: Bot) ->None:
# Т.к. нам известно, откуда пришло сообщение, у `pybotx` есть необходимый# контекст для отправки ответа.awaitbot.answer_message("Text")
@collector.command("/send", description="Send message to specified chat")asyncdefsend_message_handler(message: IncomingMessage, bot: Bot) ->None:
try:
chat_id=UUID(message.argument)
exceptValueError:
awaitbot.answer_message("Invalid chat id")
return# В данном случае нас интересует не ответ, а отправка сообщения# в другой чат. Чат должен существовать и бот должен быть в нём.try:
awaitbot.send_message(
bot_id=message.bot.id,
chat_id=chat_id,
body="Text",
)
exceptExceptionasexc:
awaitbot.answer_message(f"Error: {exc}")
returnawaitbot.answer_message("Message was send")
@collector.command("/prebuild-answer", description="Answer with prebuild message")asyncdefprebuild_answer_handler(message: IncomingMessage, bot: Bot) ->None:
# С помощью OutgoingMessage можно выносить логику# формирования ответов в другие модули.answer=OutgoingMessage(
bot_id=message.bot.id,
chat_id=message.chat.id,
body="Text",
)
awaitbot.send(message=answer)

Отправка сообщения с кнопками

(подробное описание функции)

frompybotximport*collector=HandlerCollector()
@collector.command("/bubbles", description="Send buttons")asyncdefbubbles_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",
)
awaitbot.answer_message(
"The time has come to make a choice, Mr. Anderson:",
bubbles=bubbles,
)

Упоминание пользователя

(подробное описание функции)

frompybotximport*collector=HandlerCollector()
@collector.command("/send-contact", description="Send author's contact")asyncdefsend_contact_handler(message: IncomingMessage, bot: Bot) ->None:
contact=MentionBuilder.contact(message.sender.huid)
awaitbot.answer_message(f"Author is {contact}")
@collector.command("/echo-contacts", description="Send back recieved contacts")asyncdefecho_contact_handler(message: IncomingMessage, bot: Bot) ->None:
ifnot (contacts:=message.mentions.contacts):
awaitbot.answer_message("Please send at least one contact")
returnanswer=", ".join(map(str, contacts))
awaitbot.answer_message(answer)

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

(подробное описание функции)

fromaiofiles.tempfileimportNamedTemporaryFilefrompybotximport*collector=HandlerCollector()
@collector.command("/send-file", description="Send file")asyncdefsend_file_handler(message: IncomingMessage, bot: Bot) ->None:
# Для создания файла используется file-like object# с поддержкой асинхронных операций.asyncwithNamedTemporaryFile("wb+") asasync_buffer:
awaitasync_buffer.write(b"Hello, world!\n")
awaitasync_buffer.seek(0)
file=awaitOutgoingAttachment.from_async_buffer(async_buffer, "test.txt")
awaitbot.answer_message("Attached file", file=file)
@collector.command("/echo-file", description="Echo file")asyncdefecho_file_handler(message: IncomingMessage, bot: Bot) ->None:
ifnot (attached_file:=message.file):
awaitbot.answer_message("Attached file is required")
returnawaitbot.answer_message("", file=attached_file)

Редактирование сообщения

(подробное описание функции)

frompybotximport*collector=HandlerCollector()
@collector.command("/increment", description="Self-updating widget")asyncdefincrement_handler(message: IncomingMessage, bot: Bot) ->None:
ifmessage.source_sync_id: # ID сообщения, в котором была нажата кнопка.current_value=message.data["current_value"]
next_value=current_value+1else:
current_value=0next_value=1answer_text=f"Counter: {current_value}"bubbles=BubbleMarkup()
bubbles.add_button(
command="/increment",
label="+",
data={"current_value": next_value},
)
ifmessage.source_sync_id:
awaitbot.edit_message(
bot_id=message.bot.id,
sync_id=message.source_sync_id,
body=answer_text,
bubbles=bubbles,
)
else:
awaitbot.answer_message(answer_text, bubbles=bubbles)

Удаление сообщения

(подробное описание функции)

frompybotximport*collector=HandlerCollector()
@collector.command("/deleted-message", description="Self-deleted message")asyncdefdeleted_message_handler(message: IncomingMessage, bot: Bot) ->None:
ifmessage.source_sync_id: # ID сообщения, в котором была нажата кнопка.awaitbot.delete_message(
bot_id=message.bot.id,
sync_id=message.source_sync_id,
)
returnbubbles=BubbleMarkup()
bubbles.add_button(
command="/deleted-message",
label="Delete",
)
awaitbot.answer_message("Self-deleted message", bubbles=bubbles)

Обработчики ошибок

(Этот функционал относится исключительно к pybotx)

fromloguruimportloggerfrompybotximport*asyncdefinternal_error_handler(
message: IncomingMessage,
bot: Bot,
exc: Exception,
) ->None:
logger.exception("Internal error:")
awaitbot.answer_message(
"**Error:** internal error, please contact your system administrator",
)
# Для перехвата исключений существуют специальные обработчики.# Бот принимает словарь из типов исключений и их обработчиков.bot=Bot(
collectors=[],
bot_accounts=[],
exception_handlers={Exception: internal_error_handler},
)

Создание чата

(подробное описание функции)

frompybotximport*collector=HandlerCollector()
@collector.command("/create-group-chat", description="Create group chat")asyncdefcreate_group_chat_handler(message: IncomingMessage, bot: Bot) ->None:
ifnot (contacts:=message.mentions.contacts):
awaitbot.answer_message("Please send at least one contact")
returntry:
chat_id=awaitbot.create_chat(
bot_id=message.bot.id,
name="New group chat",
chat_type=ChatTypes.GROUP_CHAT,
huids=[contact.entity_idforcontactincontacts],
)
except (ChatCreationProhibitedError, ChatCreationError) asexc:
awaitbot.answer_message(str(exc))
returnchat_mention=MentionBuilder.chat(chat_id)
awaitbot.answer_message(f"Chat created: {chat_mention}")

Поиск пользователей

(подробное описание функции)

importdataclassesfrompybotximport*collector=HandlerCollector()
@collector.command("/my-info", description="Get info of current user")asyncdefsearch_user_handler(message: IncomingMessage, bot: Bot) ->None:
try:
user_info=awaitbot.search_user_by_huid(
bot_id=message.bot.id,
huid=message.sender.huid,
)
exceptUserNotFoundError: # Если пользователь и бот находятся на разных CTSawaitbot.answer_message("User not found. Maybe you are on a different cts.")
returnawaitbot.answer_message(f"Your info:\n{dataclasses.asdict(user_info)}\n")

Получение списка пользователей

(подробное описание функции)

frompybotximport*collector=HandlerCollector()
@collector.command("/get_users_list", description="Get a list of users")asyncdefusers_list_handler(message: IncomingMessage, bot: Bot) ->None:
asyncwithbot.users_as_csv(
bot_id=message.bot.id,
cts_user=True,
unregistered=False,
botx=False,
) asusers:
asyncforuserinusers:
print(user)