第2回 PythonでGSC APIデータを取得しBigQueryへ保存する方法【4サイト対応】

スポンサーリンク
AIエージェント実装
2026.08.02
スポンサーリンク

前回は、Google Cloudプロジェクト、サービスアカウント、Search Consoleの閲覧権限、BigQueryデータセットを準備しました。今回は、その環境上で実際に動くPythonプログラムを作成します。

取得するのは、日付・ページ・検索クエリ・デバイス・国ごとの検索パフォーマンスデータです。毎朝直近7日分を再取得し、同じ期間の既存データを入れ替えることで、Search Console側で後から数値が更新された場合にも追従できる設計にします。

コードはGSC取得、BigQuery保存、設定、日付計算に分割します。将来GA4やAdSenseを追加しやすい構成にしているため、単発のサンプルコードではなく、ブログ分析基盤として拡張できます。

スポンサーリンク

この記事で完成する状態

  • 4サイトを管理するYAML設定ファイルが作成されている
  • GSCの取得対象期間を自動計算できる
  • Search Console APIから明細データを取得できる
  • 指定サイト・期間のBigQueryデータを入れ替えて保存できる
  • 4サイトを順番に処理するmain.pyが完成している
  • Cloud Run Jobsへ配置できるファイル構成になっている

シリーズ構成

前回までに必要な設定

この記事は、第1回で次の設定が完了していることを前提にしています。

  • プロジェクトID：lennonsoft-blog-analytics（アカウントによって名前変わります）
  • BigQueryデータセット：blog_analytics
  • 実行用サービスアカウント：blog-analytics-runner
  • Search Console APIとBigQuery APIが有効
  • Search Consoleの4プロパティにサービスアカウントを追加済み

Cloud Shellを起動して作業フォルダを作る

GSC取得用のPythonプログラムを作ります。今回はCloud Shellの起動とフォルダ作成だけ行います。今回はほとんどコピペ作業ばかりになります。

1．Cloud Shellを起動する

Google Cloud Console右上にある、次のアイコンをクリックしてください。

クリックすると初回は確認画面が出るますが、「続行」・「承認」を押して進めてください。

画面下部に黒いターミナルが開きます。

2．対象プロジェクトを設定する

Cloud Shellへ次のコマンドを貼り付けて、Enterを押してください。

gcloud config set project lennonsoft-blog-analytics

次のような表示になれば正常です。

Updated property [core/project].

課金プロジェクトに関する警告が出ても、プロジェクトIDが正しければ通常は問題ありません。

3．作業フォルダを作る

続けて、次のコマンドを実行してください。

mkdir -p ~/blog-analytics-platform
cd ~/blog-analytics-platform

現在地を確認します。

pwd

次のように表示されれば完了です。

/home/ユーザー名/blog-analytics-platform

拡張可能なフォルダ構成を作る

今回はPythonコードの中身はまだ書きません。GSCの後にGA4・AdSenseを追加できるよう、先にフォルダだけ作ります。

前回に続けてCloud Shellで以下をまとめて貼り付けてEnterを押してください。

mkdir -p collectors loaders common config tests

touch main.py
touch requirements.txt
touch collectors/__init__.py
touch collectors/gsc.py
touch collectors/ga4.py
touch collectors/adsense.py
touch loaders/__init__.py
touch loaders/bigquery.py
touch common/__init__.py
touch common/settings.py
touch common/date_utils.py
touch config/sites.yaml
touch .gcloudignore

続けて作成結果を確認します。

find . -maxdepth 2 -type f | sort

次のように表示されれば完了です。

./.gcloudignore
./collectors/__init__.py
./collectors/adsense.py
./collectors/ga4.py
./collectors/gsc.py
./common/__init__.py
./common/date_utils.py
./common/settings.py
./config/sites.yaml
./loaders/__init__.py
./loaders/bigquery.py
./main.py
./requirements.txt

この構成では、取得処理をサービス別に分離します。

コード役割
collectors/gsc.pySearch Console取得
collectors/ga4.pyGA4取得
※今回未使用
collectors/adsense.pyAdSense取得
※今回未使用
loaders/bigquery.pyBigQuery保存
config/sites.yaml4サイトの設定
main.py処理の起点

Cloud Run Jobsは処理を実行して終了するバッチ用途に適しており、Pythonソースからコンテナを構築してジョブとして展開できます。

4サイトの設定ファイルを作成する

次に、config/sites.yaml に対象サイト情報を登録します。GA4とAdSenseを後から追加できる形にしておきます。今回は私の管理している4サイトを登録しますが、皆さんは必要に応じて変更してください。

