"""エントリ: config読込 → DB初期化 → 全サイトループ(GA4/SC取得→upsert) → CSV出力 → ログ。
毎日1回、無人で実行する取得バッチ。分析は蓄積された SQLite を別途読んで行う。
使い方:
（プロジェクトのルートで実行する）
python -X utf8 scripts/analytics/main.py # config.yaml の全サイトを取得
python -X utf8 scripts/analytics/main.py --config x.yaml # 別の設定ファイルを指定
python -X utf8 scripts/analytics/main.py --no-csv # CSVエクスポートをスキップ
設計方針:
- 部分失敗を許容: 1サイト/1ソースの失敗が全体を止めない。失敗は batch_runs に記録。
- 冪等: 日次系・GSC は取得窓を delete→insert、他は INSERT OR REPLACE。
同じ日に2回流しても行数は増えない。
"""
from __future__ import annotations
import argparse
import csv
import os
import sqlite3
from datetime import date, datetime
import common
import fetch_ga4
import fetch_gsc
import fetch_url_inspection
# CSVエクスポート対象テーブル
EXPORT_TABLES = [
"page_metrics", "channel_metrics", "landing_metrics", "user_type_metrics",
"device_geo_metrics", "hourly_metrics", "event_metrics",
"affiliate_click_metrics",
"gsc_query_metrics", "gsc_page_metrics", "gsc_page_query_metrics",
"inspect_metrics", "batch_runs",
]
# ga4_property_id が未設定とみなす値（空 / サンプル設定のプレースホルダ）
UNSET_PROPERTY_IDS = ("", "PROPERTYID", "000000000")
def _now() -> str:
return datetime.now().isoformat(timespec="seconds")
def export_csv(db_path: str, export_dir: str, run_date: str, logger) -> None:
"""各テーブルを <export_dir>/<run_date>/<table>.csv にダンプ（確認/バックアップ用）。"""
out_dir = os.path.join(export_dir, run_date)
os.makedirs(out_dir, exist_ok=True)
conn = sqlite3.connect(db_path)
try:
for table in EXPORT_TABLES:
cur = conn.execute(f"SELECT * FROM {table}")
cols = [d[0] for d in cur.description]
path = os.path.join(out_dir, f"{table}.csv")
with open(path, "w", newline="", encoding="utf-8-sig") as f:
w = csv.writer(f)
w.writerow(cols)
w.writerows(cur.fetchall())
logger.info("CSVエクスポート完了: %s", out_dir)
finally:
conn.close()
def run_source(conn, source, fetch_fn, client, site_cfg, cfg, today, run_date, logger):
"""1サイト×1ソースを実行し、結果を batch_runs に記録。失敗しても例外を飲み込む。"""
site = site_cfg["site"]
try:
rows = fetch_fn(conn, client, site_cfg, cfg, today, logger)
common.record_batch_run(
conn, site, run_date, source, "success", rows,
"ok", _now(),
)
return rows
except Exception as exc: # noqa: BLE001
logger.exception("[%s][%s] 取得失敗: %s", site, source, exc)
# 未コミットの delete/insert を破棄してから失敗を記録する。
# record_batch_run は commit するため、rollback しないと取得処理の途中で
# 行った削除だけが確定して既存データが消えてしまう。
conn.rollback()
try:
common.record_batch_run(
conn, site, run_date, source, "failed", 0,
str(exc)[:500], _now(),
)
except Exception: # noqa: BLE001
logger.error("[%s][%s] batch_runs記録も失敗", site, source)
return 0
def main() -> int:
parser = argparse.ArgumentParser(description="アクセス解析バッチ (GA4 + Search Console → SQLite)")
parser.add_argument("--config", default=None, help="設定ファイルのパス（既定: プロジェクトのルートの config.yaml。読むのは analytics: 節）")
parser.add_argument("--no-csv", action="store_true", help="CSVエクスポートをスキップ")
parser.add_argument("--inspect-now", action="store_true",
help="URL Inspectionを曜日に関わらず今すぐ実行")
parser.add_argument("--inspect-limit", type=int, default=None,
help="URL Inspectionの検査URL数上限（テスト用）")
args = parser.parse_args()
cfg = common.load_config(args.config)
if args.inspect_limit is not None:
cfg["inspect_max_urls"] = args.inspect_limit
logger = common.setup_logging(cfg["log_file"])
today = date.today()
run_date = common.ymd(today)
logger.info("=" * 60)
logger.info("バッチ開始: run_date=%s", run_date)
# 認証クレデンシャルを取得（auth_mode: oauth / service_account）
logger.info("認証方式: %s", cfg.get("auth_mode", "service_account"))
try:
credentials = common.load_credentials(cfg, logger)
except Exception as exc: # noqa: BLE001
logger.error("認証に失敗しました: %s", exc)
return 2
# DB初期化（IF NOT EXISTS で冪等）
common.init_db(cfg["database"])
conn = common.connect(cfg["database"])
# クライアントは1回だけ構築（サイト間で再利用）
ga4_client = None
gsc_service = None
try:
ga4_client = fetch_ga4.build_client(credentials)
except Exception as exc: # noqa: BLE001
logger.exception("GA4クライアント構築失敗: %s", exc)
try:
gsc_service = fetch_gsc.build_service(credentials)
except Exception as exc: # noqa: BLE001
logger.exception("SCサービス構築失敗: %s", exc)
total_rows = 0
failures = 0
for site_cfg in cfg.get("sites", []) or []:
site = site_cfg["site"]
logger.info("-" * 40)
logger.info("サイト処理: %s", site)
# GA4 property_id 未設定の保険
if str(site_cfg.get("ga4_property_id") or "").upper() in UNSET_PROPERTY_IDS:
logger.warning("[%s] ga4_property_id が未設定です。GA4取得をスキップ。", site)
common.record_batch_run(
conn, site, run_date, "ga4", "failed", 0,
"ga4_property_id未設定", _now(),
)
failures += 1
elif ga4_client is None:
failures += 1
common.record_batch_run(
conn, site, run_date, "ga4", "failed", 0,
"GA4クライアント未構築", _now(),
)
else:
n = run_source(conn, "ga4", fetch_ga4.fetch, ga4_client,
site_cfg, cfg, today, run_date, logger)
total_rows += n
if n == 0:
failures += 1
# アフィリエイトクリック（任意）。サイトに ga4_affiliate_event が
# 設定されているときだけ実行する。失敗しても failures には数えない
# （本体の取得結果に影響させない）。
if site_cfg.get("ga4_affiliate_event"):
n = run_source(conn, "ga4_affiliate", fetch_ga4.fetch_affiliate, ga4_client,
site_cfg, cfg, today, run_date, logger)
total_rows += n
if gsc_service is None:
failures += 1
common.record_batch_run(
conn, site, run_date, "gsc", "failed", 0,
"SCサービス未構築", _now(),
)
else:
n = run_source(conn, "gsc", fetch_gsc.fetch, gsc_service,
site_cfg, cfg, today, run_date, logger)
total_rows += n
# URL Inspection（インデックス状況・既定は週1回）。gsc_service を流用。
inspect_weekday = cfg.get("inspect_weekday")
run_inspect = (
args.inspect_now
or inspect_weekday is None
or today.weekday() == inspect_weekday
)
if gsc_service is None or not site_cfg.get("sitemap_url"):
pass # SC未構築 or sitemap未設定のサイトは対象外
elif not run_inspect:
common.record_batch_run(
conn, site, run_date, "url_inspection", "skipped", 0,
f"inspect_weekday={inspect_weekday}（本日 weekday={today.weekday()}）", _now(),
)
logger.info("[%s][url_inspection] 本日は実行日でないためスキップ", site)
else:
n = run_source(conn, "url_inspection", fetch_url_inspection.fetch,
gsc_service, site_cfg, cfg, today, run_date, logger)
total_rows += n
conn.close()
# CSVエクスポート
if not args.no_csv:
try:
export_csv(cfg["database"], cfg["csv_export_dir"], run_date, logger)
except Exception as exc: # noqa: BLE001
logger.exception("CSVエクスポート失敗: %s", exc)
logger.info("バッチ完了: 書込総行数=%d, 失敗数=%d", total_rows, failures)
logger.info("=" * 60)
# 全滅(行0かつ失敗あり)なら非ゼロ終了。部分成功は0で返す。
return 1 if (total_rows == 0 and failures > 0) else 0
if __name__ == "__main__":
raise SystemExit(main())
MIT ライセンスで配布しています。