首页 > 脚本专栏 > python > Python调用Search Console API

Python调用Search Console API完整指南

2026-10-23 09:26:43 作者：SEO_juper

本文详解如何使用Python调用Google Search Console API,实现多国家数据批量拉取、自动分页处理(突破25,000行限制)、环比分析与阈值告警,涵盖服务账号配置、资源类型区分、RE2正则筛选多国家、匿名查询影响规避等关键技术点,需要的朋友可以参考下

本文为原创技术笔记，代码在 Python 3.11 + google-api-python-client 2.x 环境下实测通过。所有配额与限制数据均引自 Google 官方文档，文内已标注来源链接，可自行核对。

一、为什么要上 API

在 Search Console 后台点鼠标能解决 80% 的问题，但下面这几种场景，后台基本无解：

场景	后台的限制	API 能做到
导出一个站的全部查询词	界面单次最多 1,000 行	每请求最多 25,000 行，且可用 `startRow` 分页拉全量
长期存档	只保留最近 16 个月数据，过期不可恢复	定期落库，自己保留任意时长
多市场横向对比	需要手动反复切换筛选条件	一次脚本循环跑完，输出一张合并表
异常自动告警	只能人肉看	脚本跑完直接输出下跌市场清单

几个必须知道的硬限制（引自 Google 官方文档）：

数据保留窗口 16 个月：超过 16 个月的日期区间返回空结果，界面和 API 一致，且没有历史补数据的接口。
单次请求最多 25,000 行：界面是 1,000 行，API 是 25,000 行，超出的部分用 startRow 翻页。
数据延迟 2–3 天：今天跑脚本，最新能拿到的通常是 2 到 3 天前的数据。
匿名查询会被剔除：出于隐私考虑，搜索量极低的查询不会出现在按 query 聚合的结果里，但它们的展示量仍计入总数——这就是为什么 API 拉出来的加总经常比界面少一截。

二、环境准备（4 步，一次配好）

1. 装依赖

pip install google-api-python-client google-auth pandas

Windows 下如果在公司网络里装不动，加国内镜像：

pip install -i https://pypi.tuna.tsinghua.edu.cn/simple google-api-python-client google-auth pandas

2. 建 Google Cloud 项目并启用 API

进入 Google Cloud 控制台 → 新建项目 → API 和服务 → 库 → 搜索 “Google Search Console API” → 启用。

3. 建服务账号并下载密钥

IAM 和管理 → 服务账号 → 创建服务账号，名字随意，创建完后进入该账号 → 密钥 → 添加密钥 → 创建新密钥 → JSON，下载得到一个 service-account.json。

这个 JSON 等同于账号密码，不要提交到 Git。加到 .gitignore 里。

4. 把服务账号加进 Search Console 站点

打开密钥 JSON，找到 client_email，形如：

gsc-reader@your-project-123456.iam.gserviceaccount.com

进 Search Console → 选站点 → 设置 → 用户和权限 → 添加用户，把这个邮箱加进去，权限给受限用户即可（只读场景不需要完全权限）。

这一步最容易漏。漏了的表现是脚本报 403 Forbidden，而不是认证失败。

三、最小可运行示例

先确认通路通了，再上复杂逻辑。

from google.oauth2 import service_account
from googleapiclient.discovery import build
SCOPES = ["https://www.googleapis.com/auth/webmasters.readonly"]
KEY_FILE = "service-account.json"
SITE_URL = "https://example.com/"   # 必须和 GSC 里的资源类型完全一致
creds = service_account.Credentials.from_service_account_file(KEY_FILE, scopes=SCOPES)
service = build("searchconsole", "v1", credentials=creds)
body = {
    "startDate": "2026-09-01",
    "endDate": "2026-09-28",
    "dimensions": ["query"],
    "rowLimit": 10,
}
resp = service.searchanalytics().query(siteUrl=SITE_URL, body=body).execute()
for row in resp.get("rows", []):
    q, clicks, impr, ctr, pos = row["keys"][0], row["clicks"], row["impressions"], row["ctr"], row["position"]
    print(f"{q[:40]:<40} 点击={clicks:<6} 展示={impr:<8} CTR={ctr:.2%} 均位={pos:.1f}")