site_idSearch Consoleプロパティ
hytale_labhttps://hytale-lab.com/
lennonsofthttps://lennonsoft.com/
ai_lennonsofthttps://ai.lennonsoft.com/
witcher4https://witcher4.net/

Cloud Shellで、次のコマンドをそのまま貼り付けて実行してください。

cat > config/sites.yaml <<'EOF'
sites:
  - site_id: hytale_lab
    display_name: Hytale Lab
    domain: hytale-lab.com
    gsc_property: https://hytale-lab.com/
    ga4_property_id: ""
    adsense_enabled: true

  - site_id: lennonsoft
    display_name: Lennonsoft
    domain: lennonsoft.com
    gsc_property: https://lennonsoft.com/
    ga4_property_id: ""
    adsense_enabled: true

  - site_id: ai_lennonsoft
    display_name: AI Lennonsoft
    domain: ai.lennonsoft.com
    gsc_property: https://ai.lennonsoft.com/
    ga4_property_id: ""
    adsense_enabled: true

  - site_id: witcher4
    display_name: Witcher4
    domain: witcher4.net
    gsc_property: https://witcher4.net/
    ga4_property_id: ""
    adsense_enabled: true

settings:
  project_id: lennonsoft-blog-analytics
  bigquery_dataset: blog_analytics
  timezone: Asia/Tokyo
  gsc_refresh_days: 7
EOF

続けて、内容を確認します。

cat config/sites.yaml

注意点として、gsc_propertyはSearch Consoleに登録されているURLと完全に一致させています。末尾の / を削除しないでください。

ga4_property_idは現時点では空欄です。GA4実装時に各サイトのプロパティIDを追加します。

gsc_refresh_days: 7は、毎朝直近7日分を再取得する設定です。Search Consoleの直近データが後から更新されても、BigQuery側を修正できる設計にします。

必要なPythonライブラリを登録する

Cloud Shellで、次のコマンドをそのまま貼り付けて実行してください。

cat > requirements.txt <<'EOF'
google-api-python-client
google-auth
google-cloud-bigquery
PyYAML
EOF

それぞれの用途は次のとおりです。

コマンド役割
google-api-python-clientSearch Console APIを呼び出す
google-authCloud Runのサービスアカウントで認証する
google-cloud-bigquery取得したデータをBigQueryへ保存する
PyYAMLconfig/sites.yamlを読み込む

続けて内容を確認します。

cat requirements.txt

次の4行が表示されれば完了です。

google-api-python-client
google-auth
google-cloud-bigquery
PyYAML

まだライブラリのインストールは行いません。Cloud Runへのビルド時に自動でインストールされます。

設定ファイルを読み込む処理を作る

次は common/settings.py に、config/sites.yaml を読み込む処理を書きます。今回はこの1ファイルだけです。

Cloud Shellで、次をそのまま貼り付けて実行してください。

cat > common/settings.py <<'EOF'
from pathlib import Path
from typing import Any

import yaml


BASE_DIR = Path(__file__).resolve().parent.parent
DEFAULT_CONFIG_PATH = BASE_DIR / "config" / "sites.yaml"


def load_settings(config_path: Path = DEFAULT_CONFIG_PATH) -> dict[str, Any]:
    """YAML設定ファイルを読み込んで返す。"""
    if not config_path.exists():
        raise FileNotFoundError(
            f"設定ファイルが見つかりません: {config_path}"
        )

    with config_path.open("r", encoding="utf-8") as file:
        config = yaml.safe_load(file)

    if not isinstance(config, dict):
        raise ValueError("設定ファイルの形式が正しくありません。")

    if "sites" not in config or "settings" not in config:
        raise ValueError(
            "設定ファイルには sites と settings が必要です。"
        )

    return config
EOF

続けて、Pythonの文法チェックをします。

python3 -m py_compile common/settings.py

何も表示されず、次の入力待ち表示に戻れば正常です。さらに、設定ファイルを実際に読み込めるか確認します。

python3 - <<'EOF'
from common.settings import load_settings

config = load_settings()

print("project_id:", config["settings"]["project_id"])
print("dataset:", config["settings"]["bigquery_dataset"])
print("site_count:", len(config["sites"]))

for site in config["sites"]:
    print(site["site_id"], site["gsc_property"])
EOF

次のように表示されれば完了です。

