Google Sheets API × Gemini API でつくるデータ処理パイプライン — Apps Script 不要の Python 自動化

pip install google-auth google-auth-oauthlib google-api-python-client google-generativeai

import os
from google.oauth2 import service_account
from googleapiclient.discovery import build
 
# サービスアカウント認証
SCOPES = [
    "https://www.googleapis.com/auth/spreadsheets",
    "https://www.googleapis.com/auth/drive.readonly",
]
 
def get_sheets_service():
    """Google Sheets API サービスオブジェクトを取得する"""
    credentials = service_account.Credentials.from_service_account_file(
        "credentials.json",
        scopes=SCOPES,
    )
    return build("sheets", "v4", credentials=credentials)
 
def read_sheet(spreadsheet_id: str, range_name: str) -> list[list[str]]:
    """
    スプレッドシートから指定範囲のデータを読み取る
 
    Args:
        spreadsheet_id: URL の /d/{ID}/ 部分
        range_name: 例 "Sheet1\!A2:C100"（ヘッダー行を除いた範囲）
 
    Returns:
        2次元リスト。空の場合は []
    """
    service = get_sheets_service()
    result = (
        service.spreadsheets()
        .values()
        .get(spreadsheetId=spreadsheet_id, range=range_name)
        .execute()
    )
    return result.get("values", [])

import json
import google.generativeai as genai
import typing_extensions as typing
 
# 環境変数から API キーを読み込む
genai.configure(api_key=os.environ["GEMINI_API_KEY"])
model = genai.GenerativeModel("gemini-2.5-flash-preview")
 
# 出力スキーマを定義（TypedDict で型安全に）
class FeedbackAnalysis(typing.TypedDict):
    category: str          # バグ報告 / 機能要望 / 使い方の質問 / 称賛 / その他
    sentiment: str         # positive / neutral / negative
    urgency: str           # high / medium / low
    summary: str           # 30文字以内の要約
    suggestion: str        # 改善提案（簡潔に）
 
def analyze_feedback(feedback_text: str) -> FeedbackAnalysis:
    """
    顧客フィードバックを Gemini で分析する
 
    Args:
        feedback_text: 分析対象のフィードバックテキスト
 
    Returns:
        FeedbackAnalysis 型の辞書
 
    Raises:
        ValueError: Gemini の応答が JSON パースできない場合
    """
    prompt = f"""以下の顧客フィードバックを分析してください。
 
フィードバック:
{feedback_text}
 
分類ルール:
- category: バグ報告 / 機能要望 / 使い方の質問 / 称賛 / その他 のいずれか
- sentiment: positive / neutral / negative のいずれか
- urgency: high（即対応必要） / medium（近いうちに対応） / low（余裕あり）
- summary: 30文字以内で内容を要約
- suggestion: 具体的な改善提案（改善不要なら「対応不要」）"""
 
    response = model.generate_content(
        prompt,
        generation_config=genai.GenerationConfig(
            response_mime_type="application/json",
            response_schema=FeedbackAnalysis,
        ),
    )
 
    return json.loads(response.text)

def write_results(
    spreadsheet_id: str,
    results: list[FeedbackAnalysis],
    start_row: int = 2,
    start_col: str = "D",
) -> None:
    """
    分析結果をスプレッドシートに書き込む
 
    Args:
        spreadsheet_id: 書き込み先のスプレッドシート ID
        results: analyze_feedback() の結果リスト
        start_row: データ書き込み開始行（デフォルト 2 = ヘッダー行の次）
        start_col: 書き込み開始列（デフォルト D 列）
    """
    service = get_sheets_service()
 
    # 2次元リストに変換（API の expects する形式）
    values = [
        [
            r["category"],
            r["sentiment"],
            r["urgency"],
            r["summary"],
            r["suggestion"],
        ]
        for r in results
    ]
 
    end_row = start_row + len(values) - 1
    end_col = chr(ord(start_col) + 4)  # D + 4列 = H
    range_name = f"Sheet1\!{start_col}{start_row}:{end_col}{end_row}"
 
    body = {"values": values}
    service.spreadsheets().values().update(
        spreadsheetId=spreadsheet_id,
        range=range_name,
        valueInputOption="USER_ENTERED",  # 書式を Sheets が解釈する
        body=body,
    ).execute()
 
    print(f"✅ {len(values)} 件の結果を {range_name} に書き込みました")

import time
 
SPREADSHEET_ID = "あなたのスプレッドシートID"
INPUT_RANGE = "Sheet1\!C2:C1000"  # C列: フィードバックテキスト
 
def run_pipeline() -> None:
    """フィードバック分析パイプラインのメイン処理"""
 
    print("📊 スプレッドシートからデータを読み取り中...")
    rows = read_sheet(SPREADSHEET_ID, INPUT_RANGE)
 
    if not rows:
        print("処理対象データがありませんでした")
        return
 
    # 空セルをスキップ
    feedback_texts = [row[0].strip() for row in rows if row and row[0].strip()]
    print(f"📝 {len(feedback_texts)} 件のフィードバックを処理します")
 
    results = []
    for i, text in enumerate(feedback_texts, 1):
        print(f"  [{i}/{len(feedback_texts)}] 処理中: {text[:30]}...")
        try:
            analysis = analyze_feedback(text)
            results.append(analysis)
        except Exception as e:
            # 1件失敗しても全体を止めない
            print(f"  ⚠️ エラー（行 {i + 1}）: {e}")
            results.append({
                "category": "エラー",
                "sentiment": "unknown",
                "urgency": "low",
                "summary": "処理エラー",
                "suggestion": str(e)[:50],
            })
 
        # レート制限対策: 1秒待機（60rpm の無料枠を超えないように）
        if i % 10 == 0:
            time.sleep(1)
 
    print("📤 結果をスプレッドシートに書き込み中...")
    write_results(SPREADSHEET_ID, results)
    print("🎉 パイプライン完了！")
 
if __name__ == "__main__":
    run_pipeline()

try:
    result = json.loads(response.text)
except json.JSONDecodeError:
    # フォールバック: テキストから手動パース or スキップ
    result = {"category": "パースエラー", "sentiment": "unknown", ...}

# crontab -e で設定例（毎日 9:00 JST に実行）
0 0 * * * cd /path/to/project && python pipeline.py >> logs/pipeline.log 2>&1

from fastapi import FastAPI, BackgroundTasks
 
app = FastAPI()
 
@app.post("/analyze")
async def trigger_analysis(background_tasks: BackgroundTasks):
    background_tasks.add_task(run_pipeline)
    return {"status": "started"}