117 lines
5.4 KiB
Python
117 lines
5.4 KiB
Python
import json
|
|
from typing import Optional
|
|
from fastapi import HTTPException, Depends
|
|
from googleapiclient.discovery import build
|
|
from googleapiclient.errors import HttpError
|
|
from google.oauth2.credentials import Credentials
|
|
from google.auth.transport.requests import Request as GoogleRequest
|
|
from .utils import load_client_config
|
|
from .redis import RedisYouTubeStorage, RedisOAuthStorage, get_youtube_storage, get_oauth_storage
|
|
|
|
class YoutubeService:
|
|
def __init__(self, youtube_storage: RedisYouTubeStorage, oauth_storage: RedisOAuthStorage):
|
|
self.youtube_storage = youtube_storage
|
|
self.oauth_storage = oauth_storage
|
|
self.scopes = [
|
|
"https://www.googleapis.com/auth/youtube.upload",
|
|
"https://www.googleapis.com/auth/youtube.readonly",
|
|
"https://www.googleapis.com/auth/userinfo.email",
|
|
"https://www.googleapis.com/auth/userinfo.profile",
|
|
"openid"
|
|
]
|
|
self.client_secret = "client_secret.json"
|
|
|
|
async def _create_youtube_service(self, access_token: str, refresh_token: Optional[str] = None):
|
|
'''YouTube API 서비스 생성'''
|
|
try:
|
|
client_config = load_client_config(self.client_secret)
|
|
|
|
# Credentials 객체 생성
|
|
credentials = Credentials(
|
|
token=access_token,
|
|
refresh_token=refresh_token,
|
|
token_uri="https://oauth2.googleapis.com/token",
|
|
client_id=client_config["web"]["client_id"],
|
|
client_secret=client_config["web"]["client_secret"],
|
|
scopes=self.scopes
|
|
)
|
|
|
|
# 토큰 만료 시 자동 갱신
|
|
if credentials.expired and credentials.refresh_token:
|
|
credentials.refresh(GoogleRequest())
|
|
|
|
# 갱신된 토큰을 Redis에 저장
|
|
updated_token_data = {
|
|
"access_token": credentials.token,
|
|
"refresh_token": credentials.refresh_token,
|
|
"expires_at": credentials.expiry.isoformat() if credentials.expiry else None
|
|
}
|
|
await self.oauth_storage.store_updated_token(access_token[:10], updated_token_data)
|
|
|
|
# YouTube API 서비스 생성
|
|
youtube = build('youtube', 'v3', credentials=credentials)
|
|
return youtube
|
|
|
|
except Exception as e:
|
|
raise HTTPException(status_code=401, detail=f"YouTube 서비스 생성 오류: {str(e)}")
|
|
|
|
async def get_all_channel_info(self, access_token: str, refresh_token: Optional[str] = None):
|
|
'''YouTube 모든 채널 정보 조회'''
|
|
try:
|
|
# 캐시 키 생성
|
|
cache_key = access_token[:10]
|
|
|
|
# 캐시된 데이터 확인
|
|
cached_data = await self.youtube_storage.get_cached_channels(cache_key)
|
|
if cached_data:
|
|
return cached_data
|
|
|
|
youtube = await self._create_youtube_service(access_token, refresh_token)
|
|
|
|
# 내 채널 정보 조회
|
|
response = youtube.channels().list(
|
|
part='snippet,statistics,contentDetails',
|
|
mine=True
|
|
).execute()
|
|
|
|
if not response.get('items'):
|
|
raise HTTPException(status_code=404, detail="YouTube 채널을 찾을 수 없습니다. YouTube 채널을 먼저 생성해주세요.")
|
|
|
|
# 모든 채널 정보를 리스트로 반환
|
|
channels = []
|
|
for channel in response['items']:
|
|
channel_info = {
|
|
"channel_id": channel['id'],
|
|
"channel_title": channel['snippet']['title'],
|
|
"description": channel['snippet']['description'],
|
|
"custom_url": channel['snippet'].get('customUrl', ''),
|
|
"published_at": channel['snippet']['publishedAt'],
|
|
"thumbnail": channel['snippet']['thumbnails'].get('default', {}).get('url'),
|
|
"high_res_thumbnail": channel['snippet']['thumbnails'].get('high', {}).get('url'),
|
|
"subscriber_count": channel['statistics'].get('subscriberCount', '0'),
|
|
"video_count": channel['statistics'].get('videoCount', '0'),
|
|
"view_count": channel['statistics'].get('viewCount', '0'),
|
|
"uploads_playlist_id": channel['contentDetails']['relatedPlaylists']['uploads'],
|
|
"country": channel['snippet'].get('country', ''),
|
|
"default_language": channel['snippet'].get('defaultLanguage', '')
|
|
}
|
|
channels.append(channel_info)
|
|
|
|
result = {
|
|
"total_channels": len(channels),
|
|
"channels": channels
|
|
}
|
|
|
|
# 결과를 Redis에 캐싱
|
|
await self.youtube_storage.cache_channels(cache_key, result)
|
|
|
|
return result
|
|
|
|
except HttpError as e:
|
|
error_content = json.loads(e.content.decode('utf-8'))
|
|
error_message = error_content.get('error', {}).get('message', str(e))
|
|
|
|
raise HTTPException(status_code=e.resp.status, detail=f"채널 정보 조회 실패: {error_message}")
|
|
|
|
except Exception as e:
|
|
raise HTTPException(status_code=500, detail=f"채널 정보 조회 오류: {str(e)}") |