Telegram 频道监测 API 教程

监测一个公开 Telegram 频道,归根到底是三个动作:建立基线、在新帖出现时抓住它、以及衡量它的反响。这篇教程用 SandBase Telegram API 把这三个动作串成一套工作流——一个 SandBase 密钥,不需要 MTProto 客户端,也不需要登录 Telegram。这些是同步的请求/响应读取,所以这里的”监测”指的是按计划轮询,而不是实时流。

完整的端点全景见 Telegram 公开数据 API 总览。这一篇是落地的监测工作流。

先说结论

三步:channel-info(基线)→ channel-posts(新帖)→ post-comments(互动)。

每次都是 POST /v1/api/telegram/<path>,一个 SANDBASE_API_KEY;completed 运行才带 outputs 数组,failed/timeout 运行带 error 而没有 outputs。

channel-posts 响应用游标分页,所以记住上次见到的 after_cursor,下次轮询把它作为 after 请求参数传回去,拉更新的帖子。

仅公开、只读数据;你这边不用登录 Telegram,但仍需要一个 SandBase API 密钥。

工作流全貌

步骤	端点	输入	你拿到
1. 建基线	`telegram/web/channel-info`	`channel`	订阅数、计数器、认证
2. 拉新帖	`telegram/web/channel-posts`	`channel`	近期消息 + 分页游标
3. 读互动	`telegram/web/post-comments`	`channel`、整数 `post_id`	某帖的评论串

端点 API 参考是每个参数名和响应路径的事实来源。

第 1 步 —— 给频道建基线

先写一个判 status 的辅助函数,再读频道拿一份基线:

import os
import requests

BASE = "https://api.sandbase.ai/v1/api/telegram"
HEADERS = {
    "Authorization": f"Bearer {os.environ['SANDBASE_API_KEY']}",
    "Content-Type": "application/json",
}


def call(path: str, payload: dict) -> dict:
    resp = requests.post(f"{BASE}/{path}", headers=HEADERS, json=payload, timeout=90)
    resp.raise_for_status()
    body = resp.json()
    if body.get("status") != "completed":
        raise RuntimeError(body.get("error", {}).get("message", "请求未完成"))
    # 参考只保证信封;业务字段随端点而定,
    # 用防御式读取,并以一份真实响应核对。
    return body["outputs"][0]["data"]


info = call("web/channel-info", {"channel": "telegram"})
print(info.get("title"), info.get("subscribers"), info.get("verified"))

把订阅数和计数器存下来,后续运行就能画出随时间的增长。下面是我测试于 2026-09-27(UTC)的真实响应——completed 运行会把业务字段放在 outputs[0].data 下,所以代码仍对每个字段用 .get() 读取,因为这些数值(订阅数、计数器)会变、以真实响应为准:

{
  "id": "8c815356-e1f6-4668-959c-375f9471822f",
  "status": "completed",
  "model": "telegram/web/channel-info",
  "outputs": [
    {
      "data": {
        "title": "Telegram News",
        "username": "telegram",
        "subscribers": "9.46M",
        "verified": true,
        "counters": { "photos": "16", "videos": "228", "links": "378" }
      }
    }
  ]
}

第 2 步 —— 轮询新帖

读频道的近期帖子。响应带一个 messages 列表和一个带游标字段的 pagination 对象。channel-posts 接受 after、before、limit(1-100,默认 20)请求参数:after 传一个游标值拉比它更新的帖子,before 传游标值拉更早的。在下面这份实测响应里,after_cursor 是本页最新的游标,before_cursor 是最旧的,has_more_before 表示还有更早的帖子。参考并不保证帖子 id 单调递增,所以增量轮询要用游标驱动——记住上次见到的 after_cursor,作为 after 传回去:

def poll_new(channel: str, after_cursor=None) -> tuple[list[dict], object]:
    payload = {"channel": channel}
    if after_cursor is not None:
        # 把上次见到的游标传进去,拉比它更新的帖子
        payload["after"] = after_cursor
    data = call("web/channel-posts", payload)
    messages = data.get("messages", [])
    pagination = data.get("pagination", {})
    # 推进游标;本页为空时保留上一个
    next_cursor = pagination.get("after_cursor", after_cursor)
    return messages, next_cursor


new_posts, cursor = poll_new("telegram")
for m in new_posts[:5]:
    print(m.get("id"), m.get("date"), (m.get("text") or "")[:60])

每条消息带 id、date、text、text_html、一个 author、一个 reactions 列表、views、media、link_preview、type、url 和转发标记(is_forwarded、forwarded_from、reply_to)等字段。记住上次见到的 after_cursor、作为 after 传回去,就把整整一页变成一个干净的”自上次运行以来有什么新的”拉取。若要反过来回补更早的历史,传 before 请求参数加上 before_cursor 值来取上一页。下面是我测试于 2026-09-27(UTC)的真实响应;其中 id、views、游标等取值会随时间变化,所以要防御式读取,并以真实响应为准:

