用 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 | 当前帖子标识 | 建立帖子索引和去重 |
下面是用于解释关系的示意数据:
| 帖子 | 作者 | 所属会话 | 直接父帖 |
|---|---|---|---|
| A | Alice | A | 未显示 |
| B | Alice | A | A |
| C | Bob | A | B |
| D | Alice | A | C |
四条帖子都属于会话 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,展开你的目标讨论。
来源与延伸阅读
- SocQ Thread Context 文档:公开输入、数量上限和标准关系字段。
常见问题
可以从中间一条回复开始,而不是第一条帖子吗?
可以,将该公开回复的 URL 放入 urls。任务按设定上限读取目标帖、沿回复链向前的公开帖子和来源返回的回复。
这与获取某条帖子的回复一样吗?
Post Replies 以目标帖子的回复为主;Thread Context 同时读取目标帖、前文和回复,适合按回复关系组织上下文。
能返回一段会话里的所有回复吗?
任务按每个输入的 results_limit 采集公开记录。读完结果页后得到这次保存的帖子;完整讨论的判断还需结合缺失父帖和回复记录。
如何处理回复下面的回复?
用 in_reply_to_id 连接直接父帖,用 conversation_id 分组。父帖未采集到时保留其 ID,并显示缺口。
为什么有父帖 ID,却没有父帖正文?
这表示父帖引用已返回,但对应正文未包含在本次记录中。将引用保留下来,需要补充上下文时再查询该帖;具体缺失原因以来源返回的信息为准。
如何只保留原作者的串文?
确认作者 ID,再保留父帖与子帖均属于该作者的关系边。以目标帖子为起点选择相连片段,自回复有分支时由阅读界面确定展示范围。
能把线程保存下来交给 AI 总结吗?
可以,将正文、作者、时间和关系保存为 JSON,再执行独立摘要步骤。摘要旁应保留来源链接和上下文缺口标记。