用 Python 调用 Threads 关键词搜索 API：监测品牌提及

发布于 2026年10月16日 · 更新于2026年10月16日

桌上三张便签，每张一个词：产品名、审核早就认得的错拼、这一季的活动标签。这就是全部输入。不是词库，不是客户表，也不是 Instagram 那种 pagination_token。运营早会对的是这三件事：公开讨论里有没有产品名、错拼会不会漏、活动标签是不是只剩自己人在用。一次配方吃这三个词，比先画「全网声量」更接近真实工作。Threads 公开数据 API 的 search-posts 每次只收一个必填 query。成品是当前目录里的 threads_matches.csv：每一行对应一条嵌套在 thread_items 里的帖子，带上命中的词和你观测到它的 UTC 时间。关键词命中是公开样本里的一条帖子，不是投诉单，不是客户身份，更不是「全网提及已经收齐」。

登录后从控制台复制自己有权使用的平台 Key，放进服务端环境变量 OPENDATA_API_KEY。文中的 YOUR_API_KEY 只是占位符。打开本页或复制脚本都不会发请求；替换占位符并主动执行才会调用、扣额度。Key 出现在查询字符串里，生产流量放服务端，日志、工单、截图里的 URL 要脱敏。不需要 Threads 登录凭据。面向台湾市场做「脆」的公开讨论研究时，也先把原文和来源留下来，同名、反讽、路人闲聊交给人看，不要让脚本替你定性。

三个词就是全部监测名单

品牌待审队列不必从一千条同义词起家。产品名、常见错拼、活动标签——你本来就要人工看的三个词，够写出这张表。词一多，额度先花在重复样本上，审核队列也会被同义句淹没。本文只跑文档里的公开示例 coffee，避免复制页面变成对你品牌词的一次真实搜索。另外两个词，等你打算花额度时再写进同一个 Python 列表。coffee 和你补上的词都不保证有结果，也不代表任何品牌的口碑结论。

每个词对应一次 GET：https://api.opendata-api.com/api/project/threads-scraper/search-posts。这里只接受 query。高级语法、日期、排序、page、pagination_token 都未提供，不要自造参数。中文、空格、标点交给 params 编码，不要手拼查询串。三个词就是三次调用，不是一次请求里塞数组。

传输失败、HTTP 4xx/5xx、网关信封不是业务成功，是三种停法。信封是 {code, message, data}。HTTP 2xx 只说明字节到了；网关 code 为 200 只说明平台接受了这次调用。外层 data 里还有业务 status（成功时观察到为 ok）、request_id 和内层 data。HTTP 200 不是提及，网关 200 也不是提及。两层都过了之后，results 为空数组是「这次样本里没有」，不是故障，更不要重试到表格看起来热闹为止。某一词空、另外一词有行，也是正常采样，不要把空词改成「失败」。

展开 thread_items，留下帖子 id

帖子不在外层包装上。两层状态都成功后，沿业务对象上的 data.results[].node.thread.thread_items[].post 往下走。一个包装里可以有多帖：有人回了主帖、同线程里还有一条带关键词的跟帖，外层仍可能只算一条结果。用 len(results) 当提及数会少算；跳过 thread_items 会丢掉同一条线程里的帖。这不是 Instagram 用户帖翻页，也没有可回传的游标。看见 cursor 就当它是只读痕迹，不要抄进下一轮参数。

每条 post 把 pk、id、短代码 code 分开存成字符串，不要转成浮点数。表格软件一开，长数字最容易变成科学计数，标识就对不上了。集合里的 id 可能带账号后缀，不能拿去当 code，也不是 get-post-details 的入参——那个接口要 code 或 threads.com / threads.net 上的 HTTPS 帖子 url。本配方不调用它。CSV 是待审队列；只有审核员点名某一条时，再拿短代码去查详情。数字看起来的帖子 ID 不是短代码，不要拿去当 code 试。

caption.text 和 user.username 有就留下，没有就空着。taken_at 是 Unix 秒的发布时间。正文为空也不要丢行，审核还需要 id，有时图、转发或残缺正文仍然值得核对来源。命中 coffee 不是投诉、不是情感、更不是可以去私信作者的许可。同一帖撞上两个词，是两条查询关联，不是两个品牌；去重按帖子键，关联按词另记，才不会把一条讨论算成两次声量。

审核能打开的 CSV

先单独安装 requests，在服务端跑脚本。它在工作目录写入 threads_matches.csv：按字符串 pk 去重，先到的观测留下，同一词或另一词再次出现则跳过。审核打开时要能按词筛选、按时间排序、点开短代码去核对原文，而不是面对一坨无法引用的包装 JSON。列名固定，审核能直接打开、又不会把标识弄坏的列：

pk，id，code，username，text，taken_at，like_count，query，observed_at_utc，request_id

like_count 是采集当时的观测，不是触达。缺值就留空，不要写成 0。表格软件里把标识和正文指定为纯文本，避免长数字变科学计数、不可信正文被当成公式。若有人用 Excel 打开，对以 =、+、-、@ 开头的单元格加前缀。