{
  "id": "8c815356-e1f6-4668-959c-375f9471822f",
  "status": "completed",
  "model": "telegram/web/channel-posts",
  "outputs": [
    {
      "data": {
        "messages": [
          {
            "id": 460,
            "date": "2026-08-26T19:12:34+00:00",
            "text": "…",
            "views": "1.48M",
            "is_forwarded": false,
            "reactions": []
          }
        ],
        "pagination": { "after_cursor": 460, "before_cursor": 441, "has_more_before": true, "limit": 20 }
      }
    }
  ]
}

channel-posts 返回一个 messages 列表和一个 pagination 对象;用 after 拉更新的、用 before 请求参数往前翻更早的。

第 3 步 —— 读一条帖子的互动

对值得细看的帖子,读它的评论串。post-comments 接 channel 和一个整数 post_id:

for m in new_posts:
    post_id = m.get("id")
    if not isinstance(post_id, int):
        continue
    thread = call("web/post-comments", {"channel": "telegram", "post_id": post_id})
    comments = thread.get("comments", [])
    print(post_id, "->", len(comments), "条评论")

评论负载嵌在 comments 下。某些深度读取的响应里可能注明有上游要求——先看一次真实响应、读一遍结构,再依赖某个具体字段。

post-comments 按 channel 和整数 post_id 读取某帖的回复。

把它串起来

一次最小的监测过程长这样——建一次基线,然后按计划轮询、每次推进 after_cursor:

CHANNEL = "telegram"
cursor = None

def monitor_once():
    global cursor
    fresh, cursor = poll_new(CHANNEL, cursor)
    for m in fresh:
        record = {
            "id": m.get("id"),
            "date": m.get("date"),
            "text": m.get("text"),
            "views": m.get("views"),
            "reactions": m.get("reactions", []),
        }
        # 存下 record;需要的话对它调 post-comments 拿互动
    return len(fresh)

因为每次调用共用同一个信封和同一个 call 辅助函数,加重试或速率退避是一处改动的事。当你需要的不止这些读取时,查线上 Telegram 列表找到合适的端点,接入前先确认它的参数。

为什么在 API 层做监测

你当然可以搭一个 MTProto 客户端去盯一个频道,但那意味着要管一个会话、处理重连、维护一个解析器,而不是在做产品。通过一层统一 API 来读,意味着你的代码依赖的是有名字的 JSON 字段和单个响应信封。鉴权是一个密钥,而且因为 completed 运行都返回相同的 { id, status, model, outputs } 结构(failed/timeout 运行则带 error 而没有 outputs),重试、日志和错误处理都在一个辅助函数里。记住上次见到的 after_cursor,让每次轮询成为一个干净的拉取,于是你的流水线只看到新帖——不重复告警、不重复存同一条消息。

这种一致性让工作流可组合。把频道换成任意一个公开频道,加一次 similar-channels 调用来扩大覆盖,它也照样接在同一个辅助函数后面。你的时间花在”这些帖子对你的监测意味着什么”上,而不是花在维持一个协议客户端上。

局限与边界

仅公开、只读数据。 不发帖、不做机器人操作、不涉及私有或仅账号可见的数据。
轮询,不是流式。 这些是同步读取;按节奏轮询、每次推进 after_cursor。
分页是请求参数。 channel-posts 接受 after、before、limit(1-100,默认 20);after 传上次的 after_cursor 拉更新的帖子,before 传 before_cursor 往前翻更早的。确切字段看 channel-posts 的结构。
速率与量级。 把响应当作尽力而为的读取;遇到 HTTP 429 等瞬时错误时按退避策略重试,并控制轮询节奏。
以线上参考核对。 可用性和字段可能变化;在依赖某个具体端点前先确认。

常见问题

我需要 Telegram 机器人 token 或登录吗? 不需要。你用自己的 SANDBASE_API_KEY 向 SandBase 鉴权。这套工作流读的是公开频道数据,不需要你这边有 Telegram 账号、机器人 token 或 MTProto 会话。

我怎么只抓新帖? 参考并不保证帖子 id 单调递增,所以用游标来驱动:记住上次见到的 after_cursor,下次轮询把它作为 after 请求参数传回去,只拉更新的帖子。若要回补更早的历史,传 before 请求参数加上 before_cursor 值。

我能读私聊或群组吗? 不能。这套工作流只是公开频道数据——频道信息、帖子和公开的评论串。私聊、群组和账号授权内容不在范围内。