Step 2 ・ Lv2 ・目安 90 分

Step 2 SERP 順位監視パイプライン

SERP API で日次クエリ → CSV / SQLite に保存 → 変動検知アラートまでの最小パイプラインを作る。

リード

このページを完了すると、Bright Data の SERP API を使って毎日キーワードの検索順位を取得し、結果を CSV と SQLite に保存、前日との差分で順位変動（±3 位以上）を検知して Slack へアラートを送信、さらに cron または GitHub Actions で自動実行できるパイプラインが構築できます。

SERP 順位監視パイプライン — キーワード → SERP API → SQLite → 差分検知 → Slack/メール通知の日次バッチ全体像

ゴールと所要時間

ゴール：SERP API で日次クエリを投げ、結果を CSV / SQLite に保存し、前日差分で順位変動（±3 位以上）をアラート、cron / GitHub Actions で日次実行
所要時間：約 90 分
難易度：Lv2（基礎的なシェル・Python・SQL の知識があれば実装可能）

前提

Step 1 が完了し、Bright Data の認証情報（$BRIGHTDATA_API_KEY）が環境変数に設定されていること
SERP API 用ゾーン（例：serp_api1）が作成済みであること
監視したい キーワード 5〜10 個 が決まっていること（例：["東京天気", "東京レストラン", "東京観光"]）

詳細は /hands-on/step-1-setup を参照してください。

手順 1：SERP API を curl で 1 クエリ叩く（Google 検索・日本語・location=Tokyo）

curl -X POST "https://api.brightdata.com/request" \
  -H "Authorization: Bearer $BRIGHTDATA_API_KEY" \
  -H "Content-Type: application/json" \
  -d '{
        "zone": "serp_api1",
        "payload": {
          "engine": "google",
          "q": "東京 天気",
          "language": "ja",
          "location": "Tokyo",
          "num": 10
        }
      }' \
  -o response.json

response.json に API の生データが保存されます。
Node.js でも同様にリクエリできます（参考例は後述）。

手順 2：Python で複数キーワードをループ、結果を pandas DataFrame 化

# file: fetch_serp.py
import os, json, time, sqlite3
import requests
import pandas as pd
 
API_KEY = os.getenv("BRIGHTDATA_API_KEY")
ZONE = "serp_api1"
KEYWORDS = ["東京 天気", "東京 レストラン", "東京 観光"]  # 必要に応じて増減
 
def fetch_keyword(keyword: str) -> pd.DataFrame:
    payload = {
        "zone": ZONE,
        "payload": {
            "engine": "google",
            "q": keyword,
            "language": "ja",
            "location": "Tokyo",
            "num": 10
        }
    }
    headers = {
        "Authorization": f"Bearer {API_KEY}",
        "Content-Type": "application/json"
    }
    resp = requests.post("https://api.brightdata.com/request", headers=headers, json=payload)
    resp.raise_for_status()
    data = resp.json()
    rows = []
    for org in data.get("organic", []):
        rows.append({
            "keyword": keyword,
            "url": org["link"],
            "title": org["title"],
            "position": org["rank"],
            "snapshot_at": pd.Timestamp.utcnow()
        })
    return pd.DataFrame(rows)
 
def main():
    dfs = [fetch_keyword(k) for k in KEYWORDS]
    result = pd.concat(dfs, ignore_index=True)
    # CSV 保存
    result.to_csv("serp_snapshot.csv", index=False, encoding="utf-8")
    # SQLite 保存（次の手順で使用）
    result.to_sql("ranks", con=sqlite3.connect("serp.db"), if_exists="append", index=False)
 
if __name__ == "__main__":
    main()

必要なパッケージは pip install requests pandas でインストールできます。
各キーワードの結果が serp_snapshot.csv と SQLite データベース serp.db に保存されます。

手順 3：SQLite に `ranks(keyword, url, position, snapshot_at)` テーブル作成・書き込み

-- file: init_schema.sql
CREATE TABLE IF NOT EXISTS ranks (
    id INTEGER PRIMARY KEY AUTOINCREMENT,
    keyword TEXT NOT NULL,
    url TEXT NOT NULL,
    position INTEGER NOT NULL,
    snapshot_at DATETIME NOT NULL
);

# テーブル作成
sqlite3 serp.db < init_schema.sql

Python の to_sql で自動的にテーブルが存在しない場合は作成されますが、上記のスキーマを明示的に管理すると運用が楽です。

手順 4：前日比較ロジック（SQL 差分クエリ）

-- file: diff_query.sql
WITH today AS (
    SELECT keyword, url, position
    FROM ranks
    WHERE DATE(snapshot_at) = DATE('now')
),
yesterday AS (
    SELECT keyword, url, position
    FROM ranks
    WHERE DATE(snapshot_at) = DATE('now', '-1 day')
)
SELECT
    t.keyword,
    t.url,
    y.position AS pos_yesterday,
    t.position AS pos_today,
    (t.position - y.position) AS diff
FROM today t
JOIN yesterday y ON t.keyword = y.keyword AND t.url = y.url
WHERE ABS(t.position - y.position) >= 3;

# file: detect_changes.py
import sqlite3, pandas as pd
 
conn = sqlite3.connect("serp.db")
df = pd.read_sql_query(open("diff_query.sql").read(), conn)
if not df.empty:
    print("⚠️ 順位変動が検知されました")
    print(df)
else:
    print("✅ 前日比で大きな変動はありません")

diff が正負どちらでも 3 位以上の変動が対象です。
このクエリは毎日実行し、結果が空でなければアラートへ渡します。

