"""GA4 Data API v1beta からデータを取得して SQLite に upsert する。
- ページ別(#1)は ローリング期間の合計・上位N の単発レポート（snapshot_date = 実行日）。
- 日次系(#2〜#7)は batchRunReports でまとめて取得（リクエスト数の節約）。
- 鮮度ルール: 日次系は直近 reprocess_days 日を毎回取り直す（GA4 の再処理対策）。
- アフィリエイトクリック(#8)は任意。サイトに ga4_affiliate_event があるときだけ取得。
"""
from __future__ import annotations
import logging
import sqlite3
from datetime import date, timedelta
from typing import Any
from google.analytics.data_v1beta import BetaAnalyticsDataClient
from google.analytics.data_v1beta.types import (
BatchRunReportsRequest,
DateRange,
Dimension,
Filter,
FilterExpression,
Metric,
OrderBy,
RunReportRequest,
)
import common
def build_client(credentials) -> BetaAnalyticsDataClient:
"""OAuth / サービスアカウントいずれのクレデンシャルでも GA4 クライアントを構築する。"""
return BetaAnalyticsDataClient(credentials=credentials)
def _fmt_ga4_date(v: str) -> str:
"""GA4 の date ディメンション 'YYYYMMDD' を 'YYYY-MM-DD' に変換。"""
if v and len(v) == 8 and v.isdigit():
return f"{v[0:4]}-{v[4:6]}-{v[6:8]}"
return v
def _rows(resp: Any) -> list[tuple[list[str], list[str]]]:
"""RunReportResponse から (dim値リスト, metric値リスト) のリストを返す。"""
out = []
for row in resp.rows:
dims = [d.value for d in row.dimension_values]
mets = [m.value for m in row.metric_values]
out.append((dims, mets))
return out
# --------------------------------------------------------------------------- #
# #1 ページ別（単発: ローリング期間の合計・上位N）
# --------------------------------------------------------------------------- #
def _page_report_request(property_id: str, cfg: dict, today: date) -> RunReportRequest:
# 「ローリング期間ぶんの合計」: 終端=昨日、開始=page_rolling_days 日前。
start = common.ymd(today - timedelta(days=cfg["page_rolling_days"]))
end = common.ymd(today - timedelta(days=1))
return RunReportRequest(
property=f"properties/{property_id}",
dimensions=[Dimension(name="pagePath"), Dimension(name="pageTitle")],
metrics=[
Metric(name="screenPageViews"),
Metric(name="activeUsers"),
Metric(name="userEngagementDuration"),
Metric(name="engagementRate"),
Metric(name="bounceRate"),
],
date_ranges=[DateRange(start_date=start, end_date=end)],
order_bys=[
OrderBy(
metric=OrderBy.MetricOrderBy(metric_name="screenPageViews"),
desc=True,
)
],
limit=cfg["page_top_n"],
)
def _write_page_metrics(
conn: sqlite3.Connection, site: str, domain: str, snapshot_date: str, resp: Any
) -> int:
# GA4 は (pagePath, pageTitle) の組でグループ化するため、ページのタイトルを変更すると
# 同じパスが新旧タイトルの2行に分かれて返る。主キーは (site, snapshot_date, page_path)
# の INSERT OR REPLACE なので、そのまま書くと後勝ちで片方の PV が消える。
# → 正規化パス単位に合算してから upsert する。
agg: dict[str, list] = {}
for dims, mets in _rows(resp):
page_path = common.normalize_path(dims[0], domain)
pv = common.to_int(mets[0])
au = common.to_int(mets[1])
dur = common.to_float(mets[2])
er = common.to_float(mets[3])
br = common.to_float(mets[4])
cur = agg.get(page_path)
if cur is None:
# [title, pv, au, dur, er加重和, br加重和, title側pv]
agg[page_path] = [dims[1], pv, au, dur, er * pv, br * pv, pv]
else:
cur[1] += pv
cur[2] += au # activeUsers は行間で重複しうるので近似の合算
cur[3] += dur
cur[4] += er * pv # 率は PV 加重平均で合成
cur[5] += br * pv
if pv > cur[6]: # タイトルは PV 最大の行（=現行タイトル想定）を代表にする
cur[0] = dims[1]
cur[6] = pv
rows = []
for page_path, (title, pv, au, dur, er_w, br_w, _tpv) in agg.items():
rows.append([
site, snapshot_date, page_path, title,
pv, au, dur,
(er_w / pv) if pv else 0.0,
(br_w / pv) if pv else 0.0,
])
return common.upsert(
conn, "page_metrics",
["site", "snapshot_date", "page_path", "page_title", "page_views",
"active_users", "engagement_duration", "engagement_rate", "bounce_rate"],
rows,
)
# --------------------------------------------------------------------------- #
# #2〜#7 日次系（batchRunReports に集約）
# --------------------------------------------------------------------------- #
def _daily_requests(cfg: dict, today: date) -> list[RunReportRequest]:
start, end = common.ga4_date_range(cfg["ga4_reprocess_days"], today)
dr = [DateRange(start_date=start, end_date=end)]
# 1レポートの行数上限。API の既定値のままだと長期間のバックフィルで行が切り捨てられる。
daily_limit = cfg.get("ga4_daily_row_limit", 100000)
def req(dims, mets, order_metric=None, limit=None):
kwargs = dict(
dimensions=[Dimension(name=d) for d in dims],
metrics=[Metric(name=m) for m in mets],
date_ranges=dr,
limit=limit or daily_limit,
)
if order_metric:
kwargs["order_bys"] = [
OrderBy(metric=OrderBy.MetricOrderBy(metric_name=order_metric), desc=True)
]
return RunReportRequest(**kwargs)
return [
# 2. 流入チャネル
req(["date", "sessionDefaultChannelGroup", "sessionSource", "sessionMedium"],
["sessions", "activeUsers", "engagedSessions"]),
# 3. ランディングページ
# 注意: ここに「上位N」の limit を付けないこと。limit はレポート全体に掛かるので、
# 複数日のレンジに付けると「日×ページ」の上位N組しか残らず、1日あたりの
# 行が大きく欠ける。
req(["date", "landingPagePlusQueryString"],
["sessions", "bounceRate", "engagementRate"],
order_metric="sessions"),
# 4. 新規 vs リピーター
req(["date", "newVsReturning"], ["activeUsers", "sessions"]),
# 5. デバイス・地域
req(["date", "deviceCategory", "country", "region"],
["activeUsers", "sessions"]),
# 6. 時間別
req(["date", "dayOfWeek", "hour"],
["sessions", "activeUsers", "screenPageViews"]),
# 7. イベント / キーイベント
req(["date", "eventName"], ["keyEvents", "eventCount", "totalRevenue"]),
]
def _write_channel(conn, site, resp) -> int:
rows = []
for dims, mets in _rows(resp):
rows.append([
site, _fmt_ga4_date(dims[0]), dims[1], dims[2], dims[3],
common.to_int(mets[0]), common.to_int(mets[1]), common.to_int(mets[2]),
])
return common.upsert(
conn, "channel_metrics",
["site", "date", "channel_group", "source", "medium",
"sessions", "active_users", "engaged_sessions"],
rows,
)
def _write_landing(conn, site, domain, resp) -> int:
# landingPagePlusQueryString はクエリ文字列違いで同じパスが複数行になる。
# 正規化後にそのまま INSERT OR REPLACE すると後勝ちでセッションが消えるので、
# 正規化パス単位で合算してから upsert する（sessions=合算、率=セッション加重平均）。
agg: dict = {}
for dims, mets in _rows(resp):
key = (_fmt_ga4_date(dims[0]), common.normalize_path(dims[1], domain))
sess = common.to_int(mets[0])
rec = agg.setdefault(key, {"sessions": 0, "br_w": 0.0, "er_w": 0.0})
rec["sessions"] += sess
rec["br_w"] += common.to_float(mets[1]) * sess
rec["er_w"] += common.to_float(mets[2]) * sess
rows = [
[site, d, path, r["sessions"],
(r["br_w"] / r["sessions"]) if r["sessions"] else 0.0,
(r["er_w"] / r["sessions"]) if r["sessions"] else 0.0]
for (d, path), r in agg.items()
]
return common.upsert(
conn, "landing_metrics",
["site", "date", "landing_page", "sessions", "bounce_rate", "engagement_rate"],
rows,
)
def _write_user_type(conn, site, resp) -> int:
rows = []
for dims, mets in _rows(resp):
rows.append([
site, _fmt_ga4_date(dims[0]), (dims[1] or "unknown"),
common.to_int(mets[0]), common.to_int(mets[1]),
])
return common.upsert(
conn, "user_type_metrics",
["site", "date", "user_type", "active_users", "sessions"],
rows,
)
def _write_device_geo(conn, site, resp) -> int:
rows = []
for dims, mets in _rows(resp):
rows.append([
site, _fmt_ga4_date(dims[0]), dims[1], dims[2], dims[3],
common.to_int(mets[0]), common.to_int(mets[1]),
])
return common.upsert(
conn, "device_geo_metrics",
["site", "date", "device_category", "country", "region",
"active_users", "sessions"],
rows,
)
def _write_hourly(conn, site, resp) -> int:
rows = []
for dims, mets in _rows(resp):
rows.append([
site, _fmt_ga4_date(dims[0]),
common.to_int(dims[2]), # hour
common.to_int(dims[1]), # day_of_week
common.to_int(mets[0]), common.to_int(mets[1]), common.to_int(mets[2]),
])
return common.upsert(
conn, "hourly_metrics",
["site", "date", "hour", "day_of_week", "sessions", "active_users", "page_views"],
rows,
)
def _write_event(conn, site, resp) -> int:
rows = []
for dims, mets in _rows(resp):
rows.append([
site, _fmt_ga4_date(dims[0]), dims[1],
common.to_float(mets[0]), common.to_int(mets[1]), common.to_float(mets[2]),
])
return common.upsert(
conn, "event_metrics",
["site", "date", "event_name", "key_events", "event_count", "total_revenue"],
rows,
)
# --------------------------------------------------------------------------- #
# #8 アフィリエイトクリック（任意・ページ別×遷移先ホスト別）
# --------------------------------------------------------------------------- #
# サイト側の計測タグが「広告リンクのクリック」を GA4 イベントとして送っている場合に、
# そのイベントをページ別・遷移先ホスト別に日次集計する。
#
# 前提（サイト側で用意するもの）:
# - イベント名: config の各サイトの ga4_affiliate_event に書く（例: affiliate_click）。
# 未設定のサイトはこのレポートを実行しない。
# - イベントパラメータ article_path（クリックされたページのパス）と
# link_host（遷移先のホスト名）を送り、GA4 管理画面で同名のイベントスコープの
# カスタムディメンションとして登録しておく。
# 注意: カスタムディメンションが未登録のプロパティでは API が
# "not a valid dimension" エラーを返す。その場合は warning を出してスキップする
# （バッチ全体は止めない）。また GA4 は登録前のイベントに遡って値を付けないので、
# 登録日より前は "(not set)" になる。
def _is_missing_custom_dimension(exc: Exception) -> bool:
"""GA4 の「customEvent:* が未登録」エラーか判定する。
設定不備なのでリトライしても直らない。with_retry(give_up_if=...) に渡して
即あきらめさせる。
"""
msg = str(exc)
return "not a valid dimension" in msg and "customEvent:" in msg
_AFF_DIMS = [
"date",
"customEvent:article_path",
"customEvent:link_host",
]
def _affiliate_request(
property_id: str, cfg: dict, start: str, end: str, event_name: str
) -> RunReportRequest:
return RunReportRequest(
property=f"properties/{property_id}",
dimensions=[Dimension(name=d) for d in _AFF_DIMS],
metrics=[Metric(name="eventCount")],
date_ranges=[DateRange(start_date=start, end_date=end)],
dimension_filter=FilterExpression(
filter=Filter(
field_name="eventName",
string_filter=Filter.StringFilter(value=event_name),
)
),
limit=cfg.get("ga4_daily_row_limit", 100000),
)
def _write_affiliate(conn, site, domain, resp) -> int:
# article_path は他テーブルと JOIN できるよう正規化する。"(not set)" はそのまま保持。
# 正規化で別の行が同じキーに畳まれることがあるので、upsert 前に合算する。
agg: dict[tuple[str, str, str], int] = {}
for dims, mets in _rows(resp):
path = dims[1]
if path and path != "(not set)":
path = common.normalize_path(path, domain)
else:
path = path or "(not set)"
key = (_fmt_ga4_date(dims[0]), path, dims[2] or "(not set)")
agg[key] = agg.get(key, 0) + common.to_int(mets[0])
rows = [[site, d, p, h, c] for (d, p, h), c in agg.items()]
return common.upsert(
conn, "affiliate_click_metrics",
["site", "date", "article_path", "link_host", "event_count"],
rows,
)
def fetch_affiliate(
conn: sqlite3.Connection,
client: BetaAnalyticsDataClient,
site_cfg: dict,
cfg: dict,
today: date,
logger: logging.Logger,
start: str | None = None,
end: str | None = None,
) -> int:
"""アフィリエイトクリックのページ別・遷移先ホスト別の件数を取得→upsert。
サイトに ga4_affiliate_event が無ければ API を呼ばず 0 行で返す。
通常バッチでは日次系と同じ鮮度ルール（直近 reprocess_days 日を delete→insert）。
start/end を指定するとその期間を取り直す（バックフィル用）。
カスタムディメンション未登録のプロパティは warning を出して 0 行でスキップする。
"""
site = site_cfg["site"]
event_name = site_cfg.get("ga4_affiliate_event")
if not event_name:
logger.info("[%s][ga4_affiliate] ga4_affiliate_event 未設定のためスキップ", site)
return 0
domain = site_cfg["domain"]
property_id = str(site_cfg["ga4_property_id"])
if start is None or end is None:
start, end = common.ga4_date_range(cfg["ga4_reprocess_days"], today)
req = _affiliate_request(property_id, cfg, start, end, str(event_name))
try:
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,
give_up_if=_is_missing_custom_dimension,
)
except Exception as exc: # noqa: BLE001
msg = str(exc)
if "not a valid dimension" in msg or "customEvent:" in msg:
logger.warning(
"[%s][ga4_affiliate] カスタムディメンション未登録のためスキップ: %s",
site, msg[:200],
)
return 0
raise
common.delete_date_range(conn, "affiliate_click_metrics", site, start, end)
n = _write_affiliate(conn, site, domain, resp)
conn.commit()
logger.info(
"[%s][ga4_affiliate] affiliate_click_metrics: %d 行 (%s〜%s)",
site, n, start, end,
)
return n
# --------------------------------------------------------------------------- #
# エントリ
# --------------------------------------------------------------------------- #
def fetch(
conn: sqlite3.Connection,
client: BetaAnalyticsDataClient,
site_cfg: dict,
cfg: dict,
today: date,
logger: logging.Logger,
) -> int:
"""1サイト分のGA4データを取得→upsert。書き込んだ総行数を返す。失敗時は例外送出。"""
site = site_cfg["site"]
domain = site_cfg["domain"]
property_id = str(site_cfg["ga4_property_id"])
snapshot_date = common.ymd(today)
max_retries = cfg.get("retry_max", 3)
base_delay = cfg.get("retry_base_delay", 2.0)
total = 0
# --- #1 ページ別（単発レポート） ---
page_req = _page_report_request(property_id, cfg, today)
page_resp = common.with_retry(
client.run_report, page_req,
max_retries=max_retries, base_delay=base_delay, logger=logger,
)
n = _write_page_metrics(conn, site, domain, snapshot_date, page_resp)
total += n
logger.info("[%s][ga4] page_metrics: %d 行", site, n)
# --- #2〜#7 日次系（batchRunReports） ---
# 再取得対象期間を delete→insert にする（GA4 が返さなくなった古い行を取り除く）。
daily_tables = [
"channel_metrics", "landing_metrics", "user_type_metrics",
"device_geo_metrics", "hourly_metrics", "event_metrics",
]
d_start, d_end = common.ga4_date_range(cfg["ga4_reprocess_days"], today)
for t in daily_tables:
common.delete_date_range(conn, t, site, d_start, d_end)
# batchRunReports は1回あたり5レポートまでなので、5件ずつに分割して投げる。
daily_reqs = _daily_requests(cfg, today)
writers = [
("channel_metrics", lambda r: _write_channel(conn, site, r)),
("landing_metrics", lambda r: _write_landing(conn, site, domain, r)),
("user_type_metrics", lambda r: _write_user_type(conn, site, r)),
("device_geo_metrics", lambda r: _write_device_geo(conn, site, r)),
("hourly_metrics", lambda r: _write_hourly(conn, site, r)),
("event_metrics", lambda r: _write_event(conn, site, r)),
]
MAX_BATCH = 5
for i in range(0, len(daily_reqs), MAX_BATCH):
chunk = daily_reqs[i:i + MAX_BATCH]
batch_req = BatchRunReportsRequest(
property=f"properties/{property_id}",
requests=chunk,
)
batch_resp = common.with_retry(
client.batch_run_reports, batch_req,
max_retries=max_retries, base_delay=base_delay, logger=logger,
)
for (name, writer), report in zip(writers[i:i + MAX_BATCH], batch_resp.reports):
n = writer(report)
total += n
logger.info("[%s][ga4] %s: %d 行", site, name, n)
conn.commit()
return total
MIT ライセンスで配布しています。