"""Search Console API (Search Analytics) からデータを取得して SQLite に upsert。
- クエリ別(date, query)、ページ別(date, page)、ページ×クエリ別(date, page, query) を取得。
- 鮮度ルール: 終端 today-lag_days、そこから rolling 日ぶん遡る（SC は数日遅れで確定する）。
- page は URL 正規化して、GA4 の page_metrics と JOIN できるようにする。
"""
from __future__ import annotations
import logging
import sqlite3
from datetime import date
from typing import Any
from googleapiclient.discovery import build
import common
ROW_LIMIT = 25000
def build_service(credentials):
"""OAuth / サービスアカウントいずれのクレデンシャルでも Search Console サービスを構築する。"""
# cache_discovery=False: ファイルキャッシュ警告を抑止（無人実行向け）
return build("searchconsole", "v1", credentials=credentials, cache_discovery=False)
def _query(service, site_url: str, body: dict) -> list[dict]:
"""startRow でページングして全行取得する（1回あたりの行数上限による欠落を防ぐ）。"""
page_size = body.get("rowLimit", ROW_LIMIT)
rows: list[dict] = []
start = 0
while True:
page_body = dict(body, startRow=start, rowLimit=page_size)
resp = service.searchanalytics().query(siteUrl=site_url, body=page_body).execute()
batch = resp.get("rows", [])
rows.extend(batch)
if len(batch) < page_size: # 最終ページ
break
start += page_size
return rows
def _write_query_rows(conn, site, rows) -> int:
out = []
for r in rows:
keys = r.get("keys", [])
out.append([
site, keys[0], keys[1],
common.to_int(r.get("clicks")),
common.to_int(r.get("impressions")),
common.to_float(r.get("ctr")),
common.to_float(r.get("position")),
])
return common.upsert(
conn, "gsc_query_metrics",
["site", "date", "query", "clicks", "impressions", "ctr", "position"],
out,
)
# 注意: 検索結果のジャンプリンク経由だと、アンカー付きURL（.../page#section）が
# 本体とは別の行で返ることがある。normalize_path でフラグメントを除去すると同じキーに
# 畳まれるが、upsert は INSERT OR REPLACE（後勝ち）なので、そのまま書くと片方の
# クリック・表示回数が消えてしまう。
# → 正規化キー単位で合算してから upsert する。
# clicks/impressions=合算、position=表示回数の加重平均、ctr=clicks/impressions で再計算。
def _aggregate(rows, key_fn):
"""APIの行を key_fn(keys) 単位に合算し、(key, clicks, impressions, ctr, position) を返す。"""
agg: dict = {}
for r in rows:
k = key_fn(r.get("keys", []))
clk = common.to_int(r.get("clicks"))
imp = common.to_int(r.get("impressions"))
rec = agg.setdefault(k, {"clicks": 0, "impressions": 0, "pos_w": 0.0})
rec["clicks"] += clk
rec["impressions"] += imp
rec["pos_w"] += common.to_float(r.get("position")) * imp
return [
(k, rec["clicks"], rec["impressions"],
(rec["clicks"] / rec["impressions"]) if rec["impressions"] else 0.0,
(rec["pos_w"] / rec["impressions"]) if rec["impressions"] else 0.0)
for k, rec in agg.items()
]
def _write_page_rows(conn, site, domain, rows) -> int:
out = [
[site, date_, path, clk, imp, ctr, pos]
for (date_, path), clk, imp, ctr, pos in _aggregate(
rows, lambda keys: (keys[0], common.normalize_path(keys[1], domain))
)
]
return common.upsert(
conn, "gsc_page_metrics",
["site", "date", "page_path", "clicks", "impressions", "ctr", "position"],
out,
)
def _write_page_query_rows(conn, site, domain, rows) -> int:
"""keys = [date, page, query]。page_path は正規化する（GA4 と JOIN 可能）。"""
out = [
[site, date_, path, query, clk, imp, ctr, pos]
for (date_, path, query), clk, imp, ctr, pos in _aggregate(
rows, lambda keys: (keys[0], common.normalize_path(keys[1], domain), keys[2])
)
]
return common.upsert(
conn, "gsc_page_query_metrics",
["site", "date", "page_path", "query",
"clicks", "impressions", "ctr", "position"],
out,
)
def fetch(
conn: sqlite3.Connection,
service: Any,
site_cfg: dict,
cfg: dict,
today: date,
logger: logging.Logger,
) -> int:
"""1サイト分のSCデータを取得→upsert。書き込んだ総行数を返す。失敗時は例外送出。"""
site = site_cfg["site"]
domain = site_cfg["domain"]
site_url = site_cfg["gsc_site_url"]
start, end = common.gsc_date_range(
cfg["gsc_lag_days"], cfg["page_rolling_days"], today
)
max_retries = cfg.get("retry_max", 3)
base_delay = cfg.get("retry_base_delay", 2.0)
total = 0
def body(dims):
return {
"startDate": start,
"endDate": end,
"dimensions": dims,
"rowLimit": ROW_LIMIT,
}
# 再取得対象期間は delete→insert（SC が返さなくなった行を取り除く。GA4 日次系と同方式）。
# 注意: delete は各テーブルの API 取得が成功した「後」に行う。取得失敗時に、
# 呼び出し側の失敗記録 commit に削除だけが巻き込まれて既存データが消えるのを防ぐため。
# クエリ別
q_rows = common.with_retry(
_query, service, site_url, body(["date", "query"]),
max_retries=max_retries, base_delay=base_delay, logger=logger,
)
common.delete_date_range(conn, "gsc_query_metrics", site, start, end)
n = _write_query_rows(conn, site, q_rows)
total += n
logger.info("[%s][gsc] gsc_query_metrics: %d 行 (%s〜%s)", site, n, start, end)
# ページ別
p_rows = common.with_retry(
_query, service, site_url, body(["date", "page"]),
max_retries=max_retries, base_delay=base_delay, logger=logger,
)
common.delete_date_range(conn, "gsc_page_metrics", site, start, end)
n = _write_page_rows(conn, site, domain, p_rows)
total += n
logger.info("[%s][gsc] gsc_page_metrics: %d 行 (%s〜%s)", site, n, start, end)
# ページ×クエリ別（ページごとの流入キーワード）
pq_rows = common.with_retry(
_query, service, site_url, body(["date", "page", "query"]),
max_retries=max_retries, base_delay=base_delay, logger=logger,
)
common.delete_date_range(conn, "gsc_page_query_metrics", site, start, end)
n = _write_page_query_rows(conn, site, domain, pq_rows)
total += n
logger.info("[%s][gsc] gsc_page_query_metrics: %d 行 (%s〜%s)", site, n, start, end)
conn.commit()
return total
MIT ライセンスで配布しています。