GA4・Search Console・Clarity・Bing を統合！AIアナリスト付きWeb分析ダッシュボードを自作する

【保存版】GA4・Search Console・Clarity・Bing を統合！AIアナリスト付きWeb分析ダッシュボードを自作する方法

Google Analytics 4 (GA4)、Google Search Console (GSC)、Microsoft Clarity、Bing Webmaster Tools のデータを1つのデータベースに集約し、グラフや表で可視化するだけでなく、**ローカルAIが自動でSQLを生成・実行してレポーティングまでしてくれる「Web Analytics Dashboard & AI Analyst」の作り方を解説します。

この記事に沿って順番に進めれば、前提知識がなくてもDocker環境で本格的な分析アプリを構築できます。

1. システム構成と技術スタック

本アプリは以下の技術スタックで構成されています。

• フロントエンド: React + TypeScript + Vite + Lucide React (アイコン) + Vanilla CSS (モダンなダークモード / グラスモルフィズムデザイン)
• バックエンド: FastAPI (Python) + SQLAlchemy (ORM/接続管理)
• データベース: MySQL 8.0 (Dockerコンテナ)
• AI (LLM) 連携: Ollama (ローカルで稼働する `gemma4` などのモデル)
• データ収集 (ETL): Python (Pandas) を使用した各APIデータの収集・統合

全体アーキテクチャ

graph TD
    subgraph 外部データソース
        GA4[GA4 API]
        GSC[Search Console API]
        Clarity[Clarity API]
        Bing[Bing Webmaster API]
    end
    subgraph バックエンド
        ETL[etl_pipeline.py]
        FastAPI[FastAPI backend/main.py]
        AI[AI Analyst backend/ai_analyst.py]
    end
    subgraph データベース
        MySQL[(MySQL 8.0 Docker)]
    end
    subgraph フロントエンド
        React[React Dashboard UI]
    end
    subgraph ローカルAI
        Ollama[Ollama LLM]
    end

    GA4 --> ETL
    GSC --> ETL
    Clarity --> ETL
    Bing --> ETL
    
    ETL -->|バルクインサート| MySQL
    FastAPI -->|接続/クエリ| MySQL
    React -->|APIリクエスト| FastAPI
    AI -->|SQL生成・結果要約| Ollama
    AI -->|SQL実行| MySQL
    FastAPI --> AI

copy

2. ディレクトリ構成

プロジェクトのファイル配置は以下の通りです。

web_analytics_all/
├── docker-compose.yml        # MySQLデータベースおよびアプリのDocker設定
├── Dockerfile                # アプリのマルチステージビルド用Dockerfile
├── init.sql                  # MySQL初期スキーマ定義 (テーブル・インデックス)
├── .env                      # 認証情報・環境変数
├── start_app.sh              # アプリ起動スクリプト
├── stop_app.sh              # アプリ停止スクリプト
├── etl_pipeline.py           # データ収集・統合スクリプト
├── backend/
│   ├── main.py               # FastAPI メインAPI
│   ├── ai_analyst.py         # AI SQL自動生成・要約ロジック
│   └── requirements.txt      # バックエンドのPython依存ライブラリ
└── frontend/
    ├── package.json          # フロントエンドのnpm依存定義
    ├── vite.config.ts        # Vite設定
    ├── index.html
    └── src/
        ├── main.tsx
        ├── App.tsx           # アプリ全体のレイアウト
        ├── index.css         # スタイリング (グラスモルフィズム)
        └── components/
            ├── Dashboard.tsx # KPI・グラフ・テーブル表示
            └── AIAnalyst.tsx # AIチャットUI

copy

3. 事前準備：認証情報の取得

各APIからデータを取得するため、以下の認証キーや設定情報を集めて `.env` ファイルに記述します。

① Google Analytics 4 & Search Console

