用 SocQ 按社区收集 Twitter/X 讨论，把帖子正文、作者、发布时间和来源保存在一起。将反复出现的问题整理成选题，并为每个判断保留原帖依据，你的内容研究就有了可回溯的数据集。
本教程使用 SocQ X Community Posts 采集帖子，再通过 X Community 查询社区简介与规则。示例输出带来源的 JSON 和按社区整理的 CSV。
快速答案： 将
x.com/i/communities/{id}链接提交到POST /v1/x/community-posts，保存任务 ID，成功后读取全部结果页。同一个链接通过POST /v1/x/community查询社区资料。为每条帖子附上输入社区，便于连接资料与讨论数据。
使用公开社区链接作为输入
输入应采用这样的格式，请替换为你实际研究的公开社区：
https://x.com/i/communities/1493446837214187523
链接中的数字是社区 ID。两个社区接口都使用完整的 /i/communities/{数字ID} 链接作为输入。
只知道社区名称时，先找到对应的公开社区链接，再提交到 urls。
保存输入链接、数字社区 ID、帖子的发布时间 published_at 与采集时间 collected_at。用社区 ID 匹配 x.com 与受支持的 twitter.com 链接，分别保留名称相同但 ID 不同的社区。
获取 Twitter/X 社区信息流中的帖子
设置环境变量 SOCQ_API_KEY，提交每个社区最多 40 条的采集任务：
curl -X POST "https://api.socq.ai/v1/x/community-posts" \
-H "Authorization: Bearer $SOCQ_API_KEY" \
-H "Content-Type: application/json" \
-d '{
"urls": ["https://x.com/i/communities/1493446837214187523"],
"results_limit": 40
}'
保存响应里的 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。
Community Posts 按创建时间顺序返回记录。报表需要按点赞或评论排序时，在本地对采集记录排序，并将结果标注为样本排名。
每社区上限与结果分页有什么区别？
results_limit 默认 100，允许 20 到 2000 之间的 20 的倍数，例如 20、40、60、100。
上限分别作用于每个输入社区。两个链接设置 results_limit=40，表示每个社区最多采集 40 条；实际条数以任务保存的去重结果为准。
任务查询中的 limit 控制一页读取多少条已保存记录，next_cursor 定位下一页。采集上限由提交任务时的 results_limit 决定。
区分社区资料、成员数量与帖子数据
| 资源 | 主要标准字段 | 可以回答的问题 |
|---|---|---|
| 社区资料 | name、text、rules、join_policy、invites_policy | 记录中的简介、规则和参与政策是什么？ |
| 社区数量 | metrics.members_count、metrics.moderators_count | 来源报告了多少成员和版主？ |
| 社区帖子 | id、text、author、published_at、media | 这次采集窗口包含哪些讨论？ |
| 帖子指标 | 可用点赞、评论、转发和浏览数 | 采集时这些样本之间有什么差别？ |
社区资料需要单独请求：
curl -X POST "https://api.socq.ai/v1/x/community" \
-H "Authorization: Bearer $SOCQ_API_KEY" \
-H "Content-Type: application/json" \
-d '{"urls":["https://x.com/i/communities/1493446837214187523"]}'
使用资料任务自己的 ID 读取结果。成员数与版主数记录社区规模；帖子浏览数记录该帖的浏览计数。活跃作者数或独立触达人数需要对应的独立统计。
字段为 null 表示来源未提供该测量值，0 表示返回计数为零。导出时分别保留这两种值。
用 Python 导出社区帖子的 JSON 和 CSV
下面的流程只使用 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")
函数打印每个任务 ID，最多等待十分钟，并读取全部已保存结果页。客户端超时后保留 ID，通过原任务继续查询。
再将以下代码保存为同目录下的 export_communities.py：
import csv
import json
import sys
from urllib.parse import urlparse
from socq_collect import collect
urls = list(dict.fromkeys(sys.argv[1:]))
if not urls:
raise SystemExit("Usage: python3 export_communities.py COMMUNITY_URL [COMMUNITY_URL...]")
def cell(value):
text = "" if value is None else str(value)
return "'" + text if text.lstrip().startswith(("=", "+", "-", "@")) else text
fields = ["community_id", "community_url", "post_id", "post_url", "text",
"author_id", "author_username", "published_at", "collected_at",
"likes_count", "comments_count", "shares_count", "views_count"]
bundles, seen = [], set()
with open("community-posts.csv", "w", newline="", encoding="utf-8-sig") as output:
writer = csv.DictWriter(output, fieldnames=fields)
writer.writeheader()
for url in urls:
metadata_task, communities = collect("x/community", {"urls": [url]})
posts_task, posts = collect("x/community-posts", {
"urls": [url], "results_limit": 40,
})
bundles.append({
"input_url": url, "metadata_task_id": metadata_task,
"posts_task_id": posts_task, "communities": communities, "posts": posts,
})
# Save completed collections before processing another community.
with open("community-results.json", "w", encoding="utf-8") as saved:
json.dump(bundles, saved, ensure_ascii=False, indent=2)
community_id = urlparse(url).path.rstrip("/").split("/")[-1]
for post in posts:
post_id = post.get("id")
key = (community_id, post_id)
if post_id and key in seen:
continue
if post_id:
seen.add(key)
author = post.get("author") or {}
metrics = post.get("metrics") or {}
row = {
"community_id": community_id, "community_url": url,
"post_id": post_id, "post_url": post.get("url"), "text": post.get("text"),
"author_id": author.get("id"), "author_username": author.get("username"),
"published_at": post.get("published_at"),
"collected_at": post.get("collected_at"),
**{name: metrics.get(name) for name in
("likes_count", "comments_count", "shares_count", "views_count")},
}
writer.writerow({key: cell(value) for key, value in row.items()})
运行时传入一个或多个公开社区链接：
python3 export_communities.py \
"https://x.com/i/communities/1493446837214187523"
程序生成两份文件：
community-results.json保存各输入链接、两个任务 ID、社区资料和帖子记录。community-posts.csv按社区输出帖子行，包含来源、作者、时间以及可用指标。
示例为每个社区单独创建资料和帖子任务，并将输入社区附在帖子旁边。这样每条帖子都能明确对应采集来源。批量请求适用于合并读取；需要逐条归属时，采用这里的单社区任务方式。
去重键是 community_id + post_id。同一社区内的重复行合并，同一帖子在不同社区采集到时分别保留；缺少帖子 ID 的记录单独保存。
程序在每个社区完成后更新文件。后续请求失败时，文件保留此前完成的社区结果；用于报表前，按输入链接核对已完成的社区和错误记录。
JSON 保留完整数据，CSV 用于表格复核
CSV 会展开数据结构；嵌套规则、媒体和作者字段仍保存在 JSON 中。以后调整表格列时，可以基于保存的 JSON 重新生成，不必为了改列再次采集。
示例使用 Python 标准库的 CSV 写入器处理引号、逗号和多行文本，带字节顺序标记的 UTF-8 方便常见表格工具识别编码。CSV 为首个非空白字符是 =、+、- 或 @ 的文本加上单引号，JSON 保留原文。较长数字 ID 导入表格时应设为文本，避免精度丢失。
用社区讨论寻找内容研究方向
例如，研究“这次采集的讨论中反复出现哪些上手问题”，按问题分类，并为每类保存支持它的帖子 ID。
先阅读社区描述与规则，再结合帖子正文、发布时间和来源链接复核。将常见问题、产品抱怨或资源需求归类时，把支持判断的帖子 ID 一起保存。
使用社区详情流程保存简介和规则，再由应用执行主题归类或摘要。
按点赞或评论排序时，将结果命名为“本次采集样本排名”，并注明采集时间和条数。某条帖子在后续采集中缺席时，将其记为“本次未观察到”；需要确认删除时，单独查询该帖。
进一步查看某条讨论的上下文
需要展开某条讨论时，将返回的帖子 URL 提交到 X Thread Context，读取目标帖、沿回复链向前的帖子及回复关系。
具体方法见 Twitter 线程上下文教程。如果只需要已知帖子的回复，使用 X Post Replies；如果起点是账号而非社区，使用 X User Posts。
让内容判断有原帖依据
SocQ 的社区资料与帖子接口可以串起“确认社区、采集讨论、按主题归类”的研究流程。本文代码为帖子保留社区来源，并输出 JSON 与 CSV，方便内容团队复核问题、整理案例和准备摘要输入。
打开 Community Posts API，采集你的目标社区。
来源与延伸阅读
- SocQ Community Posts 文档：社区链接、每社区上限和标准帖子字段。
- SocQ Community 文档：资料、规则和数量字段。
常见问题
可以同时采集多个 X 社区吗？
可以，在 urls 中提交多个社区链接，上限分别作用于每个链接。Python 示例使用独立任务，以明确保留每条帖子的输入社区来源。
能获得社区从创建至今的所有帖子吗？
一次任务读取指定社区的一批帖子，上限由 results_limit 控制，范围由该次返回记录确定。历史分析使用分次保存的采集记录，并注明各次范围。
帖子之外，还能导出社区成员名单吗？
Community 返回社区资料及成员、版主数量；Community Posts 返回帖子。具体成员名单属于独立数据，不在这两个接口中。
社区信息流和用户时间线有什么区别？
社区信息流组织某个讨论空间内的帖子，用户时间线组织某个账号发布的内容。Community Posts 接收社区链接，User Posts 接收用户名。
能选择 Top 或点赞排序吗？
接口按创建时间顺序返回。在应用中按点赞或评论数排序，即可生成本次样本的指标排名。
读取下一页会采集新帖子吗？
游标读取原任务保存的下一页。需要新一轮帖子观察时，提交新任务，并按帖子 ID 处理两轮结果中的重叠记录。
社区迁移后如何采集帖子？
X 已发布 Communities 向 XChat 的迁移公告。将目标社区 URL 提交到 Community Posts，成功后读取返回的帖子；空结果和失败分别记录。保留每轮采集时间，便于区分历史记录与新一轮查询。