project_id: lennonsoft-blog-analytics
dataset: blog_analytics
site_count: 4
hytale_lab https://hytale-lab.com/
lennonsoft https://lennonsoft.com/
ai_lennonsoft https://ai.lennonsoft.com/
witcher4 https://witcher4.net/

Cloud Runではサービスアカウントを通じてApplication Default Credentialsを利用できるため、JSON鍵をコード内に保存する必要はありません。

GSC取得対象の日付を計算する処理を作る

Search Consoleの最新データは確定まで時間差があるため、当日や前日ではなく「2日前まで」を取得対象にします。毎朝、直近7日分を再取得する設計です。

Cloud Shellに次を貼り付けて実行してください。

cat > common/date_utils.py <<'EOF'
from datetime import date, timedelta
from zoneinfo import ZoneInfo
from datetime import datetime


def get_gsc_date_range(
    refresh_days: int = 7,
    data_delay_days: int = 2,
    timezone: str = "Asia/Tokyo",
) -> tuple[date, date]:
    """
    GSC取得対象期間を返す。

    data_delay_days=2 の場合、2日前を終了日とする。
    refresh_days=7 の場合、終了日を含む直近7日間を返す。
    """
    if refresh_days < 1:
        raise ValueError("refresh_daysは1以上にしてください。")

    if data_delay_days < 0:
        raise ValueError("data_delay_daysは0以上にしてください。")

    today = datetime.now(ZoneInfo(timezone)).date()
    end_date = today - timedelta(days=data_delay_days)
    start_date = end_date - timedelta(days=refresh_days - 1)

    return start_date, end_date
EOF

続けて文法チェックをします。

python3 -m py_compile common/date_utils.py

何も表示されなければ正常です。さらに動作確認をしてください。

python3 - <<'EOF'
from common.date_utils import get_gsc_date_range

start_date, end_date = get_gsc_date_range()

print("start_date:", start_date)
print("end_date:", end_date)
print("days:", (end_date - start_date).days + 1)
EOF

2026年8月2日に実行した場合、次の結果になるはずです。

start_date: 2026-07-25
end_date: 2026-07-31
days: 7

Search Console APIの取得処理を作る

今回は、GSC APIを呼び出して、ページ別・クエリ別データを取得する処理をcollectors/gsc.pyに作成します。まだ実行はしません。

Cloud Shellに次をそのまま貼り付けてください。

cat > collectors/gsc.py <<'EOF'
from datetime import date
from typing import Any

import google.auth
from googleapiclient.discovery import build


GSC_SCOPES = [
    "https://www.googleapis.com/auth/webmasters.readonly",
]

ROW_LIMIT = 25_000


def create_gsc_service() -> Any:
    """実行環境のサービスアカウントを使ってGSC APIクライアントを作成する。"""
    credentials, _ = google.auth.default(scopes=GSC_SCOPES)

    return build(
        "searchconsole",
        "v1",
        credentials=credentials,
        cache_discovery=False,
    )


def fetch_gsc_rows(
    site_id: str,
    site_url: str,
    start_date: date,
    end_date: date,
) -> list[dict[str, Any]]:
    """
    指定したSearch Consoleプロパティからデータを取得する。

    取得粒度:
    日付 × ページ × クエリ × デバイス × 国
    """
    service = create_gsc_service()
    all_rows: list[dict[str, Any]] = []
    start_row = 0

    while True:
        request_body = {
            "startDate": start_date.isoformat(),
            "endDate": end_date.isoformat(),
            "dimensions": [
                "date",
                "page",
                "query",
                "device",
                "country",
            ],
            "type": "web",
            "dataState": "final",
            "rowLimit": ROW_LIMIT,
            "startRow": start_row,
        }

        response = (
            service.searchanalytics()
            .query(
                siteUrl=site_url,
                body=request_body,
            )
            .execute()
        )

        api_rows = response.get("rows", [])

        for row in api_rows:
            keys = row.get("keys", [])

            if len(keys) != 5:
                continue

            all_rows.append(
                {
                    "site_id": site_id,
                    "data_date": keys[0],
                    "page": keys[1],
                    "query": keys[2],
                    "device": keys[3],
                    "country": keys[4],
                    "clicks": float(row.get("clicks", 0)),
                    "impressions": float(row.get("impressions", 0)),
                    "ctr": float(row.get("ctr", 0)),
                    "position": float(row.get("position", 0)),
                }
            )

        if len(api_rows) < ROW_LIMIT:
            break

        start_row += ROW_LIMIT

    return all_rows
EOF

次に文法チェックを実行します。

python3 -m py_compile collectors/gsc.py

