Go to Qiita Advent Calendar Top
LoginSignup
0
0

Delete article

Deleted articles cannot be recovered.

Draft of this article would be also deleted.

Are you sure you want to delete this article?

@kerobot(MANABU KIKUCHI)

YouTube Live Chat API に Google OAuth2 認証とチャット投稿機能を追加

Last updated at Posted at 2025-10-16

はじめに

以前作成した YouTube Live Chat API に、Google OAuth2 認証を追加して、ライブチャットの読み取りだけでなく、メッセージの投稿も可能にしました。

YouTube Data API v3 でライブチャットにメッセージを投稿するには OAuth2 認証が必須なので、Google Cloud Console での設定からコード実装までの流れを記載します。

image.png

今回追加した主な機能

  • Google OAuth2 認証フロー（認証URL生成、コールバック処理、トークン管理）
  • ライブチャットメッセージ投稿機能
  • アクセストークンの検証機能
  • リフレッシュトークンによるトークン更新機能

データフロー概要

今回実装したOAuth2認証からライブチャット投稿までの全体的なデータフローは以下のようになります。

1. OAuth2認証フロー

[クライアント] → [API Server] → [Google OAuth2] → [API Server] → [クライアント]
     ①              ②              ③              ④              ⑤

① 認証URL要求: クライアントが /api/auth/login エンドポイントに認証URL生成を要求
② 認証URL生成: サーバーがGoogle OAuth2認証URLとCSRF防止用stateを生成して返却
③ ユーザー認証: クライアントがブラウザでGoogleにアクセスし、ユーザーが認証を実行
④ 認証コード取得: Googleが認証コードと共にコールバックURL (/api/auth/callback) にリダイレクト
⑤ アクセストークン発行: サーバーが認証コードをGoogleのトークンエンドポイントでアクセストークンに交換

2. ライブチャット投稿フロー

[クライアント] → [API Server] → [YouTube API] → [ライブチャット]
     ⑥              ⑦              ⑧              ⑨

⑥ 投稿要求: クライアントが動画ID、メッセージ、アクセストークンを /api/youtube/livechat/message に送信
⑦ ライブチャットID取得: サーバーが動画IDから対応するライブチャットIDを YouTube Data API で取得
⑧ メッセージ投稿: サーバーがアクセストークンを使用してYouTube Live Chat API にメッセージを投稿
⑨ 配信反映: 投稿されたメッセージがライブチャットに表示される

作ったもの

開発環境

  • Windows 11
  • Python 3.13
  • FastAPI 0.115.6
  • Google Auth Libraries (google-auth, google-auth-oauthlib, google-auth-httplib2)
  • Poetry（依存関係管理）

Google Cloud Console での設定

OAuth2 認証を実装するには、Google Cloud Console で OAuth2 クライアントの設定が必要です。

  1. Google Cloud Console でプロジェクトを作成/選択
  2. YouTube Data API v3 を有効化
  3. OAuth2 クライアント ID を作成
  4. リダイレクト URI に http://localhost:8000/api/auth/callback を設定
  5. Client ID と Client Secret を取得

環境変数の追加

OAuth2 認証に必要な環境変数を .env に追加しました。

# YOUTUBE_API_KEY="YouTube Data API v3 API Key"
# LOG_LEVEL=WARNING
# CORS_ORIGINS=https://yourdomain.com
+
+# Google OAuth2設定 (Google Cloud Consoleで取得してください)
+GOOGLE_CLIENT_ID="Client ID"
+GOOGLE_CLIENT_SECRET="Client Secret"
+OAUTH_REDIRECT_URI=http://localhost:8000/api/auth/callback

依存関係の追加

OAuth2 認証に必要な Google 認証ライブラリを pyproject.toml に追加しました。

dependencies = [
    "fastapi (>=0.115.6,<0.116.0)",
    "uvicorn (>=0.34.0,<0.35.0)",
    "python-dotenv (>=1.1.1,<2.0.0)",
    "pydantic (>=2.11.7,<3.0.0)",
    "pytest (>=8.4.1,<9.0.0)",
+   "google-auth (>=2.16.0,<3.0.0)",
+   "google-auth-oauthlib (>=1.0.0,<2.0.0)",
+   "google-auth-httplib2 (>=0.2.0,<1.0.0)",
    "pytest-asyncio (>=1.1.0,<2.0.0)",
    "pytest-mock (>=3.14.1,<4.0.0)",
    "httpx (>=0.28.1,<0.29.0)",
]

