"""共通処理: 設定読込・認証・ログ・URL正規化・リトライ・DB初期化/upsert。
GA4(fetch_ga4.py) と Search Console(fetch_gsc.py / fetch_url_inspection.py) の
両方から使う。
"""
from __future__ import annotations
import logging
import os
import sqlite3
import time
from datetime import date, timedelta
from typing import Any, Callable, Iterable, Sequence
from urllib.parse import urlsplit
import yaml
# --------------------------------------------------------------------------- #
# 設定
# --------------------------------------------------------------------------- #
# このファイルのあるディレクトリ（scripts/analytics/）。db_schema.sql の置き場所。
BASE_DIR = os.path.dirname(os.path.abspath(__file__))
# プロジェクトのルート（scripts/analytics/ の2つ上）。config.yaml の既定の置き場所で、
# config 内の相対パスもここを基準に解決する。
PROJECT_ROOT = os.path.dirname(os.path.dirname(BASE_DIR))
# GA4 + Search Console 両方の読み取り専用スコープ（OAuth / サービスアカウント共通）
SCOPES = [
"https://www.googleapis.com/auth/analytics.readonly",
"https://www.googleapis.com/auth/webmasters.readonly",
]
def _resolve(path: str) -> str:
"""config 内の相対パスを、プロジェクトのルート基準の絶対パスに変換する。"""
if os.path.isabs(path):
return path
return os.path.normpath(os.path.join(PROJECT_ROOT, path))
def load_config(config_path: str | None = None) -> dict[str, Any]:
"""config.yaml の analytics: 節を読み込み、パス系のキーを絶対パスに解決して返す。
設定ファイルはプロジェクト全体で1つ（config.yaml）。そのうちアクセス解析が使うのは
analytics: 節だけなので、その中身を返す（呼び出し側は従来どおり cfg["database"] 等で読める）。
"""
config_path = config_path or os.path.join(PROJECT_ROOT, "config.yaml")
if not os.path.exists(config_path):
raise FileNotFoundError(
f"設定ファイルがありません: {config_path}\n"
"→ プロジェクトのルートで config.example.yaml を config.yaml にコピーして値を埋めてください。"
)
with open(config_path, "r", encoding="utf-8") as f:
whole = yaml.safe_load(f) or {}
cfg = whole.get("analytics")
if not isinstance(cfg, dict):
raise ValueError(
f"設定ファイルに analytics: 節がありません: {config_path}\n"
"→ config.example.yaml の analytics: 節を config.yaml に写して値を埋めてください。"
)
path_keys = (
"service_account_file", "oauth_client_file", "oauth_token_file",
"database", "csv_export_dir", "log_file",
)
for key in path_keys:
if key in cfg and cfg[key]:
cfg[key] = _resolve(cfg[key])
register_path_aliases(cfg)
return cfg
# --------------------------------------------------------------------------- #
# 認証（OAuthユーザー認証 / サービスアカウント）
# --------------------------------------------------------------------------- #
def load_credentials(cfg: dict[str, Any], logger: logging.Logger | None = None):
"""auth_mode に応じて Google API のクレデンシャルを返す（GA4 / SC 共用）。
oauth: auth_setup.py が生成したトークンファイルを読み、失効していれば自動更新する。
service_account: サービスアカウントの鍵ファイルを読む。
失敗時は分かりやすいメッセージで RuntimeError を送出する。
"""
mode = (cfg.get("auth_mode") or "service_account").lower()
if mode == "oauth":
from google.auth.transport.requests import Request
from google.oauth2.credentials import Credentials as UserCredentials
token_file = cfg.get("oauth_token_file")
if not token_file or not os.path.exists(token_file):
raise RuntimeError(
f"OAuthトークンがありません: {token_file}\n"
"→ 先に `python -X utf8 scripts/analytics/auth_setup.py` を実行してブラウザ認証してください。"
)
creds = UserCredentials.from_authorized_user_file(token_file, SCOPES)
if not creds.valid:
if creds.expired and creds.refresh_token:
if logger:
logger.info("OAuthトークンを自動リフレッシュします。")
creds.refresh(Request())
with open(token_file, "w", encoding="utf-8") as f:
f.write(creds.to_json())
else:
raise RuntimeError(
"OAuthトークンが無効/失効しています。"
"`python -X utf8 scripts/analytics/auth_setup.py --force` で再認証してください。"
)
return creds
# --- service_account ---
from google.oauth2 import service_account
sa_file = cfg.get("service_account_file")
if not sa_file or not os.path.exists(sa_file):
raise RuntimeError(f"サービスアカウントの鍵ファイルがありません: {sa_file}")
return service_account.Credentials.from_service_account_file(sa_file, scopes=SCOPES)
# --------------------------------------------------------------------------- #
# ログ
# --------------------------------------------------------------------------- #
def setup_logging(log_file: str) -> logging.Logger:
"""ファイル + 標準出力の両方に出すロガーを構成する。
ログは月次ローテーション（batch.log -> batch_YYYY-MM.log）。
月が替わると新ファイルに切り替わり、古いファイルは削除せず保持する。
"""
base, ext = os.path.splitext(log_file)
log_file = f"{base}_{date.today().strftime('%Y-%m')}{ext}"
os.makedirs(os.path.dirname(log_file), exist_ok=True)
logger = logging.getLogger("analytics_pipeline")
logger.setLevel(logging.INFO)
logger.handlers.clear() # 二重ハンドラ防止（再実行・import時）
fmt = logging.Formatter("%(asctime)s [%(levelname)s] %(message)s")
fh = logging.FileHandler(log_file, encoding="utf-8")
fh.setFormatter(fmt)
logger.addHandler(fh)
sh = logging.StreamHandler()
sh.setFormatter(fmt)
logger.addHandler(sh)
return logger
# --------------------------------------------------------------------------- #
# 日付ヘルパ（データ鮮度ルール）
# --------------------------------------------------------------------------- #
def ymd(d: date) -> str:
return d.strftime("%Y-%m-%d")
def ga4_date_range(reprocess_days: int, today: date | None = None) -> tuple[str, str]:
"""GA4 日次系の取得範囲。直近 reprocess_days 日を取り直す。
終端は昨日（today-1）。GA4 は直近のデータが後から再処理されて確定するため、
毎回さかのぼって取り直す。
"""
today = today or date.today()
end = today - timedelta(days=1) # 昨日
start = today - timedelta(days=reprocess_days) # reprocess_days 日前
return ymd(start), ymd(end)
def gsc_date_range(
lag_days: int, rolling_days: int, today: date | None = None
) -> tuple[str, str]:
"""Search Console の取得範囲。終端 today-lag_days、そこから rolling_days 日ぶん遡る。"""
today = today or date.today()
end = today - timedelta(days=lag_days) # データ遅延ぶん手前を終端にする
start = end - timedelta(days=rolling_days - 1) # rolling_days 日のウィンドウ
return ymd(start), ymd(end)
# --------------------------------------------------------------------------- #
# URL正規化（GA4 と Search Console の URL 表記差を吸収）
# --------------------------------------------------------------------------- #
# サイト別の名寄せ規則（domain -> 規則 dict）。
# load_config() が config.yaml の analytics.sites から登録する。normalize_path() は domain を
# 受け取ってこの登録簿を引くので、取得側のコードはサイトごとの規則を意識しなくてよい。
# 未登録のドメインは規則なし（共通の正規化だけが掛かる）。
_PATH_ALIASES: dict[str, dict[str, Any]] = {}
def register_path_aliases(cfg: dict[str, Any]) -> None:
"""config の sites から名寄せ規則を登録する（load_config から呼ばれる）。"""
_PATH_ALIASES.clear()
for site_cfg in cfg.get("sites", []) or []:
domain = site_cfg.get("domain")
if not domain:
continue
rule = {
"strip_html": bool(site_cfg.get("strip_html_suffix")),
"strip_index": bool(site_cfg.get("strip_index_html")),
"strip_prefix": site_cfg.get("strip_path_prefix") or None,
}
if rule["strip_html"] or rule["strip_index"] or rule["strip_prefix"]:
_PATH_ALIASES[domain] = rule
def normalize_path(url_or_path: str, domain: str | None = None) -> str:
"""ドメイン無しの正規化パスに統一する。
- (1) ドメイン除去（フルURLでもパスのみでもOK）
- (2) クエリ文字列除去
- (3) フラグメント除去
- (4) サイト別の名寄せ（strip_html_suffix / strip_index_html / strip_path_prefix）
- (5) 末尾スラッシュ統一（ルート '/' 以外は末尾スラッシュを削る）
GA4 の pagePath（/blog/xxx）と Search Console の page（https://example.com/blog/xxx）
を同じキー '/blog/xxx' に揃えて、テーブル間で JOIN できるようにする。
(4) は「サーバ側の書き換えで同じページになる別表記」を1キーに畳むためのもの。
サイトの URL 設計に合うものだけを config で有効にする:
- strip_html_suffix: 末尾の `.html` を落とす（/about.html -> /about、
/index.html -> /、/foo/index.html -> /foo）。
拡張子なしURLと .html 付きURLが同じページを指すサイト向け。
`.html` 付きが正規URLのサイトでは有効にしないこと（別ページが混ざる）。
- strip_index_html: パスが `/index.html` で終わるときだけ親に丸める
（/index.html -> /、/foo/index.html -> /foo）。それ以外の `.html` には触らない。
ディレクトリ index だけが二重表記になるサイト向け。
- strip_path_prefix: 指定プレフィックス配下のパスからプレフィックスを剥がす
（"/articles" なら /articles/foo -> /foo）。プレフィックスそのもの（/articles）は
一覧ページとみなして畳まない。
"""
if url_or_path is None:
return "/"
s = url_or_path.strip()
if not s:
return "/"
parts = urlsplit(s)
path = parts.path if (parts.scheme or parts.netloc) else s
# スキーム無しの入力にクエリ/フラグメントが残っている場合の保険
if "?" in path:
path = path.split("?", 1)[0]
if "#" in path:
path = path.split("#", 1)[0]
if not path.startswith("/"):
path = "/" + path
# (4) サイト別の名寄せ（登録があるドメインのみ）
rule = _PATH_ALIASES.get(domain) if domain else None
if rule:
if rule["strip_html"] and path.lower().endswith(".html"):
path = path[:-5]
# ディレクトリ index: /index -> / 、/foo/index -> /foo
if path == "/index":
path = "/"
elif path.endswith("/index"):
path = path[:-6] or "/"
elif rule.get("strip_index") and path.lower().endswith("/index.html"):
# /index.html -> / 、/foo/index.html -> /foo （それ以外の .html は触らない）
path = path[: -len("/index.html")] or "/"
prefix = rule["strip_prefix"]
if prefix:
prefix = "/" + prefix.strip("/")
# プレフィックスそのものは畳まない。配下だけ剥がす。
if path.startswith(prefix + "/"):
path = path[len(prefix):] or "/"
# (5) 末尾スラッシュ統一（ルート以外）
if len(path) > 1:
path = path.rstrip("/")
if not path:
path = "/"
return path
# --------------------------------------------------------------------------- #
# リトライ（指数バックオフ・最大N回）
# --------------------------------------------------------------------------- #
def with_retry(
func: Callable[..., Any],
*args: Any,
max_retries: int = 3,
base_delay: float = 2.0,
logger: logging.Logger | None = None,
give_up_if: Callable[[Exception], bool] | None = None,
**kwargs: Any,
) -> Any:
"""func を指数バックオフでリトライ実行する。最後まで失敗したら例外を再送出。
give_up_if を渡すと、その述語が True を返す例外はリトライせず即座に再送出する
（設定不備のように、何度投げても直らないエラーで待ち時間を浪費しないため）。
"""
attempt = 0
while True:
try:
return func(*args, **kwargs)
except Exception as exc: # noqa: BLE001 - API例外は種類が多いので包括
if give_up_if is not None and give_up_if(exc):
raise
attempt += 1
if attempt > max_retries:
if logger:
logger.error("リトライ上限(%d)に到達: %s", max_retries, exc)
raise
delay = base_delay * (2 ** (attempt - 1))
if logger:
logger.warning(
"API失敗 (試行%d/%d) %ss後に再試行: %s",
attempt, max_retries, delay, exc,
)
time.sleep(delay)
# --------------------------------------------------------------------------- #
# DB
# --------------------------------------------------------------------------- #
def init_db(db_path: str, schema_path: str | None = None) -> None:
"""db_schema.sql を流してテーブルを作成する（IF NOT EXISTS なので冪等）。"""
schema_path = schema_path or os.path.join(BASE_DIR, "db_schema.sql")
with open(schema_path, "r", encoding="utf-8") as f:
ddl = f.read()
db_dir = os.path.dirname(db_path)
if db_dir:
os.makedirs(db_dir, exist_ok=True) # data/ などのフォルダが無ければ作る
conn = sqlite3.connect(db_path)
try:
conn.executescript(ddl)
conn.commit()
finally:
conn.close()
def connect(db_path: str) -> sqlite3.Connection:
return sqlite3.connect(db_path)
def upsert(
conn: sqlite3.Connection,
table: str,
columns: Sequence[str],
rows: Iterable[Sequence[Any]],
) -> int:
"""INSERT OR REPLACE で行を upsert する。書き込んだ行数を返す。"""
rows = list(rows)
if not rows:
return 0
placeholders = ",".join("?" * len(columns))
col_list = ",".join(columns)
sql = f"INSERT OR REPLACE INTO {table} ({col_list}) VALUES ({placeholders})"
conn.executemany(sql, rows)
return len(rows)
def delete_date_range(
conn: sqlite3.Connection, table: str, site: str, start: str, end: str
) -> int:
"""再取得前に [site, date∈[start,end]] の既存行を削除する。
upsert(INSERT OR REPLACE) だけでは「API が返さなくなった行」が消えない
（例: GA4 が後から流入元を解決して (not set) 行を返さなくなると、古い行が残る）。
再取得対象期間を delete→insert にすることで、消えた行も正しく取り除く。
"""
cur = conn.execute(
f"DELETE FROM {table} WHERE site=? AND date BETWEEN ? AND ?", (site, start, end)
)
return cur.rowcount
def record_batch_run(
conn: sqlite3.Connection,
site: str,
run_date: str,
source: str,
status: str,
rows_written: int,
message: str,
finished_at: str,
) -> None:
"""batch_runs に実行結果を記録（upsert）。"""
upsert(
conn,
"batch_runs",
["site", "run_date", "source", "status", "rows_written", "message", "finished_at"],
[[site, run_date, source, status, rows_written, message, finished_at]],
)
conn.commit()
# --------------------------------------------------------------------------- #
# 値変換ヘルパ（API が返す文字列値を安全に数値化）
# --------------------------------------------------------------------------- #
def to_int(v: Any, default: int = 0) -> int:
try:
return int(float(v))
except (TypeError, ValueError):
return default
def to_float(v: Any, default: float = 0.0) -> float:
try:
return float(v)
except (TypeError, ValueError):
return default
MIT ライセンスで配布しています。