observed_at_utc 是你取到样本的时间，不是 taken_at，两列都留。request_id 留给脱敏排查，不要把带 Key 的完整 URL 记进日志。

import csv
import os
from datetime import datetime, timezone
from pathlib import Path

import requests

BASE = "https://api.opendata-api.com/api/project/threads-scraper"
API_KEY = os.environ["OPENDATA_API_KEY"]
# Documented public example. Add the other two phrases locally when you mean to spend credits.
WATCHLIST = ["coffee"]
OUTPUT = Path("threads_matches.csv")
FIELDS = [
    "pk", "id", "code", "username", "text", "taken_at",
    "like_count", "query", "observed_at_utc", "request_id",
]


def read_search(query):
    try:
        response = requests.get(
            f"{BASE}/search-posts",
            params={"key": API_KEY, "query": query},
            timeout=30,
        )
    except requests.RequestException:
        raise RuntimeError("Transport failure; inspect redacted logs") from None
    if not 200 <= response.status_code < 300:
        raise RuntimeError(f"HTTP failure: {response.status_code}")
    try:
        envelope = response.json()
    except ValueError:
        raise RuntimeError("Expected a JSON response") from None
    if not isinstance(envelope, dict) or str(envelope.get("code")) != "200":
        raise RuntimeError("Gateway business failure")
    payload = envelope.get("data")
    if not isinstance(payload, dict) or payload.get("status") != "ok":
        raise RuntimeError("Operation failed or has an unexpected status")
    nested = payload.get("data")
    if not isinstance(nested, dict) or not isinstance(nested.get("results"), list):
        raise RuntimeError("Expected a search results array")
    return payload, nested["results"]


def existing_pks(path):
    if not path.exists() or path.stat().st_size == 0:
        return set()
    with path.open(newline="", encoding="utf-8") as source:
        return {row["pk"] for row in csv.DictReader(source) if row.get("pk")}


def flatten(results):
    for result in results:
        try:
            items = result["node"]["thread"]["thread_items"]
        except (TypeError, KeyError):
            raise RuntimeError("Unexpected search wrapper") from None
        if not isinstance(items, list):
            raise RuntimeError("Expected thread_items to be a list")
        for item in items:
            try:
                post = item["post"]
            except (TypeError, KeyError):
                raise RuntimeError("Missing post on thread item") from None
            yield post


def rows_for(query, payload, results, seen):
    observed_at = datetime.now(timezone.utc).isoformat()
    new_rows = []
    for post in flatten(results):
        pk = post.get("pk")
        if not isinstance(pk, str) or not pk:
            raise RuntimeError("Expected a string post pk")
        if pk in seen:
            continue
        seen.add(pk)
        new_rows.append({
            "pk": pk,
            "id": post.get("id"),
            "code": post.get("code"),
            "username": (post.get("user") or {}).get("username"),
            "text": (post.get("caption") or {}).get("text"),
            "taken_at": post.get("taken_at"),
            "like_count": post.get("like_count"),
            "query": query,
            "observed_at_utc": observed_at,
            "request_id": payload.get("request_id"),
        })
    return new_rows


if __name__ == "__main__":
    seen = existing_pks(OUTPUT)
    write_header = not OUTPUT.exists() or OUTPUT.stat().st_size == 0
    with OUTPUT.open("a", newline="", encoding="utf-8-sig") as target:
        writer = csv.DictWriter(target, fieldnames=FIELDS)
        if write_header:
            writer.writeheader()
        for query in WATCHLIST:
            payload, results = read_search(query)
            writer.writerows(rows_for(query, payload, results, seen))

单进程追加不是数据库。并发写入要在字符串帖子键上加唯一约束，否则两台机器同一秒追加，表里会出现重复行。多词监测时，查询与帖子的关联单独存，同一帖命中两次仍是一条帖。中文环境用 utf-8-sig 方便表格软件认编码。审核流程里把新行标成「待看」，相关、无关、嘲讽分列，不要把脚本输出的行数直接写进周报。

搜索不是完整提及库

响应里即使出现 data.cursor，也不表示可以再传一个翻页参数。search-posts 只收 query。list-user-posts 是另一份样本（路径 data.posts[].node.thread_items[].post，入参 username），同样没有翻页输入。账号时间线解决的是「这个号最近能看到什么」，关键词搜索解决的是「这个词这次命中了什么」，两份样本不能拼成一部完整历史。加快采集频率不能证明覆盖，只能更快地重复同一窗口。脚本里 30 秒超时是客户端自己选的，不是服务时延承诺。

按「每个词、每轮」估额度。三个词一轮三次搜索，不要对每一行都打详情。资料为空、用户搜索空数组、view_count 为 null，属于别的操作，不能在这张表里填零；未知浏览量尤其不能拿来算「观看转化」。定时、告警、情感分析自己做。遵守平台条款、合法处理、最小必要和留存期限，公开帖也不是可以无限囤积的档案。

带时间戳、从 thread_items 摊平的 CSV 是样本的工作副本，不是全部公开提及的档案。从 Threads API 文档接入。