""" YouTube OAuth Client Google OAuth를 사용한 YouTube 인증 클라이언트입니다. """ import logging from urllib.parse import urlencode import httpx from config import social_oauth_settings from app.social.constants import SocialPlatform, YOUTUBE_SCOPES from app.social.exceptions import ( OAuthCodeExchangeError, OAuthTokenRefreshError, SocialAccountError, ) from app.social.oauth.base import BaseOAuthClient from app.social.schemas import OAuthTokenResponse, PlatformUserInfo logger = logging.getLogger(__name__) class YouTubeOAuthClient(BaseOAuthClient): """ YouTube OAuth 클라이언트 Google OAuth 2.0을 사용하여 YouTube 계정 인증을 처리합니다. """ platform = SocialPlatform.YOUTUBE # Google OAuth 엔드포인트 AUTHORIZATION_URL = "https://accounts.google.com/o/oauth2/v2/auth" TOKEN_URL = "https://oauth2.googleapis.com/token" USERINFO_URL = "https://www.googleapis.com/oauth2/v2/userinfo" YOUTUBE_CHANNEL_URL = "https://www.googleapis.com/youtube/v3/channels" REVOKE_URL = "https://oauth2.googleapis.com/revoke" def __init__(self) -> None: self.client_id = social_oauth_settings.YOUTUBE_CLIENT_ID self.client_secret = social_oauth_settings.YOUTUBE_CLIENT_SECRET self.redirect_uri = social_oauth_settings.YOUTUBE_REDIRECT_URI def get_authorization_url(self, state: str) -> str: """ Google OAuth 인증 URL 생성 Args: state: CSRF 방지용 state 토큰 Returns: str: Google OAuth 인증 페이지 URL """ params = { "client_id": self.client_id, "redirect_uri": self.redirect_uri, "response_type": "code", "scope": " ".join(YOUTUBE_SCOPES), "access_type": "offline", # refresh_token 받기 위해 필요 "prompt": "select_account", # 계정 선택만 표시 (동의 화면은 최초 1회만) "state": state, } url = f"{self.AUTHORIZATION_URL}?{urlencode(params)}" logger.debug(f"[YOUTUBE_OAUTH] 인증 URL 생성: {url[:100]}...") return url async def exchange_code(self, code: str) -> OAuthTokenResponse: """ 인가 코드로 액세스 토큰 교환 Args: code: OAuth 인가 코드 Returns: OAuthTokenResponse: 액세스 토큰 및 리프레시 토큰 Raises: OAuthCodeExchangeError: 토큰 교환 실패 시 """ logger.info(f"[YOUTUBE_OAUTH] 토큰 교환 시작 - code: {code[:20]}...") data = { "client_id": self.client_id, "client_secret": self.client_secret, "code": code, "grant_type": "authorization_code", "redirect_uri": self.redirect_uri, } async with httpx.AsyncClient() as client: try: response = await client.post( self.TOKEN_URL, data=data, headers={"Content-Type": "application/x-www-form-urlencoded"}, ) response.raise_for_status() token_data = response.json() logger.info("[YOUTUBE_OAUTH] 토큰 교환 성공") logger.debug( f"[YOUTUBE_OAUTH] 토큰 정보 - " f"expires_in: {token_data.get('expires_in')}, " f"scope: {token_data.get('scope')}" ) return OAuthTokenResponse( access_token=token_data["access_token"], refresh_token=token_data.get("refresh_token"), expires_in=token_data["expires_in"], token_type=token_data.get("token_type", "Bearer"), scope=token_data.get("scope"), ) except httpx.HTTPStatusError as e: error_detail = e.response.text if e.response else str(e) logger.error( f"[YOUTUBE_OAUTH] 토큰 교환 실패 - " f"status: {e.response.status_code}, error: {error_detail}" ) raise OAuthCodeExchangeError( platform=self.platform.value, detail=f"토큰 교환 실패: {error_detail}", ) except Exception as e: logger.error(f"[YOUTUBE_OAUTH] 토큰 교환 중 예외 발생: {e}") raise OAuthCodeExchangeError( platform=self.platform.value, detail=str(e), ) async def refresh_token(self, refresh_token: str) -> OAuthTokenResponse: """ 리프레시 토큰으로 액세스 토큰 갱신 Args: refresh_token: 리프레시 토큰 Returns: OAuthTokenResponse: 새 액세스 토큰 Raises: OAuthTokenRefreshError: 토큰 갱신 실패 시 """ logger.info("[YOUTUBE_OAUTH] 토큰 갱신 시작") data = { "client_id": self.client_id, "client_secret": self.client_secret, "refresh_token": refresh_token, "grant_type": "refresh_token", } async with httpx.AsyncClient() as client: try: response = await client.post( self.TOKEN_URL, data=data, headers={"Content-Type": "application/x-www-form-urlencoded"}, ) response.raise_for_status() token_data = response.json() logger.info("[YOUTUBE_OAUTH] 토큰 갱신 성공") return OAuthTokenResponse( access_token=token_data["access_token"], refresh_token=refresh_token, # Google은 refresh_token 재발급 안함 expires_in=token_data["expires_in"], token_type=token_data.get("token_type", "Bearer"), scope=token_data.get("scope"), ) except httpx.HTTPStatusError as e: error_detail = e.response.text if e.response else str(e) logger.error( f"[YOUTUBE_OAUTH] 토큰 갱신 실패 - " f"status: {e.response.status_code}, error: {error_detail}" ) raise OAuthTokenRefreshError( platform=self.platform.value, detail=f"토큰 갱신 실패: {error_detail}", ) except Exception as e: logger.error(f"[YOUTUBE_OAUTH] 토큰 갱신 중 예외 발생: {e}") raise OAuthTokenRefreshError( platform=self.platform.value, detail=str(e), ) async def get_user_info(self, access_token: str) -> PlatformUserInfo: """ YouTube 채널 정보 조회 Args: access_token: 액세스 토큰 Returns: PlatformUserInfo: YouTube 채널 정보 Raises: SocialAccountError: 정보 조회 실패 시 """ logger.info("[YOUTUBE_OAUTH] 사용자/채널 정보 조회 시작") headers = {"Authorization": f"Bearer {access_token}"} async with httpx.AsyncClient() as client: try: # 1. Google 사용자 기본 정보 조회 userinfo_response = await client.get( self.USERINFO_URL, headers=headers, ) userinfo_response.raise_for_status() userinfo = userinfo_response.json() # 2. YouTube 채널 정보 조회 channel_params = { "part": "snippet,statistics", "mine": "true", } channel_response = await client.get( self.YOUTUBE_CHANNEL_URL, headers=headers, params=channel_params, ) channel_response.raise_for_status() channel_data = channel_response.json() # 채널이 없는 경우 if not channel_data.get("items"): logger.warning("[YOUTUBE_OAUTH] YouTube 채널 없음") raise SocialAccountError( platform=self.platform.value, detail="YouTube 채널이 없습니다. 채널을 먼저 생성해주세요.", ) channel = channel_data["items"][0] snippet = channel.get("snippet", {}) statistics = channel.get("statistics", {}) logger.info( f"[YOUTUBE_OAUTH] 채널 정보 조회 성공 - " f"channel_id: {channel['id']}, " f"title: {snippet.get('title')}" ) return PlatformUserInfo( platform_user_id=channel["id"], username=snippet.get("customUrl"), # @username 형태 display_name=snippet.get("title"), profile_image_url=snippet.get("thumbnails", {}) .get("default", {}) .get("url"), platform_data={ "channel_id": channel["id"], "channel_title": snippet.get("title"), "channel_description": snippet.get("description"), "custom_url": snippet.get("customUrl"), "subscriber_count": statistics.get("subscriberCount"), "video_count": statistics.get("videoCount"), "view_count": statistics.get("viewCount"), "google_user_id": userinfo.get("id"), "google_email": userinfo.get("email"), }, ) except httpx.HTTPStatusError as e: error_detail = e.response.text if e.response else str(e) logger.error( f"[YOUTUBE_OAUTH] 정보 조회 실패 - " f"status: {e.response.status_code}, error: {error_detail}" ) raise SocialAccountError( platform=self.platform.value, detail=f"사용자 정보 조회 실패: {error_detail}", ) except SocialAccountError: raise except Exception as e: logger.error(f"[YOUTUBE_OAUTH] 정보 조회 중 예외 발생: {e}") raise SocialAccountError( platform=self.platform.value, detail=str(e), ) async def revoke_token(self, token: str) -> bool: """ 토큰 폐기 (연동 해제 시) Args: token: 폐기할 토큰 (access_token 또는 refresh_token) Returns: bool: 폐기 성공 여부 """ logger.info("[YOUTUBE_OAUTH] 토큰 폐기 시작") async with httpx.AsyncClient() as client: try: response = await client.post( self.REVOKE_URL, data={"token": token}, headers={"Content-Type": "application/x-www-form-urlencoded"}, ) if response.status_code == 200: logger.info("[YOUTUBE_OAUTH] 토큰 폐기 성공") return True else: logger.warning( f"[YOUTUBE_OAUTH] 토큰 폐기 실패 - " f"status: {response.status_code}, body: {response.text}" ) return False except Exception as e: logger.error(f"[YOUTUBE_OAUTH] 토큰 폐기 중 예외 발생: {e}") return False # 싱글톤 인스턴스 youtube_oauth_client = YouTubeOAuthClient()