手順 5：Slack Webhook またはメールでアラート送信

Slack Webhook 例（Python）

# file: alert_slack.py
import os, json, requests, pandas as pd
 
WEBHOOK_URL = os.getenv("SLACK_WEBHOOK_URL")  # 事前に設定
 
def send_slack(df: pd.DataFrame):
    if df.empty:
        return
    blocks = [
        {"type": "section", "text": {"type": "mrkdwn", "text": "*SERP 順位変動アラート*"}},
        {"type": "divider"}
    ]
    for _, row in df.iterrows():
        blocks.append({
            "type": "section",
            "text": {"type": "mrkdwn",
                     "text": f"`{row['keyword']}` の `{row['url']}` が *{row['pos_yesterday']} 位* → *{row['pos_today']} 位* (Δ {row['diff']})"}
        })
    payload = {"blocks": blocks}
    requests.post(WEBHOOK_URL, json=payload)
 
if __name__ == "__main__":
    import sqlite3
    conn = sqlite3.connect("serp.db")
    df = pd.read_sql_query(open("diff_query.sql").read(), conn)
    send_slack(df)

メール送信例（Python + smtplib）

import os, smtplib, ssl
from email.mime.text import MIMEText
from email.mime.multipart import MIMEMultipart
import pandas as pd, sqlite3
 
SMTP_HOST = os.getenv("SMTP_HOST")
SMTP_PORT = int(os.getenv("SMTP_PORT", "587"))
SMTP_USER = os.getenv("SMTP_USER")
SMTP_PASS = os.getenv("SMTP_PASS")
TO_ADDR = os.getenv("ALERT_EMAIL")
 
def send_mail(df: pd.DataFrame):
    if df.empty:
        return
    msg = MIMEMultipart()
    msg["Subject"] = "SERP 順位変動アラート"
    msg["From"] = SMTP_USER
    msg["To"] = TO_ADDR
 
    html = "<h3>SERP 順位変動アラート</h3><table border='1' cellpadding='5'><tr><th>Keyword</th><th>URL</th><th>前日</th><th>本日</th><th>Δ</th></tr>"
    for _, r in df.iterrows():
        html += f"<tr><td>{r['keyword']}</td><td>{r['url']}</td><td>{r['pos_yesterday']}</td><td>{r['pos_today']}</td><td>{r['diff']}</td></tr>"
    html += "</table>"
    msg.attach(MIMEText(html, "html"))
 
    ctx = ssl.create_default_context()
    with smtplib.SMTP(SMTP_HOST, SMTP_PORT) as server:
        server.starttls(context=ctx)
        server.login(SMTP_USER, SMTP_PASS)
        server.send_message(msg)
 
if __name__ == "__main__":
    conn = sqlite3.connect("serp.db")
    df = pd.read_sql_query(open("diff_query.sql").read(), conn)
    send_mail(df)

環境変数で認証情報や webhook URL を管理し、コードにシークレットを書かないようにしてください。

手順 6：cron / GitHub Actions で日次実行するサンプル YAML

cron（Linux）

# /etc/cron.d/serp_monitor
0 6 * * *   /usr/bin/python3 /path/to/fetch_serp.py >> /var/log/serp_fetch.log 2>&1
15 6 * * *  /usr/bin/python3 /path/to/detect_changes.py >> /var/log/serp_diff.log 2>&1
30 6 * * *  /usr/bin/python3 /path/to/alert_slack.py >> /var/log/serp_alert.log 2>&1

毎日 06:00 に取得 → 差分計算 → Slack アラートの順に実行します。

GitHub Actions

# .github/workflows/serp-monitor.yml
name: SERP 監視パイプライン
 
on:
  schedule:
    - cron: '0 6 * * *'   # UTC 06:00（日本時間 15:00）
 
jobs:
  monitor:
    runs-on: ubuntu-latest
    env:
      BRIGHTDATA_API_KEY: ${{ secrets.BRIGHTDATA_API_KEY }}
      SLACK_WEBHOOK_URL: ${{ secrets.SLACK_WEBHOOK_URL }}
    steps:
      - uses: actions/checkout@v3
      - name: Python 環境設定
        uses: actions/setup-python@v5
        with:
          python-version: "3.11"
      - name: 依存パッケージインストール
        run: pip install -r requirements.txt
      - name: データ取得
        run: python fetch_serp.py
      - name: 変動検知
        run: python detect_changes.py
      - name: Slack アラート送信
        run: python alert_slack.py

requirements.txt には requests pandas が含まれます。
シークレットは GitHub の Settings → Secrets に登録してください。

運用上の注意

項目	内容
コスト見積	SERP API はリクエリ数 × 料金です。5 キーワード × 10 件取得で 1 日 50 リクエリ前想定。月額約 $X（実際はプランに依存）。
レート制御	1 秒あたりのリクエリ上限はゾーン設定により異なります。`time.sleep(1)` 等で間隔を調整してください。
失敗時リトライ	`requests` の `Retry` 機能や `curl --retry` を利用し、最大 3 回リトライしてください。
データ保持	SQLite はファイルサイズが肥大化しやすいので、古いスナップショットは定期的に削除（例：30 日以上前）しましょう。
モニタリング	GitHub Actions の `workflow_run` イベントで失敗通知を Slack に転送すると、障害検知が容易になります。

次に読む

/hands-on/step-3-crawl-rag – 取得した URL をクロールし、RAG 用データセットを作成する手順
/usecases/UC-SERP-001 – SERP 監視の実装例とベストプラクティス
/basics/operations – データパイプライン運用に役立つモニタリング・ロギングの基本