用 Search Console API + Python 做关键词排名监控：服务账号鉴权、多维度拉数与下跌告警

  1.  Home
  2. Text  
    3. Share to：

为什么要用 API 拉 Search Console 数据

做 SEO 的站长几乎每天都会打开 Google Search Console 看效果报告：哪些关键词带来了曝光、点击率是多少、平均排名有没有变化。但界面版有几个硬伤：只能看最近 16 个月、每次最多导出 1000 行、无法做多站点聚合、更不能自动告警。当你想「每天早上自动收到一封邮件，告诉我哪些关键词排名掉了 5 位以上」时，界面就完全无能为力了。

Search Console 提供了一套 REST API，可以程序化获取同样的数据，然后你自己想怎么处理就怎么处理——落库做趋势、生成日报、对接告警机器人、做关键词排名监控看板。这就是本文要做的事：用 Python 调 GSC API，搭一套能自动跑的 SEO 关键词排名监控。

需要说明的是，GSC API 只能读数据，不能提交 sitemap（那是另一套 API）。本文聚焦最常用的搜索分析（Search Analytics）接口。

一、开通 API 与创建服务账号

GSC API 走 Google Cloud 的 OAuth2 或服务账号认证。对无人值守的定时任务，服务账号（Service Account）是最优选择：一次配置，永久可用，不需要交互式登录，也不会像用户 OAuth 那样 refresh token 7 天过期。

步骤：

  1. 打开 Google Cloud Console，新建一个项目（比如 gsc-seo-monitor）。
  2. 进入「API 和服务 → 库」，搜索 Google Search Console API，启用它。
  3. 进入「API 和服务 → 凭据 → 创建凭据 → 服务账号」，命名后创建。
  4. 进入该服务账号 → 密钥 → 添加密钥 → 创建新密钥 → 选 JSON，下载得到 service-account.json。
  5. 复制服务账号邮箱（形如 xxx@项目.iam.gserviceaccount.com）。
  6. 回到 Search Console，选中你的站点 → 设置 → 用户和权限 → 添加用户，把服务账号邮箱加进去，权限选「完全」或「受限」均可，只读数据用「受限」就够。

这一步最关键也最容易漏：服务账号必须在 GSC 里被授予站点访问权限，否则 API 会返回 403「User does not have sufficient permission」。很多人以为拿到 JSON 就能用，结果一直 403 就是卡在这。

二、安装客户端库

用官方的 Google API Python 客户端：

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

如果是 Ubuntu 24.04 这类 PEP 668 环境，用 venv：

python3 -m venv /opt/gsc/venv
/opt/gsc/venv/bin/pip install google-api-python-client google-auth

三、最小可用示例：拉取关键词数据

下面这段代码是整套监控的核心，先跑通它。把它保存为 gsc_fetch.py：

from google.oauth2 import service_account
from googleapiclient.discovery import build
from datetime import date, timedelta

KEY_FILE = "/opt/gsc/service-account.json"
SITE_URL = "https://www.example.com/"   # 注意末尾斜杠，要与 GSC 属性完全一致

SCOPES = ["https://www.googleapis.com/auth/webmasters.readonly"]

creds = service_account.Credentials.from_service_account_file(KEY_FILE, scopes=SCOPES)
service = build("searchconsole", "v1", credentials=creds)

end = date.today() - timedelta(days=3)   # GSC 数据有 2-3 天延迟
start = end - timedelta(days=7)

body = {
    "startDate": start.isoformat(),
    "endDate": end.isoformat(),
    "dimensions": ["query"],
    "rowLimit": 25000,
    "dataState": "final",
}
resp = service.searchanalytics().query(siteUrl=SITE_URL, body=body).execute()
rows = resp.get("rows", [])
print(f"拿到 {len(rows)} 个关键词")
for r in rows[:10]:
    print(f"{r['keys'][0][:30]:32} 点击={r['clicks']:5} 曝光={r['impressions']:7} "
          f"CTR={r['ctr']*100:5.2f}% 排名={r['position']:.1f}")