設定ファイルの更新

OAuth2 関連の設定を app/config.py に追加しました。

# OAuth2設定
GOOGLE_CLIENT_ID = os.getenv("GOOGLE_CLIENT_ID")
GOOGLE_CLIENT_SECRET = os.getenv("GOOGLE_CLIENT_SECRET")
OAUTH_REDIRECT_URI = os.getenv(
    "OAUTH_REDIRECT_URI", "http://localhost:8000/api/auth/callback"
)

if not GOOGLE_CLIENT_ID or not GOOGLE_CLIENT_SECRET:
    print(
        "⚠️  OAuth2認証が無効: GOOGLE_CLIENT_ID と GOOGLE_CLIENT_SECRET を設定してください"
    )

# OAuth2スコープ
YOUTUBE_OAUTH_SCOPES = ["https://www.googleapis.com/auth/youtube.force-ssl"]

データモデルの定義

OAuth2 認証とライブチャット投稿用のPydanticモデルを app/models/auth.py に定義しました。

from pydantic import BaseModel, Field
from typing import Optional

class AuthUrlResponse(BaseModel):
    """OAuth2認証URL生成レスポンス"""
    auth_url: str = Field(..., description="Google OAuth2認証URL")
    state: str = Field(..., description="CSRF攻撃防止用のstateパラメータ")

class TokenRequest(BaseModel):
    """OAuth2トークン取得リクエスト"""
    code: str = Field(..., description="認証コード")
    state: Optional[str] = Field(
        None, description="stateパラメータ（callback_url含む）"
    )

class TokenResponse(BaseModel):
    """OAuth2トークンレスポンス"""
    access_token: str = Field(..., description="アクセストークン")
    refresh_token: Optional[str] = Field(None, description="リフレッシュトークン")
    expires_in: int = Field(..., description="トークンの有効期限（秒）")
    token_type: str = Field(default="Bearer", description="トークンタイプ")

class PostChatMessageRequest(BaseModel):
    """ライブチャット投稿リクエスト"""
    video_id: str = Field(
        ..., description="YouTube動画ID（11文字）", min_length=11, max_length=11
    )
    message_text: str = Field(
        ..., min_length=1, max_length=200, description="投稿するメッセージ"
    )
    access_token: str = Field(..., description="OAuth2アクセストークン")

class PostChatMessageResponse(BaseModel):
    """ライブチャット投稿レスポンス"""
    message_id: str = Field(..., description="投稿されたメッセージのID")
    message_text: str = Field(..., description="投稿されたメッセージテキスト")
    author_name: str = Field(..., description="投稿者名")
    published_at: str = Field(..., description="投稿日時")
    success: bool = Field(default=True, description="投稿成功フラグ")

OAuth2サービスの実装

OAuth2 認証フローを管理する app/services/oauth2.py を実装しました。

import secrets
import json
from typing import Optional, Dict, Any
from google.auth.transport.requests import Request
from google_auth_oauthlib.flow import Flow
from google.oauth2.credentials import Credentials
import logging
from app.config import (
    GOOGLE_CLIENT_ID,
    GOOGLE_CLIENT_SECRET,
    OAUTH_REDIRECT_URI,
    YOUTUBE_OAUTH_SCOPES,
)

logger = logging.getLogger(__name__)