三个新手必踩的点：

SITE_URL 的格式。域名资源（Domain property）写成 sc-domain:example.com，网址资源（URL-prefix property）写成 https://example.com/，末尾的斜杠不能少。写错会返回 404 而不是 403。
API 名。build("searchconsole", "v1", ...) 是现行写法；老教程里的 build("webmasters", "v3", ...) 也能用，但建议用新的。
keys 是列表。维度的顺序和你传的 dimensions 一一对应，dimensions=["country","date"] 时 keys[0] 是国家、keys[1] 是日期。

四、完整脚本：多市场分组 + 分页 + 环比告警

下面这个脚本干四件事：

按你定义的市场分组（如 DACH = 德国 + 奥地利 + 瑞士）分别拉取数据；
自动分页，超过 25,000 行继续翻；
与上一个等长周期做环比，算出涨跌；
把触发阈值的市场输出成 Markdown 告警表，同时落一份 CSV。

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
gsc_market_watch.py
按市场分组拉取 Search Console 数据，做环比并输出异常告警。
用法:
    python gsc_market_watch.py --site https://example.com/ --days 28 --threshold -0.15
"""
import argparse
import sys
import time
from datetime import date, timedelta
import pandas as pd
from google.oauth2 import service_account
from googleapiclient.discovery import build
from googleapiclient.errors import HttpError
SCOPES = ["https://www.googleapis.com/auth/webmasters.readonly"]
KEY_FILE = "service-account.json"
# 国家维度使用 ISO 3166-1 alpha-3 三字母代码
MARKETS = {
    "DACH":      ["deu", "aut", "che"],
    "中东":       ["are", "sau", "qat"],
    "北欧":       ["swe", "nor", "dnk", "fin"],
    "拉美西语区": ["mex", "col", "chl"],
    "英语大市场": ["usa", "gbr", "can", "aus"],
}
# 常用三字母代码对照（写正则时要用到）
COUNTRY_HINT = """
usa=美国  gbr=英国  deu=德国  fra=法国  esp=西班牙  ita=意大利
nld=荷兰  are=阿联酋 sau=沙特 qat=卡塔尔 swe=瑞典   nor=挪威
dnk=丹麦  fin=芬兰  mex=墨西哥 col=哥伦比亚 chl=智利 jpn=日本
kor=韩国  ind=印度  bra=巴西  aus=澳大利亚 can=加拿大
"""
def build_service():
    creds = service_account.Credentials.from_service_account_file(KEY_FILE, scopes=SCOPES)
    return build("searchconsole", "v1", credentials=creds, cache_discovery=False)
def date_window(days: int, lag: int = 3, back: int = 0):
    """返回一个 [start, end] 日期区间。
    lag=3  : 官方数据延迟 2-3 天，默认跳过最近 3 天
    back=1 : 再往前推一个等长周期，用于算环比
    """
    end = date.today() - timedelta(days=lag + back * days)
    start = end - timedelta(days=days - 1)
    return start.isoformat(), end.isoformat()
def fetch_all(service, site_url, start, end, dims, filt=None, page_size=25000):
    """带分页的完整拉取。注意：单次最多 25000 行。"""
    rows, start_row = [], 0
    while True:
        body = {
            "startDate": start,
            "endDate": end,
            "dimensions": dims,
            "rowLimit": page_size,
            "startRow": start_row,
        }
        # 同一 filter group 内是 AND 关系，所以多个国家必须用正则匹配
        if filt:
            body["dimensionFilterGroups"] = [{"groupType": "and", "filters": [filt]}]
        try:
            resp = service.searchanalytics().query(siteUrl=site_url, body=body).execute()
        except HttpError as e:
            print(f"[ERROR] HTTP {e.status_code}: {e.reason}", file=sys.stderr)
            break
        batch = resp.get("rows", [])
        rows.extend(batch)
        print(f"  取回 {len(batch)} 行（累计 {len(rows)}）")
        if len(batch) < page_size:
            break
        start_row += len(batch)
        time.sleep(0.3)   # 轻微限速，避免撞配额
    return rows
def to_df(rows, dims):
    recs = []
    for r in rows:
        rec = dict(zip(dims, r["keys"]))
        rec.update({
            "clicks": r.get("clicks", 0),
            "impressions": r.get("impressions", 0),
            "ctr": r.get("ctr", 0.0),
            "position": r.get("position", 0.0),
        })
        recs.append(rec)
    return pd.DataFrame(recs)
def market_totals(service, site_url, start, end, markets):
    """每个市场组拉一次，返回 {市场名: 展示量/点击}"""
    out = {}
    for name, codes in markets.items():
        filt = {
            "dimension": "country",
            "operator": "includingRegex",           # RE2 语法，默认部分匹配
            "expression": "^(" + "|".join(codes) + ")$",
        }
        rows = fetch_all(service, site_url, start, end, ["country"], filt)
        df = to_df(rows, ["country"])
        out[name] = {
            "clicks": int(df["clicks"].sum()) if not df.empty else 0,
            "impressions": int(df["impressions"].sum()) if not df.empty else 0,
        }
    return out
def country_detail(service, site_url, start, end, codes):
    """逐国拆分——合并数好看不等于每个国家都健康。"""
    filt = {
        "dimension": "country",
        "operator": "includingRegex",
        "expression": "^(" + "|".join(codes) + ")$",
    }
    rows = fetch_all(service, site_url, start, end, ["country"], filt)
    return to_df(rows, ["country"])
def main():
    ap = argparse.ArgumentParser()
    ap.add_argument("--site", required=True, help="例: https://example.com/ 或 sc-domain:example.com")
    ap.add_argument("--days", type=int, default=28, help="对比窗口天数，默认 28")
    ap.add_argument("--threshold", type=float, default=-0.15, help="环比跌幅告警线，默认 -15%")
    ap.add_argument("--min-impr", type=int, default=500, help="低于此展示量不告警，过滤噪音")
    ap.add_argument("--csv", default="gsc_market_report.csv")
    args = ap.parse_args()
    service = build_service()
    cur_s, cur_e = date_window(args.days, back=0)
    prv_s, prv_e = date_window(args.days, back=1)
    print(f"本期: {cur_s} ~ {cur_e}\n上期: {prv_s} ~ {prv_e}\n")
    print("拉取本期数据...")
    cur = market_totals(service, args.site, cur_s, cur_e, MARKETS)
    print("拉取上期数据...")
    prv = market_totals(service, args.site, prv_s, prv_e, MARKETS)
    recs = []
    for name in MARKETS:
        c_imp, p_imp = cur[name]["impressions"], prv[name]["impressions"]
        c_clk, p_clk = cur[name]["clicks"], prv[name]["clicks"]
        delta = (c_imp - p_imp) / p_imp if p_imp else 0.0
        recs.append({
            "市场组": name,
            "本期展示": c_imp, "上期展示": p_imp, "展示环比": round(delta, 4),
            "本期点击": c_clk, "上期点击": p_clk,
            "点击环比": round((c_clk - p_clk) / p_clk, 4) if p_clk else 0.0,
        })
    df = pd.DataFrame(recs)
    df.to_csv(args.csv, index=False, encoding="utf-8-sig")
    print(f"\n已写入 {args.csv}\n")
    print(df.to_string(index=False))
    alerts = df[(df["展示环比"] <= args.threshold) & (df["本期展示"] >= args.min_impr)]
    if alerts.empty:
        print("\n无市场触发告警。")
        return
    print("\n## 告警：以下市场组展示量环比下跌超过阈值\n")
    print("| 市场组 | 上期展示 | 本期展示 | 环比 |")
    print("| --- | --- | --- | --- |")
    for _, r in alerts.iterrows():
        print(f"| {r['市场组']} | {r['上期展示']} | {r['本期展示']} | {r['展示环比']:.1%} |")
    print("\n合并数会互相抵消，下面逐国拆开看：\n")
    for name in alerts["市场组"]:
        d = country_detail(service, args.site, cur_s, cur_e, MARKETS[name])
        if not d.empty:
            print(f"[{name}]")
            print(d.sort_values("impressions", ascending=False).to_string(index=False))
            print()
if __name__ == "__main__":
    main()

运行：

python gsc_market_watch.py --site https://example.com/ --days 28 --threshold -0.15

两个实现细节，值得单独说

① 多国家必须用正则，不能并列多个 filter。

dimensionFilterGroups 里同一个 group 的多个 filter 是 AND 关系。写成 country == deu AND country == aut，结果永远是空集。正确做法是 includingRegex + ^(deu|aut|che)$。RE2 语法，默认部分匹配且不区分大小写，所以要用 ^...$ 锚定，否则 mex 会把 mexico 之外的邻近匹配也带进来。

② 合并视图用于监控，逐国视图用于诊断。

这是整套脚本最想表达的一点。五个国家加起来，A 国涨 20%、B 国跌 14%，合并曲线是一条几乎不动的直线 —— 你从合并数据里永远看不出西班牙已经掉了两个月。所以脚本里告警之后又跑了一次 country_detail()，把组内每个国家单独列出来。

五、数字对不上？先查这三个原因

排查顺序建议按这个来，能省掉大量无效争论：

匿名查询（最常见）。搜索量极低的查询不出现在 query 维度里，但计入总数。所以按 query 聚合的加总 < 界面显示的总量，是正常现象，不是 bug。
数据还没到。最近 2–3 天的数据不完整甚至缺失。脚本里 date_window() 默认跳了 3 天（lag=3），就是为了避开这个坑。
资源类型不一致。同一个域名可能同时存在 Domain property 和 URL-prefix property，两者统计范围不同（前者含所有子域名和协议，后者只含前缀匹配的路径）。用 service.sites().list().execute() 先看一下你有哪几个资源。

六、报错排查速查表

报错	原因	处理
`403 Forbidden`	服务账号邮箱没加进 GSC 站点权限	去 GSC 设置 → 用户和权限里添加 JSON 里的 `client_email`
`404 Not Found`	`siteUrl` 写错	域名资源用 `sc-domain:example.com`，网址资源用 `https://example.com/`，注意末尾斜杠
`400 Invalid dimension`	维度名拼错或组合不支持	合法维度：`query` `page` `country` `device` `date` `searchAppearance`
`429 Quota exceeded`	撞到请求频率上限	官方限制为 1,200 次/分钟（每站点每用户）。脚本里加 `time.sleep()`，或者改成一天一天拉
返回 `rows` 为空	日期超出 16 个月窗口，或该维度在这段时间内确实没数据	先把日期缩到最近 7 天验证通路
`Discovery cache` 警告	`googleapiclient` 的 discovery 缓存旧行为	`build(..., cache_discovery=False)`

七、让它自己跑

Linux / macOS，crontab 每周一早上 8 点：

0 8 * * 1 cd /opt/gsc && /usr/bin/python3 gsc_market_watch.py --site https://example.com/ >> gsc.log 2>&1

Windows，任务计划程序里建一个基本任务，程序填 python.exe，参数填：

C:\opt\gsc\gsc_market_watch.py --site https://example.com/

起始位置填 C:\opt\gsc，否则相对路径找不到 service-account.json。

八、进阶：别让 16 个月的数据白白过期

16 个月窗口是硬性的，过期的数据任何方式都拿不回来。如果这份数据对你有价值，两个选项：

自己落库：脚本跑完写 SQLite / Postgres，按月追加。成本最低，上面的 CSV 换个 to_sql() 就能实现。
BigQuery 批量导出：Search Console 后台可以开启官方的 BigQuery 导出。注意它只能从开启当天往后写，不能回填历史，所以越早开越好。

小结

API 相比界面，核心优势是 25,000 行/请求 和 可自动化；
三个必须记住的限制：16 个月窗口、2–3 天延迟、匿名查询不计入明细；
多国家筛选用 includingRegex，不要用多个并列 filter；
合并看趋势，逐国看问题 —— 这是多市场分析里最容易犯的错。

代码里的 MARKETS 字典按你自己的实际市场改就行，其余部分可直接复用。

以上就是Python调用Search Console API完整指南的详细内容，更多关于Python调用Search Console API的资料请关注脚本之家其它相关文章！