XTwitter 线程嵌套回复会话 IDAPI 教程

Twitter 线程 API:获取 X 帖子上下文与嵌套回复

用 SocQ 获取 Twitter/X 线程上下文,按会话 ID 分组并连接直接回复。附关系构建与作者筛选代码,导出用于阅读、复核和摘要的 JSON。

SocQ更新于 2026年10月16日阅读约 7 分钟

用 SocQ 把目标帖、前文和回复关系放在一起,按“谁在回答谁”组织 Twitter/X 讨论。会话 ID 与直接父帖 ID 为阅读、复核和摘要提供共同的数据结构,方便继续筛选作者自己的串文。

SocQ X Thread Context API 查询目标公开帖子、沿回复链向前的帖子及来源返回的回复。本教程展示采集、会话分组、父子关系导出和作者串文筛选。

快速答案: 将公开帖子 URL 提交到 POST /v1/x/thread-context,读取成功任务的结果页。用 conversation_id 分组,用 in_reply_to_id 连接直接父帖。父帖正文未采集到时,保留引用并标记缺口。

单帖查询、回复采集与线程上下文怎么选?

需要的内容SocQ 接口起始输入
目标帖子本身X Posts公开帖子链接
目标帖子的回复X Post Replies公开帖子链接
目标帖、回复链前文和可用回复X Thread Context公开帖子链接
某个社区内发布的帖子X Community Posts公开社区链接

理解目标帖的前文时,使用 Thread Context。“前文”指沿直接父帖关系向前追溯的帖子。

例如,B 回复 A,C 回复 B,则 C 的前文关系是 C → B → A。按 conversation_id 分组,按 in_reply_to_id 连接,按发布时间展示时间顺序。

采集完成后,将正文和关系数据交给阅读界面或独立的摘要流程。

从公开帖子链接获取线程上下文

输入应包含 /status/{数字 ID},请将示例替换为实际研究的帖子:

curl -X POST "https://api.socq.ai/v1/x/thread-context" \
  -H "Authorization: Bearer $SOCQ_API_KEY" \
  -H "Content-Type: application/json" \
  -d '{
    "urls": ["https://x.com/BaganiTohid/status/2110635249889919362"],
    "results_limit": 100
  }'

输入支持 X 和 Twitter 的公开帖子链接,也支持从中间一条回复开始。将目标帖或回复的完整 /status/{数字ID} URL 放入 urls。

保存提交响应中的 data.task_id,查询该后台任务:

curl "https://api.socq.ai/v1/tasks/$TASK_ID?limit=100" \
  -H "Authorization: Bearer $SOCQ_API_KEY"

queued、running 或 retrying 时继续等待;failed 时查看 data.error_message。任务 succeeded 后读取 data.results.items,在 has_more=true 时沿 next_cursor 读取后续页。

results_limit 默认 100,允许 1 到 2000 的整数,分别作用于每个输入 URL。两个链接各设上限 100,表示分别最多采集 100 条;合并同一会话的多次采集时,按帖子 ID 去重。

Thread Context 接受范围内的任意整数上限。读取结果后,用关系字段构建父子连接;返回数组顺序用于遍历记录。

会话 ID 与直接父帖 ID 有什么区别?

两个关系字段各有用途:

字段含义处理方式
conversation_id帖子所属会话用相同值分组
in_reply_to_id当前帖子直接回复的父帖 ID按 ID 查找父帖并连接
id当前帖子标识建立帖子索引和去重

下面是用于解释关系的示意数据:

帖子作者所属会话直接父帖
AAliceA未显示
BAliceAA
CBobAB
DAliceAC

四条帖子都属于会话 A,D 的直接父帖是 C。按父帖 ID 建立连接后,保留 D → C → B → A 的回复路径。

字段定义见 Thread Context 字段说明。构建关系时,分别使用会话 ID 和直接父帖 ID。

标准记录还包含正文、作者、发布时间、采集时间、媒体和互动计数。来源未提供的字段保留 null。

用 Python 读取线程结果页

将这个只使用标准库的函数保存为 socq_collect.py:

import json
import os
import time
from urllib.parse import urlencode
from urllib.request import Request, urlopen

BASE = "https://api.socq.ai/v1"
PENDING = {"queued", "running", "retrying"}

def api(method, path, payload=None):
    body = json.dumps(payload).encode("utf-8") if payload is not None else None
    request = Request(
        BASE + path, data=body, method=method,
        headers={
            "Authorization": "Bearer " + os.environ["SOCQ_API_KEY"],
            "Content-Type": "application/json",
        },
    )
    with urlopen(request, timeout=60) as response:
        result = json.load(response)
    if result.get("code") != 200:
        raise RuntimeError(result)
    return result["data"]