几个必须注意的点：

  • SITE_URL 必须与 GSC 中的属性完全一致，包括协议和末尾斜杠。URL 前缀属性是 https://www.example.com/，域名属性是 sc-domain:example.com。不一致会报 403 或空数据。
  • 数据有 2-3 天延迟，所以 endDate 取今天减 3 天。用今天做 endDate 会拿到不完整数据，且 dataState 可能是 all（未定型）。
  • dataState: "final" 表示只取已定型数据，趋势分析更稳定。

四、拉取多维度数据

dimensions 决定按什么维度聚合，常用的组合：

# 按页面维度：看每个 URL 的表现
body["dimensions"] = ["page"]

# 按查询+页面：定位某关键词的具体落地页
body["dimensions"] = ["query", "page"]

# 按日期：做趋势曲线
body["dimensions"] = ["date"]

# 按设备：移动端 vs 桌面端差异
body["dimensions"] = ["query", "device"]

# 按国家：地域表现
body["dimensions"] = ["query", "country"]

注意 API 有配额限制：每个站点每天 25000 行、每分钟 1200 次请求。如果你要拉多个维度，别一次性全拉，分批并加 rowLimit。分页用 startRow 参数翻页：

def fetch_all(service, site_url, body):
    rows, start = [], 0
    while True:
        body["startRow"] = start
        resp = service.searchanalytics().query(siteUrl=site_url, body=body).execute()
        batch = resp.get("rows", [])
        if not batch:
            break
        rows.extend(batch)
        start += len(batch)
        if len(batch) < 25000:   # 不足一页说明到底了
            break
    return rows

五、排名下跌告警

监控的核心价值是「发现变化」。做法是每天拉一次关键词→平均排名快照，存进 SQLite，第二天对比，把掉位超过阈值的挑出来。

import sqlite3
from datetime import date

def save_snapshot(rows, db="/opt/gsc/gsc.db"):
    conn = sqlite3.connect(db)
    conn.execute("""CREATE TABLE IF NOT EXISTS rank_history(
        snap_date TEXT, query TEXT, page TEXT,
        clicks INTEGER, impressions INTEGER, ctr REAL, position REAL,
        PRIMARY KEY(snap_date, query, page))""")
    today = date.today().isoformat()
    for r in rows:
        if len(r["keys"]) < 2:
            continue
        conn.execute(
            "INSERT OR REPLACE INTO rank_history VALUES(?,?,?,?,?,?,?)",
            (today, r["keys"][0], r["keys"][1], r["clicks"],
             r["impressions"], r["ctr"], r["position"]))
    conn.commit()
    conn.close()

def detect_drops(db="/opt/gsc/gsc.db", min_drop=5.0):
    conn = sqlite3.connect(db)
    cur = conn.cursor()
    dates = [d[0] for d in cur.execute(
        "SELECT DISTINCT snap_date FROM rank_history ORDER BY snap_date DESC LIMIT 2")]
    if len(dates) < 2:
        return []
    today, yday = dates[0], dates[1]
    q = """SELECT t.query, t.page, y.position, t.position
           FROM rank_history t JOIN rank_history y
             ON t.query=y.query AND t.page=y.page
           WHERE t.snap_date=? AND y.snap_date=?
             AND t.position - y.position > ?"""
    drops = cur.execute(q, (today, yday, min_drop)).fetchall()
    conn.close()
    return drops   # [(query, page, 昨日排名, 今日排名), ...]

注意 position 是「平均排名」，不是精确排名，数值越小越好，所以 今日 position - 昨日 position > 5 表示掉了 5 位以上。另外要过滤掉曝光量极低的关键词（比如曝光 < 10），因为平均值波动大，容易误报。加一层过滤：

WHERE t.impressions >= 10 AND t.position - y.position > ?

六、生成日报并推送

把结果格式化成一段文本，通过 Telegram Bot、钉钉 webhook 或邮件发出去。以 Telegram 为例：

import urllib.request, json

def notify(text, token, chat_id):
    url = f"https://api.telegram.org/bot{token}/sendMessage"
    data = json.dumps({"chat_id": chat_id, "text": text, "parse_mode": "HTML"}).encode()
    req = urllib.request.Request(url, data=data,
        headers={"Content-Type": "application/json"})
    urllib.request.urlopen(req, timeout=30)

def build_report(drops, rows):
    lines = ["
Last modification：October 13th, 2026 at 10:26 pm
© 允许规范转载

相关文章

Leave a Comment