> ## Documentation Index > Fetch the complete documentation index at: https://docs.brightdata.com/llms.txt > Use this file to discover all available pages before exploring further. # 构建关键词社交监听工具 > 使用Bright Data SERP API和社交媒体爬虫构建关键词社交监听工具。发现Instagram、TikTok和X上的帖子，然后收集完整数据。 你想知道什么时候一个关键词（品牌名称、产品、竞争对手）在 Instagram、TikTok 或 X 上的帖子中出现。你事先没有这些 URL，所以不能只是把一个列表交给爬虫。你需要先进行一个发现步骤。 在本教程中，我们将用一个 Python 脚本构建这个发现管道。你将使用 Bright Data 的 **SERP API** 在 `parsed_light` 模式下运行 `site:instagram.com "your keyword"` 风格的查询，从每个响应中获取前 10 个有机链接，按平台对其进行分类，并异步触发匹配的社交爬虫。最后，你将获得每个平台的一个快照，可以交付给 S3 或 webhook。 我们在触发步骤处停止。将快照路由到存储是一个交付配置更改，本系列中的其他两个教程已经涵盖。 ## 你将构建什么 一个 Python 脚本，可以： 1. 为你的关键词运行三个 SERP API 查询，每个平台一个 2. 从每个 `parsed_light` 响应中读取前 10 个有机结果 3. 将每个链接分类为 Instagram 帖子、TikTok 视频或 X 帖子 4. 为每个平台的 URL 列表触发匹配的 Bright Data 社交爬虫 5. 打印每个平台的一个快照 ID 运行一次，获得三个快照 ID。稍后将同一脚本连接到 GitHub Actions，你就有了一个每日提及监控器。 **预计时间**：25 分钟。 ## 前置条件 * 一个 [Bright Data 账户](https://www.bright.cn/?hs_signup=1\&utm_source=docs)，带有 API 密钥（[获取你的密钥](https://www.bright.cn/cp/setting/users)） * 在控制面板中创建的 **SERP API 区域**。按照 [SERP API 介绍](/products/serp-api/introduction)操作一次，然后使用区域名称返回 * 本地安装的 Python 3.9+ ## 第 1 部分：设置项目 创建一个新文件夹并放入 `requirements.txt`： ```bash theme={null} mkdir social-listener && cd social-listener ``` ```text requirements.txt theme={null} requests==2.32.3 ``` 安装依赖： ```bash theme={null} pip install -r requirements.txt ``` 导出脚本将从环境读取的凭证。API 密钥同时认证 SERP API 和社交爬虫触发；区域名称告诉 SERP API 通过哪个 SERP 区域路由请求： ```bash theme={null} export BRIGHTDATA_API_KEY=your_api_key export BRIGHT_DATA_SERP_ZONE=your_serp_zone_name ``` ## 第 2 部分：使用 SERP API 发现提及 SERP API 是一个单一 REST 端点：`POST https://api.brightdata.com/request`。你提供一个区域名称和一个 Google 搜索 URL，它返回该查询的 SERP。传递 `data_format: "parsed_light"`，你会获得一个紧凑的 JSON 有效载荷，只包含前 10 个有机结果，无广告、无知识面板、无需你这边进行解析。 创建 `listen.py` 并从发现步骤开始： ```python listen.py theme={null} from __future__ import annotations import os import sys import urllib.parse import requests SERP_URL = "https://api.brightdata.com/request" TRIGGER_URL = "https://api.brightdata.com/datasets/v3/trigger" PLATFORMS = ["instagram.com", "tiktok.com", "x.com"] def discover(keyword: str, api_key: str, zone: str) -> dict[str, list[dict]]: headers = { "Authorization": f"Bearer {api_key}", "Content-Type": "application/json", } results_by_platform: dict[str, list[dict]] = {} for site in PLATFORMS: query = f'site:{site} "{keyword}"' google_url = ( "https://www.google.com/search" f"?q={urllib.parse.quote_plus(query)}&hl=en&gl=us" ) response = requests.post( SERP_URL, headers=headers, json={ "zone": zone, "url": google_url, "format": "raw", "data_format": "parsed_light", }, timeout=180, ) response.raise_for_status() organic = response.json().get("organic", []) results_by_platform[site] = organic return results_by_platform ``` 有三件事值得注意： * **`data_format: "parsed_light"`** 是关键。它告诉 SERP API 仅返回前 10 个有机结果在一个紧凑的 `organic` 数组中，跳过广告和知识面板。更快的响应，更少的数据需要分类。 * **`format: "raw"`** 在 `data_format` 旁边是必需的。它指的是外部 HTTP 响应信封，而不是解析的有效载荷。SERP API 文档[详细说明了这一点](/products/serp-api/introduction#get-only-top-10-results-faster-response)。 * **`timeout=180`** 的设置很慷慨。大多数 SERP API 调用在不到一秒内返回，但在 Google 速率限制期间，单个查询有时可能需要更长时间。每次三秒的预算使脚本保持稳健，同时又不浪费。 ## 第 3 部分：按平台对 URL 进行分类 每个 SERP 响应给你一个 `organic` 数组，每个条目都有一个 `link` 字段指向真实的帖子 URL。你需要对这些链接进行分类，以便每个爬虫只获得它理解的 URL。 将这些助手添加到 `listen.py`： ```python listen.py theme={null} import re INSTAGRAM_POST_RE = re.compile(r"https://(?:www\.)?instagram\.com/(?:p|reel)/[^/?#]+") TIKTOK_POST_RE = re.compile(r"https://(?:www\.)?tiktok\.com/@[^/]+/video/\d+") X_POST_RE = re.compile(r"https://(?:www\.)?x\.com/[^/]+/status/\d+") def classify(results_by_platform: dict[str, list[dict]]) -> dict[str, list[str]]: buckets: dict[str, list[str]] = {"instagram": [], "tiktok": [], "x": []} for organic in results_by_platform.values(): for result in organic: link = result.get("link", "") if INSTAGRAM_POST_RE.match(link): buckets["instagram"].append(link) elif TIKTOK_POST_RE.match(link): buckets["tiktok"].append(link) elif X_POST_RE.match(link): buckets["x"].append(link) for platform in buckets: buckets[platform] = list(dict.fromkeys(buckets[platform])) return buckets ``` 注意正则表达式仅匹配**帖子形状的 `https://` URL**，而不是个人资料页面或主题页面。像 `instagram.com/tiktok` 这样的个人资料 URL 被拒绝，因为本教程的爬虫仅将帖子 URL 作为输入。`https://` 前缀的严格性是有意的：社交爬虫在验证时拒绝纯 `http://` URL，所以在这里删除它们可以避免稍后从触发步骤获得 `400 validation_error`。`dict.fromkeys` 技巧在保留顺序的同时去重。 ## 第 4 部分：触发社交爬虫 你现在有三个帖子 URL 列表。使用异步触发将它们发送到匹配的社交爬虫，以便每个爬虫可以按自己的计划运行，而不会阻止脚本： ```python listen.py theme={null} SOCIAL_DATASETS = { "instagram": "gd_lk5ns7kz21pck8jpis", # Instagram posts "tiktok": "gd_lu702nij2f790tmv9h", # TikTok posts "x": "gd_lwxkxvnf1cynvib9co", # X posts } def trigger(platform: str, urls: list[str], api_key: str) -> str | None: if not urls: print(f" {platform}: no URLs to trigger") return None response = requests.post( TRIGGER_URL, params={"dataset_id": SOCIAL_DATASETS[platform], "format": "json"}, headers={ "Authorization": f"Bearer {api_key}", "Content-Type": "application/json", }, json=[{"url": url} for url in urls], timeout=30, ) response.raise_for_status() snapshot_id = response.json().get("snapshot_id") print(f" {platform}: {len(urls)} URLs -> snapshot {snapshot_id}") return snapshot_id ``` 有两件事值得注意： * **发现是同步的，爬取是异步的。** SERP API 每个查询在不到一秒内返回，所以你在内联循环遍历三个查询。社交爬虫每个可能需要几分钟，所以你触发并退出。Bright Data 在后台运行它们。 * **脚本返回快照 ID，而不是数据。** 实际帖子会根据你的爬虫交付配置指向的位置（API 下载、webhook 或 S3/GCS/Azure）。有关 [webhook 交付](/products/scrapers/tutorials/linkedin-to-crm)和 [S3 交付](/products/scrapers/tutorials/amazon-price-monitor)，请参阅两个附带教程。 ## 第 5 部分：将其连接在一起并运行 添加一个 `main` 块来粘合这三个步骤： ```python listen.py theme={null} def main() -> int: api_key = os.environ.get("BRIGHTDATA_API_KEY") zone = os.environ.get("BRIGHT_DATA_SERP_ZONE") if not (api_key and zone): print("Set BRIGHTDATA_API_KEY and BRIGHT_DATA_SERP_ZONE before running.") return 1 if len(sys.argv) < 2: print("Usage: python listen.py \"your keyword\"") return 1 keyword = sys.argv[1] print(f"Discovering mentions of: {keyword}") results_by_platform = discover(keyword, api_key, zone) buckets = classify(results_by_platform) print("Classified URLs:") for platform, urls in buckets.items(): print(f" {platform}: {len(urls)}") print("Triggering scrapers:") for platform, urls in buckets.items(): trigger(platform, urls, api_key) return 0 if __name__ == "__main__": sys.exit(main()) ``` 使用可能在所有三个平台上出现的关键词运行它： ```bash theme={null} python listen.py "bright data" ``` 你应该会看到类似的输出： 示例输出。你的值会有所不同。URL 数量取决于你运行脚本当天的搜索结果。 ``` Discovering mentions of: bright data Classified URLs: instagram: 9 tiktok: 10 x: 10 Triggering scrapers: instagram: 9 URLs -> snapshot sd_mnte2pbv16riike0ao tiktok: 10 URLs -> snapshot sd_mnte2q1nw3g36l4z1 x: 10 URLs -> snapshot sd_mnte2qt91ga85uuq8r ``` 三个快照 ID，每个平台一个。每个现在都在 Bright Data 异步运行，并将结果交付到你为该爬虫交付目的地配置的位置。 ## 故障排除 ### 为什么某个平台显示 0 个 URL？ SERP API 搜索没有为该关键词在该平台上返回帖子 URL。第 3 部分的正则表达式只保留帖子形式的链接，因此如果前 10 个搜索结果都是个人资料页或话题页，数量同样为 0。此时脚本会为该平台打印 `no URLs to trigger`，并且不启动作业。请尝试更宽泛的关键词。 ### 如何确认结果已经送达？ 用 [监控进度](/cn/api-reference/scrapers/management-apis/monitor-progress)（`GET /datasets/v3/progress/{snapshot_id}`）轮询脚本打印的每个 `snapshot_id`，直到 `status` 字段变为 `ready`。然后用 [下载 Snapshot](/cn/api-reference/scrapers/delivery-apis/download-snapshot)（`GET /datasets/v3/snapshot/{snapshot_id}`）获取记录。每个平台作为独立作业运行，因此需要分别检查三个快照 ID。 ### 为什么触发调用返回 429？ 该社交爬虫正在运行的作业过多。Scraper API 允许 5,000 个活跃作业或快照，超过该限制的新触发请求会返回 HTTP 429 和 `You have too many running jobs for this dataset`。再次运行脚本前，请等待 `Retry-After` 标头中的秒数。在 5 分钟内收到 25 个或更多 429 响应的 IP 会被 Bright Data 封锁。 所有 Scraper API 错误消息参见 [Scraper API 错误代码](/cn/products/scrapers/error-codes)。作业、负载和下载限制参见 [Scraper API 限制](/cn/products/scrapers/limits)。 ## 祝贺 你构建了一个关键词社交监听器，它： * 使用 `parsed_light` 模式下的 **SERP API** 作为紧凑、亚秒级的发现引擎 * 获取每个查询的**前 10 个有机结果**，并使用帖子形状正则表达式按平台对其进行分类 * **触发** Instagram、TikTok 和 X 爬虫在发现的 URL 上并行运行 * 触发后立即退出，将繁重工作留给 Bright Data 整个管道是一个 Python 文件。无代理设置、无胶合服务、无批处理作业。 ## 后续步骤 将每个社交爬虫指向一个 webhook 处理程序（如 LinkedIn 教程中的 Next.js 处理程序），以实时获取映射的帖子记录。 在每个爬虫上配置 S3 交付，并使用 GitHub Actions 计划此脚本以获取每日提及报告。 可以添加到搜索 URL 的 Google 特定查询参数的完整列表，包括 `gl`、`hl`、`uule` 和时间过滤器。 所有支持的引擎、解析的输出架构和用于大量数据的异步请求模式。