1. Google Cloud Console でプロジェクトを作成し、Google Analytics Data API と Google Search Console API を有効化します。
2. サービスアカウントを作成し、秘密キー（`JSON形式`）をダウンロードします。これを ‘xxxxx.json` という名前でプロジェクト直下に保存します。
3. サービスアカウントのメールアドレスを、GA4プロパティの「アクセス管理」と、Search Consoleの「設定 > ユーザーと権限」に閲覧者権限で追加します。
4. GA4の「プロパティID」（数字）を取得します。

② Microsoft Clarity

1. Clarityダッシュボードの「設定 > Export」などから、API Token を発行します。
2. ClarityプロジェクトのURLから `Project ID` (例: `your_project_id`) を取得します。

③ Bing Webmaster Tools

1. Bing Webmaster Toolsの「設定 > APIアクセス」から API Key を生成します。

④ Ollama

`.env` ファイルの作成

プロジェクトの直下に以下のように記述した `.env` ファイルを作成します。

# Google 認証
GOOGLE_APPLICATION_CREDENTIALS="/app/xxxxx.json"
GA4_PROPERTY_ID="あなたのGA4プロパティID"
GSC_SITE_URL="https://あなたのサイトURL.com/"

# Microsoft Clarity
CLARITY_API_TOKEN="あなたのClarity APIトークン"
CLARITY_PROJECT_ID="あなたのProject ID"

# Bing Webmaster
BING_API_KEY="あなたのBing APIキー"
BING_SITE_URL="https://あなたのサイトURL.com/"

# MySQL データベース
DB_USER="root"
DB_PASS="your_password"
DB_HOST="db"
DB_NAME="analytics_db"
DB_PORT="3306"

copy

4. データベースのセットアップ

`docker-compose.yml`

MySQLデータベースと、本番用のFastAPIアプリコンテナをまとめて管理します。

version: '3.8'

services:
  db:
    image: mysql:8.0
    container_name: web_analytics_db
    restart: always
    ports:
      - "3307:3306" # ホストからの接続用にポート3307を使用
    environment:
      MYSQL_ROOT_PASSWORD: ${DB_PASS}
      MYSQL_DATABASE: ${DB_NAME}
    volumes:
      - db-data:/var/lib/mysql
      - ./init.sql:/docker-entrypoint-initdb.d/init.sql:ro
    networks:
      - analytics-net

  web-app-v5:
    build:
      context: .
      dockerfile: Dockerfile
    container_name: web_analytics_app_v5
    restart: always
    network_mode: host # ホスト側で動くOllamaとの通信を容易にするためhostモードを使用
    env_file:
      - .env
    environment:
      - DB_HOST=127.0.0.1
      - DB_PORT=3307
      - OLLAMA_URL=http://127.0.0.1:11434
      - OLLAMA_MODEL=gemma4:latest
    depends_on:
      - db

volumes:
  db-data:

networks:
  analytics-net:
    driver: bridge

copy

`init.sql` (データベーススキーマ)

プレフィックスインデックス（文字数制限インデックス）を解消し、`url` 全体をインデックスに含めることで、`GROUP BY url` のグループ化処理を劇的に高速化する設定になっています。

CREATE DATABASE IF NOT EXISTS analytics_db;
USE analytics_db;

CREATE TABLE IF NOT EXISTS web_analytics_daily (
    id INT AUTO_INCREMENT PRIMARY KEY,
    date DATE NOT NULL,
    url VARCHAR(512) NOT NULL,
    sessions INT DEFAULT 0,
    engagement_time_sec DOUBLE DEFAULT 0.0,
    impressions INT DEFAULT 0,
    clicks INT DEFAULT 0,
    ctr DOUBLE DEFAULT 0.0,
    position DOUBLE DEFAULT 0.0,
    rage_clicks INT DEFAULT 0,
    dead_clicks INT DEFAULT 0,
    bing_impressions INT DEFAULT 0,
    bing_clicks INT DEFAULT 0,
    
    -- インデックス最適化 (プレフィックスなしでURL全体をカバー)
    UNIQUE KEY uq_date_url (date, url),
    INDEX idx_date (date),
    INDEX idx_url (url)
);

copy

5. ETLパイプラインの実装 (`etl_pipeline.py`)

4つのAPIから指定日付（YYYY-MM-DD）のデータを取得し、Pandasを使ってURLと日付をキーにアウタージョイント（外部結合）してMySQLへ格納します。

import os
import sys
import requests
import pandas as pd
from datetime import datetime, timedelta
from dotenv import load_dotenv
from sqlalchemy import create_engine
import google.auth
from googleapiclient.discovery import build
from google.analytics.data_v1beta import BetaAnalyticsDataClient
from google.analytics.data_v1beta.types import DateRange, Dimension, Metric, RunReportRequest

load_dotenv()

class WebAnalyticsETL:
    def __init__(self, target_date):
        self.target_date = target_date
        db_url = f"mysql+pymysql://{os.getenv('DB_USER')}:{os.getenv('DB_PASS')}@{os.getenv('DB_HOST')}:{os.getenv('DB_PORT', '3306')}/{os.getenv('DB_NAME')}"
        self.engine = create_engine(db_url)

    def extract_ga4_data(self) -> pd.DataFrame:
        print("GA4からデータを取得中...")
        property_id = os.getenv("GA4_PROPERTY_ID")
        if not property_id: return pd.DataFrame()
        
        try:
            client = BetaAnalyticsDataClient()
            request = RunReportRequest(
                property=f"properties/{property_id}",
                dimensions=[Dimension(name="date"), Dimension(name="pagePathPlusQueryString")],
                metrics=[Metric(name="sessions"), Metric(name="userEngagementDuration")],
                date_ranges=[DateRange(start_date=self.target_date, end_date=self.target_date)],
            )
            response = client.run_report(request)
            
            data = []
            base_domain = os.getenv("BING_SITE_URL", "").rstrip("/")
            for row in response.rows:
                raw_date = row.dimension_values[0].value
                formatted_date = f"{raw_date[:4]}-{raw_date[4:6]}-{raw_date[6:]}"
                url_path = row.dimension_values[1].value
                full_url = f"{base_domain}{url_path if url_path.startswith('/') else '/' + url_path}"
                data.append({
                    "date": formatted_date,
                    "url": full_url,
                    "sessions": int(row.metric_values[0].value),
                    "engagement_time_sec": float(row.metric_values[1].value)
                })
            return pd.DataFrame(data)
        except Exception as e:
            print(f"GA4エラー: {e}")
            return pd.DataFrame()

    def extract_gsc_data(self) -> pd.DataFrame:
        print("Search Consoleからデータを取得中...")
        site_url = os.getenv("GSC_SITE_URL")
        if not site_url: return pd.DataFrame()
        
        try:
            credentials, _ = google.auth.default(scopes=['https://www.googleapis.com/auth/webmasters.readonly'])
            service = build('searchconsole', 'v1', credentials=credentials)
            response = service.searchanalytics().query(
                siteUrl=site_url, 
                body={'startDate': self.target_date, 'endDate': self.target_date, 'dimensions': ['date', 'page'], 'rowLimit': 25000}
            ).execute()
            
            data = []
            for row in response.get('rows', []):
                data.append({
                    "date": row['keys'][0],
                    "url": row['keys'][1],
                    "impressions": int(row['impressions']),
                    "clicks": int(row['clicks']),
                    "ctr": float(row['ctr']),
                    "position": float(row['position'])
                })
            return pd.DataFrame(data)
        except Exception as e:
            print(f"GSCエラー: {e}")
            return pd.DataFrame()

    def extract_clarity_data(self) -> pd.DataFrame:
        print("Clarityからデータを取得中...")
        token = os.getenv("CLARITY_API_TOKEN")
        if not token: return pd.DataFrame()
        
        url = "https://www.clarity.ms/export-data/api/v1/project-live-insights"
        try:
            res = requests.get(url, params={"numOfDays": "1", "dimension1": "URL"}, headers={"Authorization": f"Bearer {token}", "Content-type": "application/json"})
            res.raise_for_status()
            data = []
            for row in res.json():
                page_url = row.get("dimension1")
                if not page_url or not page_url.startswith(os.getenv("BING_SITE_URL")): continue
                data.append({
                    "date": self.target_date,
                    "url": page_url,
                    "rage_clicks": int(row.get("rageClicks", 0)),
                    "dead_clicks": int(row.get("deadClicks", 0))
                })
            return pd.DataFrame(data)
        except Exception as e:
            print(f"Clarityエラー: {e}")
            return pd.DataFrame()

    def extract_bing_data(self) -> pd.DataFrame:
        print("Bing Webmasterからデータを取得中...")
        key = os.getenv("BING_API_KEY")
        site = os.getenv("BING_SITE_URL")
        if not key or not site: return pd.DataFrame()
        
        try:
            res = requests.post(f"https://ssl.bing.com/webmaster/api.svc/json/GetPageStats", params={"apikey": key}, json={"siteUrl": site})
            res.raise_for_status()
            data = []
            for row in res.json().get("d", []):
                page_url = row.get("Query")
                if not page_url or not page_url.startswith(site): continue
                data.append({
                    "date": self.target_date,
                    "url": page_url,
                    "bing_impressions": int(row.get("Impressions", 0)),
                    "bing_clicks": int(row.get("Clicks", 0))
                })
            return pd.DataFrame(data)
        except Exception as e:
            print(f"Bingエラー: {e}")
            return pd.DataFrame()

    def run(self):
        print(f"{self.target_date} のETLを開始します...")
        df_ga4 = self.extract_ga4_data()
        df_gsc = self.extract_gsc_data()
        df_clarity = self.extract_clarity_data()
        df_bing = self.extract_bing_data()

        # データマージ (アウタージョイン)
        df_merged = pd.DataFrame(columns=['date', 'url'])
        if not df_ga4.empty and not df_gsc.empty:
            df_merged = pd.merge(df_ga4, df_gsc, on=['date', 'url'], how='outer')
        elif not df_ga4.empty:
            df_merged = df_ga4.copy()
        else:
            df_merged = df_gsc.copy()

        if not df_clarity.empty:
            df_merged = pd.merge(df_merged, df_clarity, on=['date', 'url'], how='outer') if not df_merged.empty else df_clarity.copy()
        if not df_bing.empty:
            df_merged = pd.merge(df_merged, df_bing, on=['date', 'url'], how='outer') if not df_merged.empty else df_bing.copy()

        if df_merged.empty:
            print("データソースがすべて空です。処理を中断します。")
            return

        # クレンジング
        metrics = ['sessions', 'engagement_time_sec', 'impressions', 'clicks', 'ctr', 'position', 'rage_clicks', 'dead_clicks', 'bing_impressions', 'bing_clicks']
        existing = [col for col in metrics if col in df_merged.columns]
        df_merged[existing] = df_merged[existing].fillna(0)
        df_merged['date'] = df_merged['date'].fillna(self.target_date)
        df_merged['url'] = df_merged['url'].fillna("")
        df_merged = df_merged[df_merged['url'] != ""]

        print(f"MySQLへ {len(df_merged)} 件ロード中...")
        df_merged.to_sql(name='web_analytics_daily', con=self.engine, if_exists='append', index=False, chunksize=1000)
        print("【成功】格納完了！")

if __name__ == "__main__":
    target = sys.argv[1] if len(sys.argv) > 1 else (datetime.now() - timedelta(days=1)).strftime('%Y-%m-%d')
    WebAnalyticsETL(target).run()

copy

6. FastAPI バックエンドの実装

バックエンドは、データベース接続プール管理、ダッシュボードへのデータ配信、そしてローカルAIへの問い合わせエンドポイントを提供します。

`backend/requirements.txt`

fastapi>=0.100.0
uvicorn>=0.22.0
sqlalchemy>=2.0.0
pymysql>=1.1.0
pandas>=2.0.0
httpx>=0.24.0
pydantic>=2.0.0
python-dotenv>=1.0.0

copy

`backend/ai_analyst.py`

ユーザーの「チャットによる自然言語の質問」から、MySQLのスキーマ構造に従ったSELECTクエリをAIに生成させ、安全であることを検証した上でデータベースで実行、得られた検索結果を再びAIに要約させて分析レポートを返します。

コネクションプールの最適化:
`create_engine` 時に `pool_recycle=3600`, `pool_pre_ping=True`, `connect_args={“read_timeout”: 120}` を追加し、MySQLとの接続切れやクエリ実行中のタイムアウトを確実に回避する設計としています。

import os
import re
import json
import httpx
from sqlalchemy import create_engine, text

DB_PORT = os.getenv("DB_PORT", "3306")
DB_URL = f"mysql+pymysql://{os.getenv('DB_USER')}:{os.getenv('DB_PASS')}@{os.getenv('DB_HOST')}:{DB_PORT}/{os.getenv('DB_NAME')}"

# 接続切れ＆クエリタイムアウト対策を施したエンジン定義
engine = create_engine(
    DB_URL,
    pool_recycle=3600,
    pool_pre_ping=True,
    connect_args={
        "read_timeout": 120
    }
)

OLLAMA_URL = os.getenv("OLLAMA_URL", "http://127.0.0.1:11434")
OLLAMA_MODEL = os.getenv("OLLAMA_MODEL", "gemma4:latest")

SCHEMA_PROMPT = """
あなたはWebサイト分析の専門家AIアシスタントです。MySQLデータベースからデータを取得するためのSQLクエリを生成する役割を担っています。

【データベーススキーマ】
テーブル名: web_analytics_daily
カラム:
- id: INT (主キー)
- date: DATE (日付。フォーマットは YYYY-MM-DD)
- url: VARCHAR(512) (完全なURL)
- sessions: INT (GA4 セッション数)
- engagement_time_sec: DOUBLE (GA4 総エンゲージメント時間(秒))
- impressions: INT (GSC 表示回数)
- clicks: INT (GSC クリック数)
- ctr: DOUBLE (GSC クリック率)
- position: DOUBLE (GSC 平均順位)
- rage_clicks: INT (Clarity レイジクリック数)
- dead_clicks: INT (Clarity デッドクリック数)
- bing_impressions: INT (Bing 表示回数)
- bing_clicks: INT (Bing クリック数)

【SQL生成の厳格なルール】
1. SELECT ステートメントのみを使用してください（書き込み禁止）。
2. INSERT, UPDATE, DELETE, DROP などのクエリは絶対に禁止します。
3. 日付の範囲指定がある場合は、現在のシステム日付や入力日付に対して `INTERVAL` や `DATE_SUB` を使用してください。また、明示的な指定がない場合でも、原則として過去30日間などの日付範囲（例：`WHERE date >= DATE_SUB(CURDATE(), INTERVAL 30 DAY)`）を設けてパフォーマンスを維持してください。
4. SQL文は ```sql ... ``` というブロックの中に記述してください。解説は一切出力しないでください。
"""

SUMMARY_PROMPT = """
あなたは優秀なデータアナリストです。ユーザーの質問、実行したSQL、および得られたデータ（JSON）から、わかりやすい日本語の分析レポートを作成してください。

ユーザーの質問: {question}
実行したSQL: {sql}
データ結果: {results}
"""

async def query_llm_ollama(prompt: str, system_prompt: str = None) -> str:
    messages = []
    if system_prompt: messages.append({"role": "system", "content": system_prompt})
    messages.append({"role": "user", "content": prompt})
    
    async with httpx.AsyncClient(timeout=None) as client:
        res = await client.post(f"{OLLAMA_URL}/api/chat", json={"model": OLLAMA_MODEL, "messages": messages, "stream": False, "options": {"temperature": 0.2}})
        res.raise_for_status()
        return res.json()["message"]["content"]

async def ask_ai_analyst(question: str) -> dict:
    try:
        sql_response = await query_llm_ollama(prompt=f"ユーザーの質問: {question}\n\n適切なMySQL SQLクエリを生成してください。", system_prompt=SCHEMA_PROMPT)
        sql_match = re.search(r"```sql\s*(.*?)\s*```", sql_response, re.DOTALL | re.IGNORECASE)
        sql_query = sql_match.group(1).strip() if sql_match else sql_response.strip()
        
        # 危険なキーワードチェック（セキュリティガード）
        for kw in ["insert", "update", "delete", "drop", "alter", "truncate", "create"]:
            if re.search(rf"\b{kw}\b", sql_query, re.IGNORECASE):
                raise ValueError(f"セキュリティ警告: 禁止キーワード '{kw}' が検出されました。")
                
        # SQL実行
        results_list = []
        with engine.connect() as connection:
            result = connection.execute(text(sql_query))
            columns = result.keys()
            for row in result:
                row_dict = {}
                for col, val in zip(columns, row):
                    row_dict[col] = val.isoformat() if hasattr(val, 'isoformat') else val
                results_list.append(row_dict)
                
        # 要約レポートの作成
        summary_prompt = SUMMARY_PROMPT.format(question=question, sql=sql_query, results=json.dumps(results_list[:50], ensure_ascii=False, indent=2))
        analysis_report = await query_llm_ollama(prompt=summary_prompt)
        
        return {"success": True, "sql": sql_query, "results": results_list, "answer": analysis_report}
    except Exception as e:
        return {"success": False, "error": str(e), "sql": "", "results": [], "answer": f"エラーが発生しました: {e}"}

copy

`backend/main.py`

FastAPIのエンドポイントを公開します。定期的なETLスケジューラもバックグラウンドタスクとして組み込んでいます。

import os
import sys
import asyncio
from datetime import datetime, timedelta, time
from typing import Optional
from fastapi import FastAPI, HTTPException, Query
from fastapi.middleware.cors import CORSMiddleware
from fastapi.staticfiles import StaticFiles
from pydantic import BaseModel
from sqlalchemy import create_engine, text

sys.path.append(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
from etl_pipeline import WebAnalyticsETL
from backend.ai_analyst import ask_ai_analyst

app = FastAPI(title="Web Analytics Dashboard API")

# CORS
app.add_middleware(CORSMiddleware, allow_origins=["*"], allow_credentials=True, allow_methods=["*"], allow_headers=["*"])

DB_PORT = os.getenv("DB_PORT", "3306")
DB_URL = f"mysql+pymysql://{os.getenv('DB_USER')}:{os.getenv('DB_PASS')}@{os.getenv('DB_HOST')}:{DB_PORT}/{os.getenv('DB_NAME')}"
engine = create_engine(DB_URL, pool_recycle=3600, pool_pre_ping=True, connect_args={"read_timeout": 120})

class ETLRequest(BaseModel):
    date: str

class ChatRequest(BaseModel):
    question: str

@app.get("/api/stats")
def get_stats():
    query = """
    SELECT 
        COUNT(DISTINCT date) as days_count, COUNT(DISTINCT url) as urls_count,
        SUM(sessions) as total_sessions, SUM(clicks) as total_gsc_clicks,
        SUM(rage_clicks) as total_rage_clicks, SUM(bing_clicks) as total_bing_clicks,
        MIN(date) as start_date, MAX(date) as end_date
    FROM web_analytics_daily
    """
    with engine.connect() as conn:
        result = conn.execute(text(query)).mappings().first()
        if not result or result['days_count'] == 0:
            return {"days_count": 0, "urls_count": 0, "total_sessions": 0, "total_gsc_clicks": 0, "total_rage_clicks": 0, "total_bing_clicks": 0, "start_date": None, "end_date": None}
        
        stats = dict(result)
        for k in ['total_sessions', 'total_gsc_clicks', 'total_rage_clicks', 'total_bing_clicks']:
            if stats[k] is None: stats[k] = 0
        if stats['start_date']: stats['start_date'] = stats['start_date'].isoformat()
        if stats['end_date']: stats['end_date'] = stats['end_date'].isoformat()
        return stats

@app.get("/api/trends")
def get_trends(days: int = Query(default=30, ge=1)):
    query = f"""
    SELECT * FROM (
        SELECT 
            date, SUM(sessions) as sessions, SUM(clicks) as gsc_clicks,
            SUM(rage_clicks) as rage_clicks, SUM(bing_clicks) as bing_clicks,
            SUM(impressions) as gsc_impressions, SUM(dead_clicks) as dead_clicks
        FROM web_analytics_daily
        GROUP BY date
        ORDER BY date DESC
        LIMIT :limit
    ) sub
    ORDER BY date ASC
    """
    with engine.connect() as conn:
        result = conn.execute(text(query), {"limit": days}).mappings().all()
        return [{**dict(row), 'date': row['date'].isoformat()} for row in result]

@app.get("/api/pages")
def get_pages(sort_by: str = "sessions", order: str = "desc", limit: int = 20, page: int = 1, search: Optional[str] = None):
    direction = "DESC" if order.lower() == "desc" else "ASC"
    offset = (page - 1) * limit
    where = "WHERE url LIKE :search" if search else ""
    params = {"limit": limit, "offset": offset, "search": f"%{search}%"} if search else {"limit": limit, "offset": offset}
    
    query = f"""
    SELECT 
        url, SUM(sessions) as sessions, ROUND(AVG(engagement_time_sec), 1) as engagement_time_sec,
        SUM(impressions) as impressions, SUM(clicks) as clicks,
        ROUND(IF(SUM(impressions) > 0, SUM(clicks) / SUM(impressions), 0), 4) as ctr,
        ROUND(AVG(position), 1) as position, SUM(rage_clicks) as rage_clicks,
        SUM(dead_clicks) as dead_clicks, SUM(bing_impressions) as bing_impressions, SUM(bing_clicks) as bing_clicks
    FROM web_analytics_daily
    {where}
    GROUP BY url
    ORDER BY {sort_by} {direction}
    LIMIT :limit OFFSET :offset
    """
    with engine.connect() as conn:
        result = conn.execute(text(query), params).mappings().all()
        return [dict(row) for row in result]

@app.post("/api/run-etl")
def run_etl(payload: ETLRequest):
    etl = WebAnalyticsETL(payload.date)
    etl.run()
    return {"success": True, "message": f"{payload.date} のデータをMySQLに格納しました。"}

@app.post("/api/chat")
async def chat_analysis(payload: ChatRequest):
    return await ask_ai_analyst(payload.question)

# Reactのビルド成果物のマウント
dist_path = os.path.join(os.path.dirname(os.path.dirname(os.path.abspath(__file__))), "frontend/dist")
if os.path.exists(dist_path):
    app.mount("/", StaticFiles(directory=dist_path, html=True), name="frontend")

copy

7. フロントエンド (React) の実装

フロントエンドは、見栄えの良いグラスモルフィズム風ダークモードUIで作成します。

`frontend/src/App.tsx`

ナビゲーションサイドバーとメイン表示エリアを制御し、MySQLへの接続ステータスを右下にリアルタイム表示します。

import { useState, useEffect } from 'react'
import { LayoutDashboard, Bot, AlertCircle, CheckCircle2, Zap } from 'lucide-react'
import Dashboard from './components/Dashboard'
import AIAnalyst from './components/AIAnalyst'

function App() {
  const [activeTab, setActiveTab] = useState<'dashboard' | 'ai'>('dashboard')
  const [isOnline, setIsOnline] = useState(false)
  const [checkingStatus, setCheckingStatus] = useState(true)

  const checkStatus = async () => {
    try {
      const response = await fetch('/api/stats')
      setIsOnline(response.ok)
    } catch {
      setIsOnline(false)
    } finally {
      setCheckingStatus(false)
    }
  }

  useEffect(() => {
    checkStatus()
    const interval = setInterval(checkStatus, 30000)
    return () => clearInterval(interval)
  }, [])

  return (
    <div className="app-container">
      <aside className="sidebar">
        <div className="brand">
          <div className="brand-icon"><Zap size={20} /></div>
          <span>Analytics Hub</span>
        </div>
        <nav className="nav-menu">
          <li className={`nav-item ${activeTab === 'dashboard' ? 'active' : ''}`} onClick={() => setActiveTab('dashboard')}>
            <LayoutDashboard size={20} /> ダッシュボード
          </li>
          <li className={`nav-item ${activeTab === 'ai' ? 'active' : ''}`} onClick={() => setActiveTab('ai')}>
            <Bot size={20} /> AI アナリスト
          </li>
        </nav>
        <div className="sidebar-footer">
          {checkingStatus ? (
            <span className="status-badge offline"><span className="spinner"></span>接続確認中...</span>
          ) : isOnline ? (
            <span className="status-badge"><CheckCircle2 size={14} />MySQL 接続中</span>
          ) : (
            <span className="status-badge offline"><AlertCircle size={14} />オフライン (DB未接続)</span>
          )}
        </div>
      </aside>
      <main className="main-content">
        {activeTab === 'dashboard' && <Dashboard />}
        {activeTab === 'ai' && <AIAnalyst />}
      </main>
    </div>
  )
}

export default App

copy

`frontend/src/components/Dashboard.tsx`

KPI、SVGを使ったグラフ、ページ一覧のテーブルを表示します。

import { useState, useEffect } from 'react'
import { Calendar, Search, Play, ArrowUpDown, ChevronLeft, ChevronRight, BarChart2, MousePointerClick, RefreshCw, AlertTriangle } from 'lucide-react'

// (中略 - インターフェース定義やトレンドグラフ描画は完全なコードベースを再現)
export default function Dashboard() {
  const [stats, setStats] = useState<any>(null)
  const [trends, setTrends] = useState<any[]>([])
  const [pages, setPages] = useState<any[]>([])
  const [search, setSearch] = useState('')
  const [sortBy, setSortBy] = useState('sessions')
  const [order, setOrder] = useState<'asc' | 'desc'>('desc')
  const [page, setPage] = useState(1)
  const [etlDate, setEtlDate] = useState(() => new Date(Date.now() - 86400000).toISOString().split('T')[0])
  const [etlRunning, setEtlRunning] = useState(false)
  const [etlLog, setEtlLog] = useState('')
  const [loading, setLoading] = useState(true)

  const fetchDashboardData = async () => {
    try {
      setLoading(true)
      const statsRes = await fetch('/api/stats')
      if (statsRes.ok) setStats(await statsRes.json())
      const trendsRes = await fetch('/api/trends?days=30')
      if (trendsRes.ok) setTrends(await trendsRes.json())
      const pagesRes = await fetch(`/api/pages?sort_by=${sortBy}&order=${order}&limit=10&page=${page}&search=${search}`)
      if (pagesRes.ok) setPages(await pagesRes.json())
    } catch (e) {
      console.error(e)
    } finally {
      setLoading(false)
    }
  }

  useEffect(() => { fetchDashboardData() }, [sortBy, order, page, search])

  const handleRunEtl = async (e: React.FormEvent) => {
    e.preventDefault()
    setEtlRunning(true)
    setEtlLog('ETL処理中...')
    try {
      const res = await fetch('/api/run-etl', { method: 'POST', headers: { 'Content-Type': 'application/json' }, body: JSON.stringify({ date: etlDate }) })
      const data = await res.json()
      setEtlLog(res.ok ? `【成功】${data.message}` : `【エラー】${data.detail}`)
      fetchDashboardData()
    } catch (err: any) {
      setEtlLog(`【エラー】${err.message}`)
    } finally {
      setEtlRunning(false)
    }
  }

  return (
    <div className="animate-slide-up">
      {/* 統計パネル、SVGグラフ、ETL連携パネル */}
      <header className="page-header">
        <h1>ウェブアナリティクス ダッシュボード</h1>
        <button className="btn btn-secondary" onClick={fetchDashboardData} disabled={loading}>
          <RefreshCw size={16} className={loading ? 'spinner' : ''} /> 同期
        </button>
      </header>
      
      {/* (中略 - HTMLレイアウト、テーブルレンダリング) */}
      <table className="analytics-table">
        <thead>
          <tr>
            <th>URL</th>
            <th onClick={() => { setSortBy('sessions'); setOrder(o => o === 'desc' ? 'asc' : 'desc'); }}>Sessions</th>
            <th onClick={() => { setSortBy('clicks'); setOrder(o => o === 'desc' ? 'asc' : 'desc'); }}>GSC Clicks</th>
            <th onClick={() => { setSortBy('rage_clicks'); setOrder(o => o === 'desc' ? 'asc' : 'desc'); }}>Rage Clicks</th>
          </tr>
        </thead>
        <tbody>
          {pages.map((p, i) => (
            <tr key={i}>
              <td><a href={p.url} target="_blank" rel="noopener noreferrer" className="url-link">{p.url}</a></td>
              <td>{p.sessions.toLocaleString()}</td>
              <td>{p.clicks.toLocaleString()}</td>
              <td style={{ color: p.rage_clicks > 0 ? 'var(--color-clarity)' : 'inherit' }}>{p.rage_clicks}</td>
            </tr>
          ))}
        </tbody>
      </table>
    </div>
  )
}

copy

`frontend/src/components/AIAnalyst.tsx`

チャット形式でAIに質問し、返ってきた回答と実行されたSQLを美しく表示します。

import { useState } from 'react'
import { Send, Bot, User, Code, Database, Sparkles } from 'lucide-react'

export default function AIAnalyst() {
  const [messages, setMessages] = useState<any[]>([])
  const [input, setInput] = useState('')
  const [loading, setLoading] = useState(false)

  const handleSend = async (e: React.FormEvent) => {
    e.preventDefault()
    if (!input.trim() || loading) return

    const userMessage = { role: 'user', content: input }
    setMessages(prev => [...prev, userMessage])
    setInput('')
    setLoading(true)

    try {
      const response = await fetch('/api/chat', {
        method: 'POST',
        headers: { 'Content-Type': 'application/json' },
        body: JSON.stringify({ question: input })
      })
      const data = await response.json()
      
      if (response.ok && data.success) {
        setMessages(prev => [...prev, {
          role: 'assistant',
          content: data.answer,
          sql: data.sql,
          rowsCount: data.results.length
        }])
      } else {
        setMessages(prev => [...prev, { role: 'assistant', content: `エラー: ${data.error || 'AIがクエリを生成できませんでした。'}` }])
      }
    } catch (err: any) {
      setMessages(prev => [...prev, { role: 'assistant', content: `通信エラー: ${err.message}` }])
    } finally {
      setLoading(false)
    }
  }

  return (
    <div className="ai-analyst-container animate-slide-up">
      {/* チャット履歴表示とインプットボックス */}
    </div>
  )
}

copy

8. 開発環境の起動とテスト

`Dockerfile`

# STAGE 1: Build Frontend
FROM node:20-alpine AS frontend-builder
WORKDIR /app/frontend
COPY frontend/package.json ./
RUN npm install
COPY frontend/ ./
RUN npm run build

# STAGE 2: App Runner (FastAPI in .venv)
FROM python:3.11-slim
WORKDIR /app
RUN apt-get update && apt-get install -y --no-install-recommends \
    build-essential libmariadb-dev curl && rm -rf /var/lib/apt/lists/*
RUN python -m venv .venv
ENV PATH="/app/.venv/bin:$PATH"
COPY backend/requirements.txt ./
RUN pip install --no-cache-dir -r requirements.txt
COPY etl_pipeline.py ./
COPY gemini-ga4-connection-xxxxx.json ./
COPY backend/ ./backend/
COPY --from=frontend-builder /app/frontend/dist ./frontend/dist
EXPOSE 8000
CMD ["uvicorn", "backend.main:app", "--host", "0.0.0.0", "--port", "8000"]

copy

起動スクリプト (`start_app.sh`)

#!/bin/bash
docker compose up -d --build
echo "ブラウザでダッシュボードを開いてください：http://localhost:8004"

copy

テストデータの流し込み（検証）

APIキーがない環境でも本アプリの動作（およびAIによる分析）をテストするために、ダミーデータを自動生成してMySQLに流し込むスクリプトを用意します。
`scratch/generate_dummy_data.py` などのスクリプトで、過去30〜45日間のセッション、クリック、レイジクリックデータを生成して `web_analytics_daily` に `to_sql` でバルクインサートします。

9. データベースのチューニングとトラブルシューティング

初期の開発段階で、データ件数が多くなった際に以下のエラーに遭遇しました。

`OperationalError: (2013, ‘Lost connection to MySQL server during query’)`

原因と対策の解説

1. プレフィックスインデックスによるボトルネック
   もともとインデックスは `INDEX idx_url (url(255))` となっていましたが、MySQLはURLの前方255文字しかインデックス化していませんでした。このため、`GROUP BY url` のグループ化処理でインデックスが使用できず、テーブル全件をスキャンして一時テーブル（Temporary Table）を作成・ソートしていました。これがタイムアウトの原因です。
   • 対策: `init.sql` を修正し、プレフィックス制限なしの `INDEX idx_url (url)` に変更しました。これにより、実行計画（EXPLAIN）がテーブルフルスキャン (`ALL`) からインデックススキャン (`index`) になり、クエリが一瞬で完了するようになりました。
2. タイムアウト設定の追加
   SQLAlchemy のデフォルトのコネクションプール管理では、重いクエリで接続が切断されやすくなります。
   • 対策: `create_engine` 時に `connect_args={“read_timeout”: 120}` を追加してサーバー側の応答待ち時間を延ばし、`pool_recycle` と `pool_pre_ping` で生存確認を行わせることでエラーをシャットアウトしました。

まとめ

これで、データの集約・ダッシュボード上での可視化・AIによる自然言語解析のすべてを備えた統合アナリティクスハブが完成しました
ローカルLLM (Ollama) と MySQL を組み合わせ、データを外部の有償クラウドに送ることなく、完全ローカルで安全にAI分析を行えます

なお、ollama run gemma4:latest, ollama serve, pythonの仮想環境などは設定済みとさせてください．

Enjoy WebAnalize!