327 lines
12 KiB
Python
327 lines
12 KiB
Python
"""
|
|
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()
|