class OAuth2Service:
    """Google OAuth2認証サービス"""

    def __init__(self):
        self._client_config = {
            "web": {
                "client_id": GOOGLE_CLIENT_ID,
                "client_secret": GOOGLE_CLIENT_SECRET,
                "auth_uri": "https://accounts.google.com/o/oauth2/auth",
                "token_uri": "https://oauth2.googleapis.com/token",
                "redirect_uris": [OAUTH_REDIRECT_URI],
            }
        }
        
        self._active_states: Dict[str, Optional[str]] = {}  # CSRF攻撃防止用

    def generate_auth_url(self, callback_url: Optional[str] = None) -> tuple[str, str]:
        """OAuth2認証URLを生成"""
        if not GOOGLE_CLIENT_ID or not GOOGLE_CLIENT_SECRET:
            raise ValueError(
                "❌ OAuth2設定が不完全です。GOOGLE_CLIENT_ID と GOOGLE_CLIENT_SECRET を設定してください。"
            )

        try:
            flow = Flow.from_client_config(
                self._client_config,
                scopes=YOUTUBE_OAUTH_SCOPES,
                redirect_uri=OAUTH_REDIRECT_URI,
            )

            # CSRF攻撃防止用のstateパラメータ生成
            state = secrets.token_urlsafe(32)
            
            # callback_urlがある場合はstateに含める
            state_data = {"csrf_token": state}
            if callback_url:
                state_data["callback_url"] = callback_url

            state_json = json.dumps(state_data)
            self._active_states[state] = callback_url

            auth_url, _ = flow.authorization_url(
                access_type="offline", include_granted_scopes="true", state=state_json
            )

            logger.info(f"🔐 OAuth2認証URL生成完了: state={state}")
            return auth_url, state

        except Exception as e:
            logger.error(f"💥 OAuth2認証URL生成エラー: {e}")
            raise Exception(f"認証URL生成に失敗しました: {str(e)}")

    def exchange_code_for_token(self, code: str, state: str) -> Dict[str, Any]:
        """認証コードをアクセストークンに交換"""
        try:
            # stateをJSONデコード
            state_data = json.loads(state)
            csrf_token = state_data.get("csrf_token")
            callback_url = state_data.get("callback_url")
        except (json.JSONDecodeError, KeyError):
            logger.warning(f"🚫 不正なstateフォーマット: {state}")
            raise ValueError("不正なstateパラメータです。")

        # CSRF トークン検証
        if csrf_token not in self._active_states:
            logger.warning(f"🚫 不正なCSRFトークン: {csrf_token}")
            raise ValueError("不正なstateパラメータです。")

        # 使用済みstateを削除
        del self._active_states[csrf_token]

        try:
            flow = Flow.from_client_config(
                self._client_config,
                scopes=YOUTUBE_OAUTH_SCOPES,
                redirect_uri=OAUTH_REDIRECT_URI,
            )

            flow.fetch_token(code=code)
            credentials = flow.credentials

            token_info = {
                "access_token": credentials.token,
                "refresh_token": credentials.refresh_token,
                "expires_in": 3600,  # Googleは通常1時間
                "token_type": "Bearer",
            }

            # callback_urlがある場合は追加
            if callback_url:
                token_info["callback_url"] = callback_url

            logger.info("✅ OAuth2トークン取得成功")
            return token_info

        except Exception as e:
            logger.error(f"💥 OAuth2トークン取得エラー: {e}")
            raise Exception(f"トークン取得に失敗しました: {str(e)}")

    def validate_token(self, access_token: str) -> bool:
        """アクセストークンの有効性を検証"""
        try:
            import requests

            response = requests.get(
                f"https://www.googleapis.com/oauth2/v1/tokeninfo?access_token={access_token}"
            )

            if response.status_code == 200:
                token_info = response.json()
                # スコープ確認
                scopes = token_info.get("scope", "").split()
                required_scope = "https://www.googleapis.com/auth/youtube.force-ssl"

                if required_scope in scopes:
                    logger.info("✅ アクセストークン有効")
                    return True
                else:
                    logger.warning("⚠️ 必要なスコープが不足")
                    return False
            else:
                logger.warning("⚠️ アクセストークン無効")
                return False

        except Exception as e:
            logger.error(f"💥 トークン検証エラー: {e}")
            return False

# シングルトンインスタンス
oauth2_service = OAuth2Service()

認証APIエンドポイントの実装

OAuth2 認証用のAPIエンドポイントを app/api/auth.py に実装しました。

from fastapi import APIRouter, HTTPException, Query
from fastapi.responses import RedirectResponse
from app.services.oauth2 import oauth2_service
from app.models.auth import (
    AuthUrlResponse,
    TokenRequest,
    TokenResponse,
)
from typing import Optional
import logging

logger = logging.getLogger(__name__)

