"""過去分バックフィル: 任意サイト・任意期間の GA4 / Search Console データを取り込む。
日次バッチ(main.py)は鮮度ルールで取得窓が狭い（GA4 日次系=直近 ga4_reprocess_days 日 /
GSC=today-lag から page_rolling_days 日）ため、サイトを後から追加したり、バッチが
しばらく止まっていたりすると、その間のデータが DB に入らない。本スクリプトはその穴埋め用。
使い方:
（プロジェクトのルートで実行する）
python -X utf8 scripts/analytics/backfill.py --site site_a --start 2025-01-01
python -X utf8 scripts/analytics/backfill.py --site site_a --start 2025-01-01 --sources gsc
python -X utf8 scripts/analytics/backfill.py --site site_a --start 2025-01-01 --end 2025-02-28
設計方針:
- 日次バッチ側の取得・書込関数をそのまま呼び、日付範囲だけを差し替える
（バックフィル専用の取得ロジックを持たないことで、日次バッチと結果がずれない）。
- 冪等: 日次系・GSC は delete→insert、page_metrics は INSERT OR REPLACE。
同じ期間を2回流しても行は増えない。
- page_metrics は「実行日スナップショット（直近 page_rolling_days 日の合計）」という
性質上、過去日ぶんは1日=1リクエストのループで再現する（--sources から page を外せば
スキップできる）。GA4 のデータは後から再処理されるため、当時実行した場合の値と
完全には一致しない。
- batch_runs には source を `*_backfill` として記録する（日次の行と衝突させない）。
- Search Console API で遡れる期間には上限がある（おおむね過去16か月）。
実行前に DB ファイルをコピーしてバックアップしておくことを推奨。
"""
from __future__ import annotations
import argparse
from datetime import date, datetime, timedelta
from google.analytics.data_v1beta.types import BatchRunReportsRequest
import common
import fetch_ga4
import fetch_gsc
# fetch_ga4.fetch の #2〜#7 と同じ並び（delete→insert の対象）
DAILY_TABLES = [
"channel_metrics", "landing_metrics", "user_type_metrics",
"device_geo_metrics", "hourly_metrics", "event_metrics",
]
MAX_BATCH = 5 # GA4 batchRunReports は1回5レポートまで
def _parse_ymd(s: str) -> date:
return datetime.strptime(s, "%Y-%m-%d").date()
def _now() -> str:
return datetime.now().isoformat(timespec="seconds")
# --------------------------------------------------------------------------- #
# GA4 日次系（#2〜#7）
# --------------------------------------------------------------------------- #
def backfill_ga4_daily(conn, client, site_cfg, cfg, start: date, end: date, logger) -> int:
"""[start, end] を batchRunReports で取り直す（delete→insert）。"""
site = site_cfg["site"]
domain = site_cfg["domain"]
property_id = str(site_cfg["ga4_property_id"])
# common.ga4_date_range(N, today) = (today-N, today-1) なので、
# today=end+1 / N=期間日数 とすれば [start, end] にちょうど一致する。
n_days = (end - start).days + 1
pseudo_today = end + timedelta(days=1)
cfg2 = dict(cfg, ga4_reprocess_days=n_days)
d_start, d_end = common.ga4_date_range(n_days, pseudo_today)
assert (d_start, d_end) == (common.ymd(start), common.ymd(end)), (d_start, d_end)
for t in DAILY_TABLES:
n_del = common.delete_date_range(conn, t, site, d_start, d_end)
if n_del:
logger.info("[%s][backfill] %s: 既存 %d 行を削除", site, t, n_del)
writers = [
("channel_metrics", lambda r: fetch_ga4._write_channel(conn, site, r)),
("landing_metrics", lambda r: fetch_ga4._write_landing(conn, site, domain, r)),
("user_type_metrics", lambda r: fetch_ga4._write_user_type(conn, site, r)),
("device_geo_metrics", lambda r: fetch_ga4._write_device_geo(conn, site, r)),
("hourly_metrics", lambda r: fetch_ga4._write_hourly(conn, site, r)),
("event_metrics", lambda r: fetch_ga4._write_event(conn, site, r)),
]
reqs = fetch_ga4._daily_requests(cfg2, pseudo_today)
total = 0
for i in range(0, len(reqs), MAX_BATCH):
batch_req = BatchRunReportsRequest(
property=f"properties/{property_id}",
requests=reqs[i:i + MAX_BATCH],
)
resp = common.with_retry(
client.batch_run_reports, batch_req,
max_retries=cfg.get("retry_max", 3),
base_delay=cfg.get("retry_base_delay", 2.0),
logger=logger,
)
for (name, writer), report in zip(writers[i:i + MAX_BATCH], resp.reports):
n = writer(report)
total += n
logger.info("[%s][backfill][ga4] %s: %d 行 (%s - %s)", site, name, n, d_start, d_end)
conn.commit()
return total
# --------------------------------------------------------------------------- #
# GA4 page_metrics（実行日スナップショット）
# --------------------------------------------------------------------------- #
def backfill_page_snapshots(conn, client, site_cfg, cfg, start: date, end: date, logger) -> int:
"""snapshot_date=d（窓 = d-page_rolling_days 〜 d-1）を d ごとに再現する。"""
site = site_cfg["site"]
domain = site_cfg["domain"]
property_id = str(site_cfg["ga4_property_id"])
total = 0
days = 0
d = start
while d <= end:
req = fetch_ga4._page_report_request(property_id, cfg, d)
resp = common.with_retry(
client.run_report, req,
max_retries=cfg.get("retry_max", 3),
base_delay=cfg.get("retry_base_delay", 2.0),
logger=logger,
)
n = fetch_ga4._write_page_metrics(conn, site, domain, common.ymd(d), resp)
total += n
days += 1
if days % 10 == 0:
conn.commit()
logger.info("[%s][backfill][ga4] page_metrics: %s まで %d 行",
site, common.ymd(d), total)
d += timedelta(days=1)
conn.commit()
logger.info("[%s][backfill][ga4] page_metrics: 合計 %d 行 / %d スナップショット",
site, total, days)
return total
# --------------------------------------------------------------------------- #
# Search Console
# --------------------------------------------------------------------------- #
def backfill_gsc(conn, service, site_cfg, cfg, start: date, end: date, logger) -> int:
"""fetch_gsc.fetch をそのまま使い、lag/rolling だけ [start, end] になるよう差し替える。"""
today = date.today()
lag = (today - end).days
rolling = (end - start).days + 1
if lag < 0:
raise ValueError(f"end({end}) が未来です")
cfg2 = dict(cfg, gsc_lag_days=lag, page_rolling_days=rolling)
s, e = common.gsc_date_range(lag, rolling, today)
assert (s, e) == (common.ymd(start), common.ymd(end)), (s, e)
return fetch_gsc.fetch(conn, service, site_cfg, cfg2, today, logger)
# --------------------------------------------------------------------------- #
def main() -> int:
p = argparse.ArgumentParser(description="GA4 / Search Console の過去分バックフィル")
p.add_argument("--site", required=True, help="設定ファイルの site 識別子（例: site_a）")
p.add_argument("--start", required=True, help="取得開始日 YYYY-MM-DD")
p.add_argument("--end", default=None,
help="取得終了日 YYYY-MM-DD（既定: GA4=昨日 / GSC=today-gsc_lag_days）")
p.add_argument("--sources", default="ga4,page,gsc,affiliate",
help="カンマ区切り: ga4(日次系) / page(page_metrics) / gsc / "
"affiliate(ga4_affiliate_event を設定したサイトのみ)")
p.add_argument("--config", default=None, help="設定ファイルのパス（既定: プロジェクトのルートの config.yaml。読むのは analytics: 節）")
args = p.parse_args()
cfg = common.load_config(args.config)
logger = common.setup_logging(cfg["log_file"])
sources = {s.strip() for s in args.sources.split(",") if s.strip()}
site_cfg = next((s for s in cfg.get("sites", []) or [] if s["site"] == args.site), None)
if site_cfg is None:
logger.error("site=%s が設定ファイルにありません", args.site)
return 2
today = date.today()
start = _parse_ymd(args.start)
end_arg = _parse_ymd(args.end) if args.end else None
logger.info("=" * 60)
logger.info("バックフィル開始: site=%s start=%s end=%s sources=%s",
args.site, args.start, args.end or "auto", sorted(sources))
credentials = common.load_credentials(cfg, logger)
common.init_db(cfg["database"])
conn = common.connect(cfg["database"])
run_date = common.ymd(today)
total = 0
failures = 0
if sources & {"ga4", "page", "affiliate"}:
client = fetch_ga4.build_client(credentials)
ga4_end = end_arg or (today - timedelta(days=1))
if "ga4" in sources:
try:
n = backfill_ga4_daily(conn, client, site_cfg, cfg, start, ga4_end, logger)
total += n
common.record_batch_run(conn, args.site, run_date, "ga4_backfill",
"success", n, f"{start} - {ga4_end}", _now())
except Exception as exc: # noqa: BLE001
failures += 1
logger.exception("GA4日次系のバックフィル失敗: %s", exc)
conn.rollback()
common.record_batch_run(conn, args.site, run_date, "ga4_backfill",
"failed", 0, str(exc)[:500], _now())
if "page" in sources:
# snapshot_date は実行日なので終端は today（窓は昨日まで）
page_end = end_arg or today
try:
n = backfill_page_snapshots(conn, client, site_cfg, cfg, start, page_end, logger)
total += n
common.record_batch_run(conn, args.site, run_date, "ga4_page_backfill",
"success", n, f"{start} - {page_end}", _now())
except Exception as exc: # noqa: BLE001
failures += 1
logger.exception("page_metrics のバックフィル失敗: %s", exc)
common.record_batch_run(conn, args.site, run_date, "ga4_page_backfill",
"failed", 0, str(exc)[:500], _now())
if "affiliate" in sources and site_cfg.get("ga4_affiliate_event"):
try:
n = fetch_ga4.fetch_affiliate(conn, client, site_cfg, cfg, today, logger,
start=common.ymd(start),
end=common.ymd(ga4_end))
total += n
common.record_batch_run(conn, args.site, run_date, "ga4_affiliate_backfill",
"success", n, f"{start} - {ga4_end}", _now())
except Exception as exc: # noqa: BLE001
failures += 1
logger.exception("アフィリエイトクリックのバックフィル失敗: %s", exc)
common.record_batch_run(conn, args.site, run_date, "ga4_affiliate_backfill",
"failed", 0, str(exc)[:500], _now())
if "gsc" in sources:
service = fetch_gsc.build_service(credentials)
gsc_end = end_arg or (today - timedelta(days=cfg["gsc_lag_days"]))
try:
n = backfill_gsc(conn, service, site_cfg, cfg, start, gsc_end, logger)
total += n
common.record_batch_run(conn, args.site, run_date, "gsc_backfill",
"success", n, f"{start} - {gsc_end}", _now())
except Exception as exc: # noqa: BLE001
failures += 1
logger.exception("GSCのバックフィル失敗: %s", exc)
common.record_batch_run(conn, args.site, run_date, "gsc_backfill",
"failed", 0, str(exc)[:500], _now())
conn.close()
logger.info("バックフィル完了: 書込総行数=%d, 失敗数=%d", total, failures)
logger.info("=" * 60)
return 1 if failures else 0
if __name__ == "__main__":
raise SystemExit(main())
MIT ライセンスで配布しています。