def collect(endpoint, payload):
    task_id = api("POST", "/" + endpoint, payload)["task_id"]
    print("Task:", task_id, flush=True)
    path = "/tasks/" + task_id
    deadline = time.monotonic() + 600
    data = api("GET", path + "?limit=100")
    while data["status"] in PENDING:
        if time.monotonic() >= deadline:
            raise TimeoutError("Resume GET " + path + "; do not resubmit automatically")
        time.sleep(2)
        data = api("GET", path + "?limit=100")
    if data["status"] != "succeeded":
        raise RuntimeError(data.get("error_message") or data["status"])

    items, seen_cursors = [], set()
    while True:
        if time.monotonic() >= deadline:
            raise TimeoutError("Resume pagination for " + task_id)
        page = data["results"]
        items.extend(page["items"])
        if not page["has_more"]:
            return task_id, items
        cursor = page.get("next_cursor")
        if not cursor or cursor in seen_cursors:
            raise RuntimeError("Missing or repeated result cursor")
        seen_cursors.add(cursor)
        data = api("GET", path + "?" + urlencode({"limit": 100, "cursor": cursor}))
        if data["status"] != "succeeded":
            raise RuntimeError("Task is not ready for result pagination")

把下面的脚本保存为同目录下的 collect_thread.py:

import json
import sys
from socq_collect import collect

if len(sys.argv) != 2:
    raise SystemExit("Usage: python3 collect_thread.py PUBLIC_POST_URL")
payload = {"urls": [sys.argv[1]], "results_limit": 100}
task_id, posts = collect("x/thread-context", payload)
with open("thread-context.json", "w", encoding="utf-8") as output:
    json.dump({"task_id": task_id, "input": payload, "items": posts},
              output, ensure_ascii=False, indent=2)

设置 SOCQ_API_KEY 后,传入选定的公开帖子:

python3 collect_thread.py \
  "https://x.com/BaganiTohid/status/2110635249889919362"

thread-context.json 保存输入链接、任务 ID 和采集记录。函数在客户端最多等待十分钟,超时后保留打印的任务 ID,继续查询服务端任务状态。

结果分页读取该任务保存的记录。读完全部页后,得到该任务的完整结果集;会话覆盖范围以这些记录和缺口标记为准。

按直接父帖 ID 构建回复关系

先按帖子 ID 建立索引,再连接父帖与回复。父帖正文未采集到时,保留该 ID 并将 parent_in_dataset 标记为 false。

将下面代码保存为 build_thread.py:

import json
import sys

if len(sys.argv) not in (2, 3):
    raise SystemExit("Usage: python3 build_thread.py INPUT_JSON [THREAD_AUTHOR_ID]")
with open(sys.argv[1], encoding="utf-8") as source:
    records = json.load(source)["items"]
author_id = sys.argv[2] if len(sys.argv) == 3 else None

nodes, missing_id_records = {}, []
for post in records:
    if post.get("id"):
        nodes[str(post["id"])] = post
    else:
        missing_id_records.append(post)

edges, groups, unknown_conversation_ids, unknown_parent_ids = [], {}, [], []
for post_id, post in nodes.items():
    conversation_id = post.get("conversation_id")
    if conversation_id:
        groups.setdefault(str(conversation_id), []).append(post_id)
    else:
        unknown_conversation_ids.append(post_id)
    parent_id = post.get("in_reply_to_id")
    if parent_id is None:
        unknown_parent_ids.append(post_id)
    else:
        parent_id = str(parent_id)
        edges.append({
            "parent_id": parent_id, "child_id": post_id,
            "parent_in_dataset": parent_id in nodes,
        })

self_reply_edges = []
if author_id:
    for edge in edges:
        parent = nodes.get(edge["parent_id"])
        child = nodes[edge["child_id"]]
        if parent and edge["parent_id"] != edge["child_id"]:
            parent_author = (parent.get("author") or {}).get("id")
            child_author = (child.get("author") or {}).get("id")
            if parent_author == author_id and child_author == author_id:
                self_reply_edges.append(edge)

graph = {
    "nodes": list(nodes.values()), "edges": edges, "conversation_groups": groups,
    "missing_parent_ids": sorted({
        edge["parent_id"] for edge in edges if not edge["parent_in_dataset"]
    }),
    "unknown_conversation_post_ids": unknown_conversation_ids,
    "unknown_parent_post_ids": unknown_parent_ids,
    "records_without_id": missing_id_records,
    "selected_author_id": author_id, "selected_author_self_reply_edges": self_reply_edges,
}
with open("thread-graph.json", "w", encoding="utf-8") as output:
    json.dump(graph, output, ensure_ascii=False, indent=2)

