简体中文 English 繁體中文 日本語 Русский
本文为静态镜像,内容以交互版为准 在交互式文档中心打开 →

Telegram平台特性文档

TelegramAdapter 是基于 Telegram Bot API 构建的适配器,支持多种消息类型和事件处理。


文档信息

基本信息

标准Api动作(Api DSL)

适配器将 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 返回 etcode=10002。


请求操作(Request DSL)

处理加群申请(chat_join_request 事件),基于 pproveChatJoinRequest / 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) 应答回调查询

原始消息发送

链式修饰方法

方法 说明
.At(user_id) @指定用户(通过 Telegram entities 实现,可多次调用)
.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 Bot 自身成员状态变更
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 获取 Telegram update ID
get_chat_title() str 获取聊天标题
get_chat_username() str 获取聊天用户名
get_forward_from() dict 获取转发来源信息
get_topic_id() str 获取话题 ID

回调查询相关

方法 返回类型 说明
get_callback_data() str 获取回调查询的 callback_data
get_callback_id() str 获取回调查询 ID(用于应答)

消息段数据提取

方法 返回类型 说明
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_Adapter.accounts.default]
token = "YOUR_BOT_TOKEN"
enabled = true

[Telegram_Adapter.accounts.bot2]
token = "ANOTHER_BOT_TOKEN"
enabled = true

运行模式

Telegram 适配器仅支持 Polling(轮询) 模式,Webhook 模式已移除。

代理配置

如需通过代理连接 Telegram API,请使用系统级代理(环境变量 ALL_PROXY / HTTPS_PROXY)。

旧版配置迁移

旧版单 token 配置会自动兼容:

# 旧版格式(仍可使用,但建议迁移)
[Telegram_Adapter]
token = "YOUR_BOT_TOKEN"

建议迁移到新格式:

[Telegram_Adapter.accounts.default]
token = "YOUR_BOT_TOKEN"
enabled = true