Telegram机器人利用Bot API导出群组消息记录的方法

为什么你需要用 Bot 导出群组消息？

在日常运营中，我们常常需要保留群组内的关键消息，作为数据备份、内容分析或合规存档。Telegram 官方客户端虽然支持手动滚动加载历史记录，但无法批量导出到本地。而 Bot API 提供了一套自动化接口，可以让我们用代码实时获取群组中的新消息并记录下来。本文将从零开始，带你掌握利用 Bot 导出群组消息的完整方法论。

Bot API 的局限与应对思路

首先要明确一点：Telegram Bot API 并不支持直接拉取群组的历史消息。机器人只能接收发送它之后产生的更新（update）。这意味着“导出”往往需要从机器人加入群组的那一刻开始记录。对于存量历史，你可以结合官方桌面客户端的“导出聊天记录”功能，但那是人工操作。这里我们聚焦于自动化方案，即通过 Bot 实时捕获消息流。

方案一：getUpdates 长轮询实现消息记录

机器人通过 getUpdates 接口轮询获取新消息。你只需要将机器人加入目标群组，并授予它“读取消息”的权限（管理员或者普通成员均可，取决于群组设置）。

步骤 1：创建机器人并获取 Token

在 Telegram 中与 @BotFather 对话，使用 /newbot 创建一个新机器人，拿到 API Token。

步骤 2：将机器人加入群组

将机器人添加到需要记录的群组中。注意，如果群组开启了“机器人无法查看消息”的限制，你需要将其设为管理员并开启“读取消息”权限。

步骤 3：编写 Python 脚本轮询消息

import requests
import json

TOKEN = "你的BOT_TOKEN"
URL = f"https://api.telegram.org/bot/getUpdates"

def get_updates(offset=None):
    params = {"timeout": 30, "offset": offset}
    resp = requests.get(URL, params=params)
    return resp.json()

def save_message(msg):
    with open("messages.jsonl", "a", encoding="utf-8") as f:
        f.write(json.dumps(msg, ensure_ascii=False) + "\n")

last_update_id = None
while True:
    data = get_updates(last_update_id)
    for result in data.get("result", []):
        last_update_id = result["update_id"] + 1
        message = result.get("message")
        if message and message.get("chat", {}).get("type") in ("group", "supergroup"):
            save_message(message)
            print(f"[{message['chat']['title']}] {message['from']['first_name']}: {message.get('text')}")

这段代码会持续监听群组消息，并将每条消息以 JSON 格式追加到本地文件。你可以根据需求扩展为写入数据库或导出为 CSV。

方案二：Webhook 模式实时推送

当机器人处于 Webhook 模式时，Telegram 会主动将更新推送到你的服务器。这种方式更高效，适合生产环境。

设置 Webhook

首先需要在你的服务器上运行一个 HTTPS 服务，接收 POST 请求。然后调用 setWebhook 接口指定回调地址。

curl -X POST "https://api.telegram.org/bot你的TOKEN/setWebhook?url=https://你的域名/webhook"

处理推送数据

以 Flask 为例，一个简单的 Webhook 接收端如下：

from flask import Flask, request
import json

app = Flask(__name__)

def save_message(update):
    msg = update.get("message")
    if msg:
        with open("messages.jsonl", "a", encoding="utf-8") as f:
            f.write(json.dumps(msg, ensure_ascii=False) + "\n")

@app.route("/webhook", methods=["POST"])
def webhook():
    update = request.get_json()
    save_message(update)
    return "OK"

if __name__ == "__main__":
    app.run(port=8443, ssl_context=("cert.pem", "key.pem"))

注意：Telegram 要求 Webhook 必须使用 HTTPS 证书，你可以申请免费证书或使用反向代理。

方案三：结合 getChat 补充群组元数据

在导出消息时，往往需要附带群组名称、ID 等元数据。调用 getChat 接口可以获取当前群组的基础信息。

GET https://api.telegram.org/bot/getChat?chat_id=@your_group_username

返回的 JSON 中包含 title、type、username 等字段，将这些信息与消息记录一并存储，可以让导出数据更完整。

实用技巧与注意事项

消息类型多样：Bot API 的消息对象包含文本、图片、视频、文档、贴纸等，需要针对不同类型进行解析，避免只记录 text 字段。
更新丢失：getUpdates 在超时或异常时可能造成消息遗漏，建议在本地维护 offset，并考虑加数据库事务。
隐私合规：导出群组消息可能涉及成员隐私，务必遵循当地法律法规，并提前告知群成员。
性能瓶颈：如果群组消息量巨大，建议使用消息队列（如 RabbitMQ）或直接写入数据库，避免文件 IO 阻塞。

总结

虽然 Telegram Bot API 无法直接导出历史消息，但通过实时监听并持久化，我们可以轻松实现群组消息的自动化归档。本文介绍的 getUpdates 和 Webhook 两种模式，覆盖了从简单脚本到生产级部署的常见场景。希望这套方案能帮助你构建自己的群组消息导出工具。