运行关系导出:

python3 build_thread.py thread-context.json

生成的 thread-graph.json 包含:

  • 可识别的去重帖子节点,以及单独保留的无 ID 记录。
  • 父子关系边,使用 parent_in_dataset 表示父帖是否存在于数据中。
  • 已知会话 ID 的分组,以及缺失会话 ID 的帖子列表。
  • 已知父帖 ID、但没有采集到对应正文的引用。
  • 直接父帖字段为 null 的帖子 ID。

in_reply_to_id=null 表示这条记录未提供直接父帖引用,程序将其放入 unknown_parent_post_ids。有父帖 ID、但索引中没有对应记录时,程序保留关系边,并标记父帖缺失。

导出保留各个会话分组、已知关系边与缺失记录标记。阅读界面可以据此分别显示相连分支和上下文缺口。

区分作者串文与读者参与的讨论

在上述示意中,Alice 写了 A、B 和 D,其中 A 到 B 是作者自回复片段,D 是 Alice 对 Bob 的回复。按作者 ID 筛选得到作者的全部发言,再按父子双方的作者 ID 识别自回复片段。

已知串文作者的账号 ID 时,可以将该字符串作为第二个参数:

python3 build_thread.py thread-context.json THREAD_AUTHOR_ID

selected_author_self_reply_edges 保存父帖和子帖都属于指定作者的关系边。筛选前先确认想阅读的作者 ID;输入链接可以来自该作者,也可以来自读者回复。

以目标作者帖子为起点,选择相连的自回复片段。同一个父帖下有多条作者自回复时,导出保留这些分支,由阅读界面选择展示范围。

同时保存完整记录和作者筛选视图,后续可以基于原记录调整阅读范围。

导出为阅读、复核或摘要输入

阅读界面展示来源 URL、作者、发布时间和父帖关系,并为缺失父帖显示缺口标记。导出纯文本或 Markdown 时,同时保存关系 JSON,供后续恢复结构。

摘要流程读取帖子正文与关系数据,分别处理作者观点和读者回应,并保留上下文缺口标记。Thread Context 负责提供采集记录,摘要由你的模型或应用生成。

合并同一会话的多个任务时,按帖子 ID 去重,并保存各自的输入链接和任务 ID。后续采集中未观察到的帖子保留原记录和原采集时间。

需要先发现讨论再查看上下文时,可以从社区帖子采集教程或 X 搜索与采集教程开始。

让阅读、复核和摘要共用一份上下文

SocQ Thread Context 将帖子正文与关系 ID 放进同一份标准结果。配合本文代码,你可以建立会话分组、导出回复关系,再生成作者筛选视图。阅读页面和摘要流程都能使用这份记录,并保留来源与上下文缺口。

打开 Thread Context API,展开你的目标讨论。

来源与延伸阅读

常见问题

可以从中间一条回复开始,而不是第一条帖子吗?

可以,将该公开回复的 URL 放入 urls。任务按设定上限读取目标帖、沿回复链向前的公开帖子和来源返回的回复。

这与获取某条帖子的回复一样吗?

Post Replies 以目标帖子的回复为主;Thread Context 同时读取目标帖、前文和回复,适合按回复关系组织上下文。

能返回一段会话里的所有回复吗?

任务按每个输入的 results_limit 采集公开记录。读完结果页后得到这次保存的帖子;完整讨论的判断还需结合缺失父帖和回复记录。

如何处理回复下面的回复?

用 in_reply_to_id 连接直接父帖,用 conversation_id 分组。父帖未采集到时保留其 ID,并显示缺口。

为什么有父帖 ID,却没有父帖正文?

这表示父帖引用已返回,但对应正文未包含在本次记录中。将引用保留下来,需要补充上下文时再查询该帖;具体缺失原因以来源返回的信息为准。

如何只保留原作者的串文?

确认作者 ID,再保留父帖与子帖均属于该作者的关系边。以目标帖子为起点选择相连片段,自回复有分支时由阅读界面确定展示范围。

能把线程保存下来交给 AI 总结吗?

可以,将正文、作者、时间和关系保存为 JSON,再执行独立摘要步骤。摘要旁应保留来源链接和上下文缺口标记。

X API

使用公开的 X 输入测试工作流

提交公开输入,获取带有可追溯来源上下文的标准化记录。

查看 X API