你有没有发现同样打开 X 的热搜榜你看到的内容和日本、巴西、德国当地用户看到的内容完全不一样这跟账号没关系纯粹是平台按地区聚合了不同的话题。前段时间我给一个做海外市场调研的团队写采集程序核心需求就是把多个国家的 Trends 热搜数据定时拉回来整理成结构化数据给后面的选品和舆情分析用。于是就有了这个项目用 Python 配合代理IP把 X 上不同国家的热搜抓下来。这套方案整体不复杂但涉及到的点不少——目标地区接口的定位、请求的伪装、多国 IP 轮换、数据结构化、异常兜底。文章会从思路拆解讲到完整代码再把我实际踩过的坑一并交代清楚。适合刚开始接触爬虫的朋友、做数据分析的同学以及所有需要“知道某个国家此刻正在聊什么”的运营和调研人员。1. 项目整体设计与思路拆解1.1 热搜数据为什么必须“用当地的眼睛”去看先说清楚一个现象X 的 Trends 热搜榜是按地区切分的。平台会根据你的出口IP、语言偏好、账号区域设置等因素决定给你展示哪个版本的热搜页。我刚开始做这个需求的时候第一个想法是“直接抓某个 API 是不是就能拿全球数据”后来发现自己想得太简单了。如果你的请求从一台固定服务器发出去那么不管你怎么换参数拿到的都只是这台服务器所在地区看到的热搜。要做“日本正在聊什么”这种数据就必须让目标服务器认为请求来自日本。这就解释了为什么这个项目里代理IP不是可选项而是刚需——它本质上是一个“地理定位”工具帮你把自己的出口节点放到目标国家去。整件事拆开其实就是三步找到 X 前端在用的热搜数据接口结构化的 JSON而不是去解析 HTML让请求看起来像一个正常用户请求头、token、频率控制让出口 IP 落在目标国家按国家轮换代理IP后面所有代码都是围绕这三件事展开的。先把思路理清楚再动手写代码会顺畅很多。1.2 技术选型这个量级的采集真的不用上重框架这个项目一开始就有个很明确的规模预期几十个国家每小时跑一轮存储量不大对实时性要求也不高。所以我没有考虑 Scrapy 这种重型框架直接用 requests 就够了。如果你后续要做的不是“每天几十个国家”而是“成千上万个用户的时间线数据”那再考虑 Scrapy 或 httpx 的异步方案。那种场景下请求量大到需要并发requests 同步阻塞的写法会很吃力。但就采 Trends 而言requests 的简单直接反而是最大的优势代码短、容易调试、出问题一眼就能定位。方案优点缺点适用场景requests上手快、代码直观、文档多同步阻塞并发能力弱小规模、低频采集httpx支持 HTTP/2、可异步需要理解 async/await有一定并发需求的接口采集Scrapy自带调度、中间件、去重学习成本高写起来重大型分布式爬虫项目解析方面X 的热搜接口直接返回 JSON不需要 BeautifulSoup 或 lxml 去解析 HTML。只需要拿到数据后做字段提取和清洗最后落成 CSV 文件。我选择 CSV 起步的原因是可以直接用 Excel 打开看结果交给非技术的同事也方便。等数据量积累到一定程度再迁到 SQLite 也不迟。2. 前置准备代理IP和反爬机制的关键认知2.1 代理IP怎么选数据中心还是住宅代理这个项目里代理IP的质量直接决定成功率所以我单独拉一节来讲。市面上主流的两类代理IP是数据中心代理和住宅代理价格和效果差距都不小。数据中心代理来自云服务商的 IP 段特点是便宜、速度快但缺点是这些 IP 段很多已经被平台标记过容易触发验证码或者直接被拒。住宅代理则是真实家庭宽带用户的出口 IP绕过程度好很多但价格也贵一个量级。对这个项目来说我会优先建议住宅代理。因为热搜采集本身量就不大选便宜的数据中心代理省下来的钱有限但被限制导致反复重试的成本反而更烦人。属性数据中心代理住宅代理价格低高速度快中等被检测概率较高低适合场景大规模通用采集对 IP 质量要求高的目标另外一定要选支持“粘性会话”的代理IP服务。粘性会话的意思是你拿到的一个 IP 可以在几分钟到几十分钟内保持不变。这个特性对 Trends 采集很重要因为一次性任务需要连续请求几次接口如果每次请求 IP 都在跳反而容易让平台觉得异常。免费 IP 我不建议碰。你想象一下一个免费 IP 池可能有几千人同时在用很可能已经因为某些人的恶意行为被平台拉黑了你拿到手就是黑名单 IP浪费排查时间。这个项目本身需要的 IP 数量不多按国家轮换一次几十个花点小钱买可靠的商业服务是最划算的决策。2.2 X平台的防采集机制先摸清对方的牌写采集代码之前得先知道对面防守是什么水平。X 的防采集不算特别极端但也不是毫无防备。最容易踩到的几个机制是这样第一guest token 机制。X 的很多前端接口并不会要求你登录但要校验一个 guest token相当于一个临时游客凭证。请求头里不带这个 token接口直接 403。token 不是永久的过一段时间会失效所以采集程序要能检测到 token 失效并自动重新获取。第二请求频率限制。同一个 IP 在短时间内高频请求会触发限流。X 的限流不像某些平台那样直接封 IP更多是让你看到请求变慢、返回异常数据或者弹验证码。应对办法很朴素控制采集频率每个国家之间加随机延时伪装成人工操作节奏。第三请求头校验。X 对 User-Agent、Accept、Authorization 这些请求头有校验。特别是 Authorization 里的 Bearer token属于固定值但缺了它接口直接拒绝。实际操作时最好把浏览器里能看到的常用请求头都带上不要求一模一样但至少别让请求头看起来太寒酸。这些机制说白了都不算黑科技核心思路就一句话别太像一个程序。后面我会把请求头配置和延时策略写进代码里。3. 手把手实现从单国采集到多国批量3.1 先搭好基础类Session、代理、请求头初始化代码实现上我建议用一个模块化的小脚本而不是把所有逻辑堆在一起。我习惯先把 Session 的构建和请求头的初始化放在一起因为这是所有请求的公共部分。import requests import time import csv import json import random from datetime import datetime class XTrendsCollector: def __init__(self, proxy_addrNone): self.session requests.Session() if proxy_addr: self.set_proxy(proxy_addr) self.session.headers { authority: x.com, accept: */*, accept-language: en-US,en;q0.9, authorization: Bearer AAAAAAAAAAAAAAAAAAAAANRILgAAAAAAnNwIzUejRCOuH5E6I8xnZz4puTs%3D1Zv7ttfk8LF81IUq16cHjhLTvJu4FA33AGWWjCpTnA, content-type: application/json, user-agent: Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36, x-twitter-active-user: yes, x-twitter-client-language: en, } self.guest_token None def set_proxy(self, proxy_addr): self.session.proxies { http: fhttp://{proxy_addr}, https: fhttp://{proxy_addr}, }这里有个项目里的关键决定为什么把代理地址作为参数传进构造函数而不是写死因为多国采集时每个国家要用不同的代理IP采集器对象应该被复用来做“同样配置下的多次请求”而不是每个请求都新建一个对象。想换国家就新建一个带新代理的采集器逻辑清晰也不会互相污染。关于 Authorization 那串 Bearer token我是从浏览器开发者工具里看到的它属于前端公开的固定值不是用户私密凭证。不过这种固定值平台随时可能换写代码时要注意如果某一天所有请求都返回 401先去检查是不是这个值变了。实际生产环境里更稳妥的方式是启动时先从页面里动态提取而不是硬编码在代码里。3.2 核心拿到guest_token并采集指定国家的Trends初始化完成后第一件事是激活 guest token。这一步是通过调用 X 的一个接口来完成的。拿到 token 后放到请求头里后续的 Trends 请求才会被正常处理。def activate_guest_token(self): url https://api.x.com/1.1/guest/activate.json try: resp self.session.post(url, timeout15) if resp.status_code 200: self.guest_token resp.json().get(guest_token) self.session.headers[x-guest-token] self.guest_token return True else: print(f[ERR] activate guest token failed: {resp.status_code} {resp.text[:200]}) return False except requests.RequestException as e: print(f[ERR] activate guest token exception: {e}) return False有一个小细节值得说一下guest token 不是说获取一次就能永远用。实测中它可能几十分钟到几小时就会失效表现为后续请求返回 401。所以我写了个ensure_guest_token方法每次采集前检查一下 token 是否存在如果请求遇到 401 就重新激活一次这样即使跑几个小时的长任务也不会突然中断。def ensure_guest_token(self): if not self.guest_token: return self.activate_guest_token() return True接下来就是核心的采集方法。X 的 Trends 数据接口路径是https://x.com/i/api/1.1/trends/place.json参数是一个id也就是 WOEIDWhere On Earth IDentifier这个编号对应具体的国家或城市。全球的 WOEID 是 1美国的 WOEID 是 23424977日本是 23424856。不同国家要传不同的编号。def fetch_trends(self, woeid): if not self.ensure_guest_token(): return [] url https://x.com/i/api/1.1/trends/place.json params {id: woeid} try: resp self.session.get(url, paramsparams, timeout15) if resp.status_code 401: print([WARN] guest token expired, re-activating) self.activate_guest_token() resp self.session.get(url, paramsparams, timeout15) if resp.status_code ! 200: print(f[WARN] fetch trends failed: {resp.status_code} {resp.text[:200]}) return [] data resp.json() trends data[0].get(trends, []) result [] for item in trends: result.append({ name: item.get(name), url: item.get(url), tweet_volume: item.get(tweet_volume), category: item.get(category), }) return result except requests.RequestException as e: print(f[ERR] fetch trends exception: {e}) return []接口返回的 JSON 结构大致是最外层是一个列表列表第一个元素里有trends字段里面每一项包含热搜关键词name、对应的搜索链接url、推文量tweet_volume和分类category。其中tweet_volume表示这个话题最近的推文量可以用来衡量话题热度但注意它可能为空代表平台没有公开具体数值。不是让程序直接去请求网页版页面再解析 HTML而是直接复用了前端页面背后的数据接口。这样做的好处是返回的是结构化数据不用去抠 HTML 标签而且字段完整后面做分析省很多事。缺点就是对接口依赖强平台一旦改接口路径代码就要跟着调整。3.3 多国批量采集WOEID映射与轮换策略有了单国采集的接口批量做起来就只是“换参数”的问题了。我维护了一个国家名和 WOEID 映射的字典需要采哪个国家就往这个列表里加一行。国家/地区WOEID全球1美国23424977日本23424856德国23424829法国23424819巴西23424768英国23424975加拿大23424775澳大利亚23424748印度23424848墨西哥23424900批量采集的正确姿势是这样对每个国家使用对应国家的代理 IP创建独立的采集器采集完后把“国家、热搜词、推文量、采集时间”写入 CSV。每个国家之间加一个随机延时模拟人工切换操作。def collect_all_countries(proxies_map): countries [ (global, 1), (us, 23424977), (jp, 23424856), (de, 23424829), (fr, 23424819), (br, 23424768), (gb, 23424975), (ca, 23424775), (au, 23424748), (in, 23424848), (mx, 23424900), ] with open(trends.csv, w, newline, encodingutf-8) as f: writer csv.writer(f) writer.writerow([country, trend_name, tweet_volume, trend_url, collected_at]) for code, woeid in countries: proxy proxies_map.get(code) if not proxy: print(f[SKIP] no proxy for {code}) continue collector XTrendsCollector(proxy_addrproxy) trends collector.fetch_trends(woeid) now datetime.now().strftime(%Y-%m-%d %H:%M:%S) for t in trends: writer.writerow([code, t[name], t[tweet_volume], t[url], now]) print(f[OK] {code} collected {len(trends)} trends) time.sleep(random.uniform(2.0, 5.0))这里有个很多新手容易忽视的问题为什么每个国家要单独换代理 IP因为如果你始终用一个美国 IP即使传了日本的 WOEID平台返回的热搜可能仍然是美国地区的版本或者说至少结果会带有“美国视角”的偏差。要拿到真正反映日本当地讨论热度的榜单出口 IP 和 WOEID 必须同时指向同一个国家两者缺一不可。我实际测试过用美国 IP 抓日本 WOEID返回的话题列表跟日本本地看到的差距非常大基本可以断定平台是综合 IP 和参数共同判断的。延时策略方面我用的是random.uniform(2.0, 5.0)也就是每次采集完一个国家之后随机等 2 到 5 秒再采下一个。这个延时对几十个国家来说总共多花两分钟左右但是对避限流帮助很大。如果你跑了一段时间发现某个 IP 被限制了可以把这个延时加大到 10 秒以上。3.4 集成后的运行效果从命令行到定时任务把所有逻辑整合好以后我最终的运行流程是读代理配置遍历国家列表依次采集写入 CSV。跑完一轮大约需要 3 到 4 分钟视代理响应速度而定。输出结果大致是这样的表格countrytrend_nametweet_volumecollected_atjp芸能人1200002025-01-15 10:02:31jpトレンド850002025-01-15 10:02:31usSuper Bowlnull2025-01-15 10:04:12brCarnaval2300002025-01-15 10:05:47拿到这种数据之后我们团队基本就直接拿去用了——哪个体量级的话题排在前面、哪个话题推文量是空的一目了然。脚本本身不到 200 行依赖只有一个 requests 库环境配置也简单pip install requests就够了。后续我把它挂到调度任务里每天固定时间跑这才算是真正“能用”的采集工具。4. 常见问题与排查技巧实录4.1 问题速查表这个项目写完以后我又调试过好几轮也帮同事处理过类似的问题。把最常见的几个现象、原因和解决办法整理在下面现象可能原因解决办法请求返回 403缺 guest token 或 token 失效重新调用 activate_guest_token确认请求头带上了 x-guest-token返回 401Authorization 或 guest token 异常检查 Bearer token 是否还是当前有效的固定值必要时从浏览器重新提取请求返回 200 但列表为空WOEID 传错或代理 IP 所在地区不匹配核对 WOEID换目标国家的代理 IP 再试超时代理 IP 质量差或网络不可达换一个代理节点设置合理的 timeout 和重试逻辑采集过程中途频繁失败请求频率太高被限流增大随机延时降低采集频率同一个话题反复重复上一次采集数据没清空用带追加模式的 CSV 时注意去重逻辑或者直接用时间戳分文件这里我想特别强调一下“代理 IP 所在地区不匹配”这个坑。很多代理服务商不会告诉你 IP 的实际地理位置属于哪个城市如果你买的是“美国”代理实际出口可能在美国某个州这本身没问题。但如果你要采的是某个具体城市的热搜WOEID 得对应到那个城市且代理 IP 最好也落在同一个国家。跨国家错配是最隐蔽的问题因为它不报错只是数据不对。4.2 我踩过的三个坑第一个坑是 guest token 硬编码失效问题。最开始我把激活 token 的逻辑写成“每次启动只激活一次”然后整个任务跑到底。跑了几十分钟后突然开始大量 401。排查半天才发现 guest token 的有效期并不是永久。后来改成“请求失败时自动重新激活”的机制这个问题就彻底解决了。凡是写长时间运行的采集任务一定要考虑凭证过期的情况不能假设拿一次就能用到天荒地老。第二个坑是代理 IP 的格式问题。代理服务商给的地址可能是host:port也可能是http://host:port还有可能带用户名密码认证。我最初直接把服务商给的裸地址拼到proxies配置里结果每个请求都报代理连接错误。后来才注意到 requests 库要求代理地址必须是完整的 URL 形式。如果是带认证的代理格式是http://username:passwordhost:port这一点在配置文件的解析逻辑里必须处理干净。第三个坑是请求太快导致 IP 被临时限制。第一版代码完全没有延时几十个国家一口气跑完结果跑到第 20 个左右开始批量失败。后来我把轮询延时加上去并且把失败重试的次数和退避策略也写了进去一个代理 IP 连续失败三次就跳过等下一轮再试。这样虽然跑得慢一点但每一轮的数据完整度高很多。5. 从单次采集到持续观察5.1 定时增量采集与数据落库单次采集只是开始热点数据的价值在于趋势变化。我把脚本改成支持“增量追加”后就挂到了调度系统里每隔一小时跑一轮。这样一天下来就有 24 个时间点的热搜快照可以清清楚楚看到话题的涨落过程。增量追加要注意一个去重逻辑不同时间点同一话题可能会反复出现所以 CSV 表里应该以“国家 话题 采集时间”作为记录维度而不是只存话题名。文件命名方面我建议按日期分文件比如trends_20250115.csv避免单个文件无限膨胀。如果跑久了文件太多可以考虑迁移到 SQLite用 SQL 去查特定时间窗口的话题热度变化比打开一堆 CSV 效率高很多。定时任务方面如果你是 Linux 服务器直接用 cron 就行一行命令搞定0 * * * * cd /path/to/project python collect_trends.py run.log 21这个 cron 表达式表示每小时整点运行一次。如果你用的是 Windows可以用计划任务程序或者直接在 Python 里用schedule库做定时循环。我个人偏向用系统级方案因为就算 Python 进程时不时挂掉系统层级的定时任务也会把它拉起来不需要额外守护进程。5.2 后续还能做什么热点对比、关键词热度分析当数据积累到一两周后能做的事情就多起来了。最简单的分析是“跨国家对比”同一个时段哪些话题同时出现在多个国家的热搜里哪些是某个国家独有的。这些独有话题往往反映出当地特有的文化或消费偏好对做区域市场的同学来说价值很高。另外一种玩法是对话题做词频统计和分类。把一段时间内的热搜词收集起来按出现频次排序能摸索出这个国家用户的长期关注点。比如某国热搜榜上长时间高频出现娱乐和体育话题那这个市场的人群画像就跟“追星”和“体育迷”高度相关。配合 twwet_volume 字段还可以算话题的平均热度识别“短暂爆发型”和“持续热门型”话题。这些分析的代码都不难统计词频用 Python 自带的collections.Counter就够了不需要上大数据框架。关键是数据要先采集回来、存储规范化后续才谈得上分析。这个项目里的代理IP策略和采集框架稍微改改也能复用到其他需要“按地区看内容”的平台——思路是通用的热点采集只是其中一个典型应用。