自动化排名跟踪看起来就是一个套着 API 调用的 cron 任务,第一周确实如此。问题会在后面出现:两次运行因为发出时间不同而无法比较,系列数据中缺失的某一天没人注意到直到图表看起来不对劲,一个基于噪声触发的警报,还有客户问为什么你的数字和他们的不一致。
这些都不是抓取问题。它们是流水线问题,值得在你积累六个月带漏洞的数据之前就设计好。以下是构建方法。
先确定测量契约
在写任何代码之前,先固定那些让两次测量可比较的变量,因为没有它们,排名毫无意义:位置、设备、语言,以及搜索是否个性化。这些选择应该成为你的任务中的常量,而不是每次运行的可选项。
原因是,这些变量中的任何漂移都会在你的数据中表现为从未真正发生的排名变动。周一从一个城市测量的关键词和周二从另一个城市测量的,看起来会像是移动了。完整的推理见测量准确关键词排名,简而言之:一致性比绝对精度更重要。
把契约写成一个 schema,因为这也是你在客户问及你的数字含义时要交给他们的东西:
# one row per keyword per market: this is the unit of measurement
TARGET = {
"keyword": "residential proxies",
"country": "de",
"city": "berlin", # optional, only where local intent matters
"language": "de",
"device": "desktop",
}通过 SERP API 获取数据
你可以自己在代理上搭建采集层,或者调用一个返回已解析结果的 API。对于每日监控任务,API 路径消除了两个真正消耗时间的维护负担:跟上结果页面布局变化,以及运行采集基础设施。
一个 SERP API 接收上述参数并返回结构化结果,因此你的任务变成了一次请求和一次写入,而不是抓取和解析:
import requests, datetime as dt
def fetch_serp(t):
r = requests.get("https://serp.shifter.io/v1", timeout=45, params={
"api_key": API_KEY,
"q": t["keyword"],
"gl": t["country"], # market
"hl": t["language"], # interface language
"location": t.get("city"),
"device": t["device"],
})
r.raise_for_status()
return r.json()请对照 API 文档核实当前的参数名称,而不是相信某篇博文,因为这些会不断演变。另一种方案,即自己在住宅出口上运行采集,详见为什么准确的排名跟踪需要住宅代理,其中的权衡是控制力与维护成本的取舍。
存储完整结果,而不仅仅是你的位置
最常见的设计错误,也是最难事后弥补的一个,就是每个关键词每天只存储一个数字。
要存储完整的排名 URL 列表、结果功能的存在与位置,以及原始的响应数据。理由有三点。你的竞争对手的变动是让你自身变动变得可解释的背景信息。功能变化,例如页面顶部出现的 AI 概述,会改变一个位置的价值,却不会改变数字本身。而你无法回溯采集上周二的 SERP,所以任何你没有存储的东西都已经永远消失了。
CREATE TABLE serp_snapshot (
id BIGSERIAL PRIMARY KEY,
keyword TEXT NOT NULL,
country TEXT NOT NULL,
city TEXT,
device TEXT NOT NULL,
captured_at TIMESTAMPTZ NOT NULL,
results JSONB NOT NULL, -- full ranked list
features JSONB NOT NULL, -- ai overview, local pack, shopping
raw JSONB -- keep it, storage is cheaper than regret
);
CREATE INDEX ON serp_snapshot (keyword, country, device, captured_at DESC);这样一来,位置就是从快照中推导出来的,而不是作为主要记录存储的,这意味着一次解析修复可以重新应用于历史数据,而不只是应用于未来的运行。
为可比性安排调度
每天在同一时间运行,保持在一个稳定的时间窗口内,因为结果会随一天中的时段变化,而在不同时段采集的系列数据会带有你无法归因的方差。
在整个窗口内分散工作,而不是一次性发出所有关键词请求。突发请求对数据源更不友好,也更容易被限流,而在截止时间还有几个小时的情况下,分散请求不会有任何代价。在请求之间加入抖动,限制并发数,让一次运行花上一个小时而不是四分钟。
import random, time
def run_daily(targets, window_seconds=3600):
gap = window_seconds / max(len(targets), 1)
for t in targets:
snapshot = fetch_with_retries(t)
store(t, snapshot)
time.sleep(gap * random.uniform(0.7, 1.3)) # spread, with jitter区分”无变化”和”无数据”
这是区分一个可信的系列数据和一个误导性数据的关键细节。如果一次抓取失败而你什么都没写入,后面的读者会看到一个空缺,它看起来和什么都没变动的那一天完全一样。然后与”昨天”的比较就会悄无声息地跨越了两天,并报告实际上没有发生过的变动。
记录每一次尝试及其明确的结果,并让下游分析在遇到空缺时拒绝进行跨期比较:
def fetch_with_retries(t, attempts=3):
for i in range(attempts):
try:
data = fetch_serp(t)
if is_valid(data): # sanity-check the payload
return {"status": "ok", "data": data}
record_status(t, "invalid") # parsed, but not a real SERP
except requests.RequestException:
record_status(t, "error")
time.sleep(2 ** i * random.uniform(0.5, 1.5))
return {"status": "failed", "data": None} # written as a failure, not skipped验证响应数据和捕获异常同样重要,因为一个响应可能成功到达,却仍然不是一个可用的结果页面,这在检测被屏蔽或虚假的内容中是一个普遍存在的问题。
对有意义的变动发出警报
一个对任何位置变化都天真触发的警报会不断响起,而一个没人信任的警报系统比没有警报系统更糟。
三条规则能让警报变得有用。要求一个与位置成比例的阈值,因为从第 3 位移动到第 6 位远比从第 47 位移动到第 50 位重要得多。要求持续性,即连续两到三次运行,因为单日的波动是正常的。以及单独对进入或离开首页发出警报,因为这个界限有真实的流量影响。
然后加上人们容易忘记的那一种警报:采集健康状况本身。如果某个市场在三天前就悄无声息地停止返回数据,这比任何排名变化都更紧迫,而这只有在你把运行结果当作一等数据来跟踪时才能看到,参见监控网页抓取流水线。
让数据自我说明
两项补充能把一个排名表变成可以放在客户面前的东西。
标注你自己的时间线:部署、内容发布、迁移。一半的”为什么我们下降了”的调查都会以那一周发布的某个版本而告终,而一个带标注的图表能立即找到它。
计算整体组合层面的变动,而不仅仅是每个关键词的位置,因为许多不相关关键词的广泛变动是与某一页面单独下滑不同的事件。这就是检测 SERP 波动性和算法更新中的波动性分析,而这只有在你存储了完整结果集而非仅仅自己位置的情况下才能做到。
负责任地采集
坚持公开的搜索结果,遵守每个搜索引擎的服务条款,礼貌地控制速率,而不是把速率限制当作障碍。每日监控任务没有理由采取激进的方式:截止时间是次日早上,所以分散工作不会有任何代价。
结论
cron 任务是容易的部分。先确定你的测量契约,让位置、设备、语言和个性化设置成为常量,而不是意外的变量。存储完整的结果集和原始响应数据,而不仅仅是你的位置,因为竞争对手的背景信息和结果功能才是让一个数字变得可解释的东西,而你事后无法回去补采集它们。在一个一致的时间运行,在窗口内带抖动地分散请求。明确记录失败,这样一个空缺就永远不会被误认为是稳定不变。基于持续性、位置加权的变动以及采集健康状况发出警报。然后标注你自己的变更,让图表回答人们真正想问的问题。
采集层可以是一个 SERP API,如果你想要已解析的结果而不必维护采集系统,也可以是带有国家和城市定位的住宅代理,如果你更愿意自己运行,配合适合小规模、高频检查的按 GB 计价方案。