Документация по функциям платформы Telegram
TelegramAdapter — это адаптер, построенный на основе Telegram Bot API, поддерживающий различные типы сообщений и обработку событий.
Информация о документации
- Соответствующая версия модуля: 4.2.0
- Поддерживается: ErisPulse
Основная информация
- Краткое описание платформы: Telegram - это мультиплатформенный мессенджер
- Имя адаптера: TelegramAdapter
- Поддерживаемый протокол/API-версия: Telegram Bot API
- Сопоставление типов сессий:
private→ при отправке используетсяuser,group/supergroup→group,channel→channel
Стандартные действия API (DSL API)
Адаптер сопоставляет стандартные действия OB12 с Telegram Bot API и стандартизирует поле data:
| Стандартное действие OB12 | Telegram API | Поле data |
|---|---|---|
| get_self_info | getMe | user_id / user_name / user_displayname |
| get_user_info(user_id) | getChat | user_id / user_name / user_displayname |
| get_group_info(group_id) | getChat | group_id / group_name |
| get_group_member_info(group_id, user_id) | getChatMember | user_id / user_name / telegram_role |
| delete_message(message_id) | deleteMessage | chat_id автоматически заполняется по таблице сообщений |
| leave_group(group_id) | leaveChat | - |
Расширенные действия: get_group_admin_list(group_id) (список администраторов), get_chat_member_count(chat_id) (количество участников).
from ErisPulse import sdk
telegram = sdk.adapter.get("telegram")
result = await telegram.Api.get_self_info()
result = await telegram.Api.get_group_info(group_id=-100123)
result = await telegram.Api.get_group_admin_list(-100123)
await telegram.Api.delete_message(message_id=55) # chat_id автоматически заполняется
result = await telegram.Api.Using("main").get_self_info()
Telegram Bot API не имеет интерфейса для получения списка друзей/групп, get_friend_list/get_group_list возвращает errorcode=10002.
Операции с запросами (Request DSL)
Обработка запросов на добавление в группу (событие chat_join_request), на основе approveChatJoinRequest / declineChatJoinRequest:
from ErisPulse.Core.Event import request
@request.on_request()
async def handle_join_request(event):
if event.get("platform") != "telegram":
return
# event["request_id"] - это синтетический идентификатор (tjr_{chat_id}_{user_id}_{date})
if event.get("user_nickname"):
await event.approve() # Подтвердить
# await event.reject() # Отклонить
# Вызов вручную (требуется предварительное получение соответствующего события запроса для регистрации контекста)
await telegram.Request(event["request_id"]).accept()
await telegram.Request(event["request_id"]).reject()
await telegram.Request(event["request_id"]).Using("main").accept()
Типы поддерживаемых сообщений
Все методы отправки реализованы с использованием цепочечного синтаксиса, например:
from ErisPulse.Core import adapter
telegram = adapter.get("telegram")
await telegram.Send.To("user", user_id).Text("Hello World!")
Основные методы отправки
| Метод | Описание | Параметры |
|---|---|---|
.Text(text) |
Отправка обычного текстового сообщения | text: str |
.Face(emoji) |
Отправка эмодзи-кости | emoji: str (например 🎲 🎯 🏀) |
.Markdown(text, content_type) |
Отправка сообщения в формате Markdown | content_type по умолчанию "MarkdownV2" |
.HTML(text) |
Отправка сообщения в формате HTML | text: str |
.Sticker(file) |
Отправка стикера | file: str (file_id/URL) | bytes |
.Location(lat, lng) |
Отправка геолокации | latitude: float, longitude: float |
.Venue(lat, lng, title, addr) |
Отправка места | С заголовком и адресом |
.Contact(phone, first, last) |
Отправка контакта | С номером телефона и именем |
Методы отправки медиа
Все методы медиа поддерживают два типа входных данных: bytes (загрузка) и str (file_id / URL):
| Метод | Описание |
|---|---|
.Image(file, caption, content_type) |
Отправка изображения |
.Video(file, caption, content_type) |
Отправка видео |
.Voice(file, caption) |
Отправка голосового сообщения |
.Audio(file, caption, content_type) |
Отправка аудио |
.File(file, caption) |
Отправка файла |
.Document(file, caption, content_type) |
Псевдоним для File |
Методы управления сообщениями
| Метод | Описание |
|---|---|
.Edit(message_id, text, content_type) |
Редактирование существующего сообщения |
.Recall(message_id) |
Удаление указанного сообщения |
.Forward(from_chat_id, message_id) |
Пересылка сообщения (с сохранением источника) |
.CopyMessage(from_chat_id, message_id) |
Копирование сообщения (без источника) |
.AnswerCallback(callback_query_id, text, show_alert) |
Ответ на запрос обратной связи |
Отправка сообщений в исходном виде
.Raw_ob12(message: List[Dict]): Отправка сообщения в формате OneBot12.Raw_json(json_str: str): Отправка сообщения в формате JSON
Методы цепочечного изменения
| Метод | Описание |
|---|---|
.At(user_id) |
Упоминание пользователя (через entities в Telegram, можно вызывать несколько раз) |
.AtAll() |
Упоминание всех участников (отправка текста @All) |
.Reply(message_id) |
Ответ на указанное сообщение |
.Keyboard(inline_keyboard) |
Установка инлайн-клавиатуры (list[list[dict]]) |
.ProtectContent(protect) |
Защита контента (предотвращение пересылки и сохранения) |
.Silent(silent) |
Отправка в тихом режиме (без уведомления пользователя) |
Примеры отправки сообщений
# Базовая отправка текста
await telegram.Send.To("user", user_id).Text("Hello World!")
# Сообщение с инлайн-клавиатурой
from ErisPulse import sdk
telegram = sdk.adapter.get("telegram")
keyboard = [
[{"text": "Кнопка 1", "callback_data": "btn1"}, {"text": "Кнопка 2", "callback_data": "btn2"}],
[{"text": "Перейти на сайт", "url": "https://example.com"}],
]
await telegram.Send.To("group", group_id).Keyboard(keyboard).Text("Выберите:")
# Отправка медиа (по URL)
await telegram.Send.To("group", group_id).Image("https://example.com/image.jpg", caption="Изображение")
# Упоминание пользователя
await telegram.Send.To("group", group_id).At("6117725680").Text("Привет!")
# Ответ + защита контента
await telegram.Send.To("group", group_id).Reply("12345").ProtectContent().Text("Секретное сообщение")
# Отправка в тихом режиме
await telegram.Send.To("group", group_id).Silent().Text("Тихое уведомление")
# Ответ на запрос обратной связи
await telegram.Send.AnswerCallback(callback_query_id, text="Обработано", show_alert=False)
# Сложное сообщение в формате OneBot12
ob12_message = [
{"type": "text", "data": {"text": "Сложное сообщение: "}},
{"type": "mention", "data": {"user_id": "6117725680", "user_name": "Имя пользователя"}},
{"type": "reply", "data": {"message_id": "12345"}},
{"type": "image", "data": {"file": "https://http.cat/200"}}
]
await telegram.Send.To("group", group_id).Raw_ob12(ob12_message)
# Отправка стикера
await telegram.Send.To("user", user_id).Sticker("CAACAgIAAxkBAA...") # file_id
# Отправка геолокации
await telegram.Send.To("user", user_id).Location(39.9042, 116.4074)
Типы специфических событий
Преобразование событий Telegram следует стандарту OneBot12, а также предоставляет расширения платформы с префиксом telegram_.
Сопоставление detail_type для событий сообщений
| Telegram chat.type | OneBot12 detail_type | Тип получателя |
|---|---|---|
private |
private |
user |
group |
group |
group |
supergroup |
group |
group |
channel |
channel |
channel |
Специфические типы событий
| detail_type | Описание |
|---|---|
telegram_callback_query |
Запрос обратной связи (нажатие кнопки встроенной клавиатуры) |
telegram_inline_query |
Встроенный запрос |
telegram_chosen_inline_result |
Выбранный результат встроенного запроса |
telegram_poll |
Событие голосования |
telegram_poll_answer |
Ответ на голосование |
telegram_my_chat_member |
Изменение статуса участника бота |
telegram_chat_member |
Изменение участника чата |
telegram_chat_join_request |
Запрос на присоединение к чату |
telegram_shipping_query |
Запрос доставки |
telegram_pre_checkout_query |
Запрос предоплаты |
Стандартные типы сообщений
Используемый формат сообщений OneBot12:
| Тип сообщения | Описание | Поля data |
|---|---|---|
text |
Чистый текст (без @имени пользователя) | text |
mention |
@пользователь (стандартный OB12) | user_id, user_name |
reply |
Ссылка на ответ | message_id, user_id |
image |
Изображение | file_id, url |
video |
Видео | file_id, url, duration, width, height |
voice |
Голосовое сообщение | file_id, url, duration |
audio |
Аудио | file_id, url, duration, title, performer |
file |
Файл | file_id, url, file_name, file_size, mime_type |
location |
Местоположение | latitude, longitude, необязательно title, address |
Расширенные сообщения платформы
Расширенные сообщения с префиксом telegram_:
| Тип сообщения | Описание | Поля data |
|---|---|---|
telegram_sticker |
Стикер | file_id, emoji, sticker_type, url |
telegram_animation |
Анимация GIF | file_id, url, duration, caption |
telegram_contact |
Контакт | phone_number, first_name, last_name, user_id |
telegram_inline_keyboard |
Встроенная клавиатура | inline_keyboard |
Примеры событий
Сообщение в группе (с упоминанием)
{
"type": "message",
"detail_type": "group",
"platform": "telegram",
"user_id": "6117725680",
"user_nickname": "WSu2059",
"group_id": "-1002850921906",
"message_id": "172",
"message": [
{"type": "text", "data": {"text": "/it.echo "}},
{"type": "mention", "data": {"user_id": "", "user_name": "@nm123_91178"}}
],
"alt_message": "/it.echo @nm123_91178",
"telegram_chat": {
"id": -1002850921906,
"title": "ErisPulse",
"username": "erispulse",
"type": "supergroup"
}
}
Событие запроса обратной связи
{
"type": "notice",
"detail_type": "telegram_callback_query",
"user_id": "123456",
"user_nickname": "YingXinche",
"telegram_callback_id": "cb_123",
"telegram_callback_data": "callback_data",
"message_id": "msg_456"
}
Событие встроенного запроса
{
"type": "request",
"detail_type": "telegram_inline_query",
"user_id": "789012",
"user_nickname": "YingXinche",
"telegram_query_id": "iq_789",
"telegram_query_text": "search_text",
"telegram_query_offset": "0"
}
Сообщение с встроенной клавиатурой
{
"type": "message",
"detail_type": "group",
"message": [
{"type": "text", "data": {"text": "Выберите: "}},
{
"type": "telegram_inline_keyboard",
"data": {
"inline_keyboard": [
[{"text": "Кнопка1", "callback_data": "btn1"}],
[{"text": "Перейти", "url": "https://example.com"}]
]
}
}
]
}
Event Mixin расширения методов
Адаптер зарегистрировал следующие методы, специфичные для платформы, доступные только при platform == "telegram":
Связанные с сообщениями
| Метод | Тип возвращаемого значения | Описание |
|---|---|---|
is_bot_message() |
bool |
Проверяет, исходит ли сообщение от бота |
is_edited_message() |
bool |
Проверяет, является ли сообщение отредактированным |
is_topic_message() |
bool |
Проверяет, является ли сообщение тематическим/Topic |
get_update_id() |
int |
Получает ID обновления Telegram |
get_chat_title() |
str |
Получает название чата |
get_chat_username() |
str |
Получает имя пользователя чата |
get_forward_from() |
dict |
Получает информацию о источнике пересылки |
get_topic_id() |
str |
Получает идентификатор темы |
Связанные с обратными запросами
| Метод | Тип возвращаемого значения | Описание |
|---|---|---|
get_callback_data() |
str |
Получает callback_data обратного запроса |
get_callback_id() |
str |
Получает идентификатор обратного запроса (для ответа) |
Извлечение данных сегментов сообщений
| Метод | Тип возвращаемого значения | Описание |
|---|---|---|
get_inline_keyboard() |
list |
Получает встроенную клавиатуру из сообщения |
get_sticker_info() |
dict |
Получает информацию о стикере |
get_contact_info() |
dict |
Получает информацию о контакте |
get_location() |
dict |
Получает информацию о местоположении |
Пример использования
from ErisPulse.Core.Event import message, notice
@message.on_message()
async def handle_message(event):
if event.get("platform") != "telegram":
return
# Свойства сообщения
if event.is_bot_message():
return # Пропустить сообщения от бота
if event.is_edited_message():
print("Это отредактированное сообщение")
# Информация о чате
title = event.get_chat_title()
username = event.get_chat_username()
# Источник пересылки
forward = event.get_forward_from()
# Данные сегментов сообщения
sticker = event.get_sticker_info()
contact = event.get_contact_info()
location = event.get_location()
keyboard = event.get_inline_keyboard()
# Тема
if event.is_topic_message():
topic_id = event.get_topic_id()
@notice.on_notice()
async def handle_notice(event):
if event.get("platform") != "telegram":
return
if event.get("detail_type") == "telegram_callback_query":
callback_data = event.get_callback_data()
callback_id = event.get_callback_id()
# Ответить на обратный запрос
telegram = sdk.adapter.get("telegram")
await telegram.Send.AnswerCallback(callback_id, text="Нажато")
# Ответить на сообщение
await event.reply(f"Вы нажали: {callback_data}")
Описание расширенных полей
- Все специфические поля идентифицируются с префиксом
telegram_ - Исходные данные сохраняются в поле
telegram_raw - Тип исходного события сохраняется в поле
telegram_raw_type - Сообщения в каналах используют
detail_type="channel" - Личные сообщения используют
detail_type="private"(при отправке необходимо преобразовать вuser) - Сообщения в темах содержат поле
thread_id - Упоминания с помощью
@используют стандартный тип сообщения упоминания (type: "mention"), текст не содержит @имени пользователя
Параметры конфигурации
Адаптер Telegram поддерживает конфигурацию нескольких аккаунтов:
Пример конфигурации
[Telegram_Adapter.accounts.default]
token = "ВАШ_ТОКЕН_БОТА"
enabled = true
[Telegram_Adapter.accounts.bot2]
token = "ДРУГОЙ_ТОКЕН_БОТА"
enabled = true
Режимы работы
Адаптер Telegram поддерживает только режим опроса (Polling), режим вебхуков (Webhook) был удален.
Конфигурация прокси
Если необходимо подключиться к Telegram API через прокси, используйте системные прокси-настройки (переменные окружения ALL_PROXY / HTTPS_PROXY).
Перенос старой конфигурации
Старый формат конфигурации с одним токеном будет автоматически поддерживаться:
# Старый формат (все еще может использоваться, но рекомендуется перейти на новый)
[Telegram_Adapter]
token = "ВАШ_ТОКЕН_БОТА"
Рекомендуется перейти на новый формат:
[Telegram_Adapter.accounts.default]
token = "ВАШ_ТОКЕН_БОТА"
enabled = true