何も表示されず入力待ちに戻れば正常です。続けて、ファイル冒頭を確認します。

sed -n '1,40p' collectors/gsc.py

今回はAPI実行までは行いません。Cloud Shell上のログインユーザーではなく、最終的にはblog-analytics-runnerサービスアカウントで実行するためです。

BigQueryへの保存処理を作る

今回は loaders/bigquery.py だけを作ります。

毎朝直近7日分を再取得するため、同じサイト・同じ期間の既存データを削除してから、最新データを入れ直す方式にします。これで重複を防げます。

Cloud Shellに次をそのまま貼り付けてください。

cat > loaders/bigquery.py <<'EOF'
from datetime import date, datetime, timezone
from typing import Any

from google.cloud import bigquery


TABLE_NAME = "gsc_query_page_daily"
BQ_LOCATION = "asia-northeast1"


GSC_TABLE_SCHEMA = [
    bigquery.SchemaField("site_id", "STRING", mode="REQUIRED"),
    bigquery.SchemaField("data_date", "DATE", mode="REQUIRED"),
    bigquery.SchemaField("page", "STRING", mode="REQUIRED"),
    bigquery.SchemaField("query", "STRING", mode="NULLABLE"),
    bigquery.SchemaField("device", "STRING", mode="NULLABLE"),
    bigquery.SchemaField("country", "STRING", mode="NULLABLE"),
    bigquery.SchemaField("clicks", "FLOAT", mode="REQUIRED"),
    bigquery.SchemaField("impressions", "FLOAT", mode="REQUIRED"),
    bigquery.SchemaField("ctr", "FLOAT", mode="REQUIRED"),
    bigquery.SchemaField("position", "FLOAT", mode="REQUIRED"),
    bigquery.SchemaField("fetched_at", "TIMESTAMP", mode="REQUIRED"),
]


def get_table_id(project_id: str, dataset_id: str) -> str:
    """BigQueryテーブルの完全修飾IDを返す。"""
    return f"{project_id}.{dataset_id}.{TABLE_NAME}"


def ensure_gsc_table(
    client: bigquery.Client,
    project_id: str,
    dataset_id: str,
) -> str:
    """GSC保存用テーブルが存在しない場合は作成する。"""
    table_id = get_table_id(project_id, dataset_id)

    table = bigquery.Table(
        table_id,
        schema=GSC_TABLE_SCHEMA,
    )

    table.time_partitioning = bigquery.TimePartitioning(
        type_=bigquery.TimePartitioningType.DAY,
        field="data_date",
    )

    table.clustering_fields = [
        "site_id",
        "page",
    ]

    client.create_table(
        table,
        exists_ok=True,
    )

    return table_id


def replace_gsc_rows(
    project_id: str,
    dataset_id: str,
    site_id: str,
    start_date: date,
    end_date: date,
    rows: list[dict[str, Any]],
) -> int:
    """
    指定サイト・期間の既存データを削除し、
    新しく取得したデータへ置き換える。
    """
    client = bigquery.Client(project=project_id)

    table_id = ensure_gsc_table(
        client=client,
        project_id=project_id,
        dataset_id=dataset_id,
    )

    delete_sql = f"""
        DELETE FROM `{table_id}`
        WHERE site_id = @site_id
          AND data_date BETWEEN @start_date AND @end_date
    """

    delete_config = bigquery.QueryJobConfig(
        query_parameters=[
            bigquery.ScalarQueryParameter(
                "site_id",
                "STRING",
                site_id,
            ),
            bigquery.ScalarQueryParameter(
                "start_date",
                "DATE",
                start_date,
            ),
            bigquery.ScalarQueryParameter(
                "end_date",
                "DATE",
                end_date,
            ),
        ]
    )

    client.query(
        delete_sql,
        job_config=delete_config,
        location=BQ_LOCATION,
    ).result()

    if not rows:
        return 0

    fetched_at = datetime.now(timezone.utc).isoformat()

    load_rows = [
        {
            **row,
            "fetched_at": fetched_at,
        }
        for row in rows
    ]

    load_config = bigquery.LoadJobConfig(
        schema=GSC_TABLE_SCHEMA,
        write_disposition=bigquery.WriteDisposition.WRITE_APPEND,
    )

    load_job = client.load_table_from_json(
        load_rows,
        table_id,
        job_config=load_config,
        location=BQ_LOCATION,
    )

    load_job.result()

    return len(load_rows)
EOF

続けて文法チェックを実行します。

python3 -m py_compile loaders/bigquery.py