router = APIRouter(
    prefix="/api/auth",
    tags=["authentication"],
)

@router.get("/login", response_model=AuthUrlResponse)
def get_auth_url(
    callback_url: Optional[str] = Query(None, description="認証後のリダイレクト先URL")
):
    """Google OAuth2認証URLを生成するエンドポイント"""
    try:
        auth_url, state = oauth2_service.generate_auth_url(callback_url)
        return AuthUrlResponse(auth_url=auth_url, state=state)
    except Exception as e:
        logger.error(f"💥 認証URL生成エラー: {e}")
        raise HTTPException(status_code=500, detail="認証URL生成に失敗しました")

@router.get("/callback")
def auth_callback(
    code: str = Query(..., description="Google認証サーバーからの認証コード"),
    state: str = Query(..., description="CSRF攻撃防止用のstateパラメータ"),
):
    """Google OAuth2認証コールバックエンドポイント"""
    try:
        token_info = oauth2_service.exchange_code_for_token(code, state)
        callback_url = token_info.pop("callback_url", None)

        # callback_urlがある場合はリダイレクト
        if callback_url:
            redirect_url = f"{callback_url}?access_token={token_info['access_token']}&status=success"
            logger.info(f"🔄 Redirecting to: {callback_url}")
            return RedirectResponse(url=redirect_url)

        # callback_urlがない場合はJSONレスポンス
        return TokenResponse(**token_info)

    except Exception as e:
        logger.error(f"💥 認証コールバックエラー: {e}")
        raise HTTPException(status_code=400, detail=str(e))

@router.post("/token", response_model=TokenResponse)
def exchange_token(request: TokenRequest):
    """認証コードをアクセストークンに交換するエンドポイント（POSTバージョン）"""
    try:
        if request.state is None:
            raise HTTPException(status_code=400, detail="stateパラメータが必要です")

        token_info = oauth2_service.exchange_code_for_token(request.code, request.state)
        token_info.pop("callback_url", None)
        return TokenResponse(**token_info)
    except Exception as e:
        logger.error(f"💥 トークン交換エラー: {e}")
        raise HTTPException(status_code=400, detail=str(e))

@router.get("/validate")
def validate_token(
    access_token: str = Query(..., description="検証するアクセストークン")
):
    """アクセストークンの有効性を検証"""
    try:
        is_valid = oauth2_service.validate_token(access_token)
        return {
            "valid": is_valid,
            "message": "トークン有効" if is_valid else "トークン無効",
        }
    except Exception as e:
        logger.error(f"💥 トークン検証エラー: {e}")
        raise HTTPException(status_code=500, detail="トークン検証に失敗しました")

ライブチャット投稿機能の実装

YouTube サービスに、ライブチャット投稿機能を app/services/youtube.py に追加しました。

def post_chat_message(
    self, video_id: str, message_text: str, access_token: str
) -> dict:
    """ライブチャットにメッセージを投稿"""
    logger.info(f"📝 Posting chat message to video: {video_id}")

    # まずライブチャットIDを取得
    live_chat_id = self.get_live_chat_id(video_id)
    if not live_chat_id:
        raise Exception(f"ライブチャットが見つかりません: {video_id}")

    self._wait_for_rate_limit()

    url = "https://www.googleapis.com/youtube/v3/liveChat/messages"
    headers = {
        "Authorization": f"Bearer {access_token}",
        "Content-Type": "application/json",
    }
    params = {
        "part": "snippet",
    }
    body = {
        "snippet": {
            "liveChatId": live_chat_id,
            "type": "textMessageEvent",
            "textMessageDetails": {"messageText": message_text},
        }
    }

    try:
        import requests

        response = requests.post(url, headers=headers, params=params, json=body)

        if response.status_code == 200:
            result = response.json()
            logger.info(f"✅ Chat message posted successfully: {message_text}")
            return {
                "message_id": result.get("id", ""),
                "message_text": message_text,
                "author_name": result.get("snippet", {}).get(
                    "authorDisplayName", ""
                ),
                "published_at": result.get("snippet", {}).get("publishedAt", ""),
                "success": True,
            }
        else:
            error_msg = (
                f"YouTube API error: {response.status_code} - {response.text}"
            )
            logger.error(f"💥 Failed to post chat message: {error_msg}")
            raise Exception(error_msg)

    except Exception as e:
        logger.error(f"💥 Error posting chat message: {e}")
        raise

YouTube APIエンドポイントの拡張

ライブチャット投稿用のエンドポイントを app/api/youtube.py に追加しました。

@router.post("/livechat/message", response_model=PostChatMessageResponse)
def post_livechat_message(request: PostChatMessageRequest):
    """ライブチャットにメッセージを投稿するエンドポイント"""
    try:
        # 動画IDバリデーション
        if not validate_youtube_video_id(request.video_id):
            logger.warning(f"🚫 Invalid video_id: {request.video_id}")
            raise HTTPException(
                status_code=400,
                detail="無効な動画IDです。11文字の英数字・ハイフン・アンダースコアのみ使用してください。",
            )

        result = youtube_service.post_chat_message(
            request.video_id, request.message_text, request.access_token
        )

        return PostChatMessageResponse(**result)

    except HTTPException:
        raise
    except Exception as e:
        logger.error(f"💥 Error posting chat message: {e}")
        raise HTTPException(
            status_code=500, detail=f"チャット投稿に失敗しました: {str(e)}"
        )

メインアプリケーションの更新

認証ルーターを app/main.py に追加しました。

from app.api.auth import router as auth_router

# ルーターの追加
app.include_router(youtube_router)
app.include_router(auth_router)

OAuth2 認証フローの使用方法

1. 認証URL取得

curl "http://localhost:8000/api/auth/login"

レスポンス：

{
  "auth_url": "https://accounts.google.com/o/oauth2/auth?...",
  "state": "csrf_token_string"
}

2. ブラウザで認証

返されたauth_urlにブラウザでアクセスしてGoogle認証を完了します。

3. アクセストークン取得

認証完了後、コールバックURLでアクセストークンが取得できます。

4. ライブチャット投稿

curl -X POST "http://localhost:8000/api/youtube/livechat/message" \
  -H "Content-Type: application/json" \
  -d '{
    "video_id": "YOUTUBE_VIDEO_ID",
    "message_text": "Hello from API!",
    "access_token": "YOUR_ACCESS_TOKEN"
  }'

動作確認

OAuth2 認証フローとライブチャット投稿が正常に動作することを確認しました。

  1. 認証URL生成 ✅
  2. Google OAuth2 認証フロー ✅
  3. アクセストークン取得 ✅
  4. ライブチャットメッセージ投稿 ✅
  5. トークン検証機能 ✅

セキュリティ対策

実装したセキュリティ対策：

  • CSRF攻撃防止: state パラメータによるCSRFトークン検証
  • トークン管理: アクセストークンとリフレッシュトークンの適切な管理
  • スコープ制限: 必要最小限のYouTubeスコープのみ要求
  • 入力値検証: 動画IDとメッセージテキストのバリデーション
  • エラーハンドリング: 詳細なエラー情報の適切な処理

おわりに

YouTube Live Chat API に OAuth2 認証機能を追加することで、ライブチャットの読み取りに加えてメッセージの投稿も可能になりました。これで凰牙るき様の歌枠配信の弾幕コメント投下が捗る？かもしれません。

とはいえ、Google Cloud Console の無料枠を利用しているので、クォータ制限に注意して必要以上の連投もしないように気を付けないといけません。

OAuth2 の実装は少々難解でしたが、Google の認証ライブラリと Claude Sonnet 4 エージェントを活用することで比較的スムーズに実装できました。特に google-auth-oauthlib ライブラリは認証フローの実装を大幅に簡単にしてくれています。

続いて、TypeScript + React で開発しているフロント側に、今回実装した機能を組み込み、リアルタイムのライブチャットの表示だけでなくメッセージの投稿も行えるようにしてみます✨

0
0
0

Register as a new user and use Qiita more conveniently

  1. You get articles that match your needs
  2. You can efficiently read back useful information
  3. You can use dark theme
What you can do with signing up
Sign upLogin
0
0

Delete article

Deleted articles cannot be recovered.

Draft of this article would be also deleted.

Are you sure you want to delete this article?