何も表示されずプロンプトに戻れば正常です。さらに、ファイルの冒頭を確認してください。

sed -n '1,60p' loaders/bigquery.py

この段階ではBigQueryテーブルはまだ作成されません。実際にプログラムを実行したとき、自動的に gsc_query_page_daily テーブルが作成されます。

全体をつなぐ main.py を作る

今回は、4サイトを順番に処理し、GSC取得 → BigQuery保存まで実行する入口を作ります。

Cloud Shellに次をそのまま貼り付けてください。

cat > main.py <<'EOF'
import logging
import sys

from collectors.gsc import fetch_gsc_rows
from common.date_utils import get_gsc_date_range
from common.settings import load_settings
from loaders.bigquery import replace_gsc_rows


logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s %(levelname)s %(message)s",
)


def main() -> int:
    config = load_settings()
    settings = config["settings"]
    sites = config["sites"]

    project_id = settings["project_id"]
    dataset_id = settings["bigquery_dataset"]
    timezone = settings.get("timezone", "Asia/Tokyo")
    refresh_days = int(settings.get("gsc_refresh_days", 7))

    start_date, end_date = get_gsc_date_range(
        refresh_days=refresh_days,
        data_delay_days=2,
        timezone=timezone,
    )

    logging.info(
        "GSC取得開始: %s ～ %s、対象サイト数=%s",
        start_date,
        end_date,
        len(sites),
    )

    failed_sites: list[str] = []

    for site in sites:
        site_id = site["site_id"]
        site_url = site["gsc_property"]

        try:
            logging.info(
                "サイト処理開始: site_id=%s, property=%s",
                site_id,
                site_url,
            )

            rows = fetch_gsc_rows(
                site_id=site_id,
                site_url=site_url,
                start_date=start_date,
                end_date=end_date,
            )

            loaded_count = replace_gsc_rows(
                project_id=project_id,
                dataset_id=dataset_id,
                site_id=site_id,
                start_date=start_date,
                end_date=end_date,
                rows=rows,
            )

            logging.info(
                "サイト処理完了: site_id=%s, 取得件数=%s, 保存件数=%s",
                site_id,
                len(rows),
                loaded_count,
            )

        except Exception:
            logging.exception(
                "サイト処理失敗: site_id=%s, property=%s",
                site_id,
                site_url,
            )
            failed_sites.append(site_id)

    if failed_sites:
        logging.error(
            "一部サイトで失敗しました: %s",
            ", ".join(failed_sites),
        )
        return 1

    logging.info("全サイトの処理が正常に完了しました。")
    return 0


if __name__ == "__main__":
    sys.exit(main())
EOF

続けて文法チェックを実行します。

python3 -m py_compile main.py

何も表示されなければ正常です。さらに、関連ファイルをまとめてチェックします。

python3 -m py_compile \
  main.py \
  collectors/gsc.py \
  loaders/bigquery.py \
  common/settings.py \
  common/date_utils.py

ここでも何も表示されず、入力待ちに戻れば問題ありません。

今回はまだ python3 main.py は実行しないでください。Cloud Shellのユーザー認証ではなく、最終的にCloud Run上の blog-analytics-runner で検証するためです。

Cloud Run Job用の実行設定を作る

Cloud Run Jobsへソースからデプロイする場合、Pythonの起動方法を明示しておくと確実です。今回はProcfileを1つ作ります。Cloud Runのソースデプロイでは、Cloud BuildとBuildpacksが自動的にコンテナを作成します。

Cloud Shellで次を実行してください。

cat > Procfile <<'EOF'
web: python main.py
EOF

続けて内容を確認します。

cat Procfile

次の1行が表示されれば完了です。

web: python main.py

さらに、.gcloudignoreに不要ファイルを除外する設定を入れます。

cat > .gcloudignore <<'EOF'
.gcloudignore
.git
.gitignore
__pycache__/
*.pyc
tests/
EOF

確認します。

cat .gcloudignore

今回はまだCloud Runへデプロイしません。

第2回のまとめ

これで、Search Console APIから4サイト分のデータを取得し、BigQueryへ保存するPythonプログラムが完成しました。ただし、まだCloud Shell上にファイルがあるだけで、毎朝自動で動く状態ではありません。

次回は、このプログラムをCloud Run Jobsへデプロイし、Cloud Schedulerから毎朝5時に起動します。再実行してもデータが二重登録されないことまで確認します。

第3回：Cloud Run JobsとCloud SchedulerでGSCデータを毎朝自動取得する

スポンサーリンク

コメント