133 lines
5.4 KiB
Python
133 lines
5.4 KiB
Python
# -*- coding: utf-8 -*-
|
|
"""유튜브(Google OAuth + YouTube Data API v3) 클라이언트.
|
|
|
|
castad app/social/oauth/youtube.py + uploader/youtube.py 이식·축약.
|
|
- OAuth: 인증 URL 생성 → 코드 교환(access/refresh) → 리프레시 → 내 채널 정보
|
|
- 업로드: resumable upload 세션 시작 → 파일 PUT → video_id
|
|
"""
|
|
from urllib.parse import urlencode
|
|
|
|
import aiohttp
|
|
from fastapi import HTTPException
|
|
|
|
from .config import settings
|
|
|
|
AUTHORIZATION_URL = "https://accounts.google.com/o/oauth2/v2/auth"
|
|
TOKEN_URL = "https://oauth2.googleapis.com/token"
|
|
CHANNELS_URL = "https://www.googleapis.com/youtube/v3/channels"
|
|
UPLOAD_URL = "https://www.googleapis.com/upload/youtube/v3/videos"
|
|
|
|
SCOPES = [
|
|
"https://www.googleapis.com/auth/youtube.upload",
|
|
"https://www.googleapis.com/auth/youtube.readonly",
|
|
]
|
|
|
|
|
|
def get_authorization_url(state: str) -> str:
|
|
if not (settings.YOUTUBE_CLIENT_ID and settings.YOUTUBE_CLIENT_SECRET):
|
|
raise HTTPException(501, "유튜브 연동 미설정 — YOUTUBE_CLIENT_ID/SECRET 필요")
|
|
params = {
|
|
"client_id": settings.YOUTUBE_CLIENT_ID,
|
|
"redirect_uri": settings.YOUTUBE_REDIRECT_URI,
|
|
"response_type": "code",
|
|
"scope": " ".join(SCOPES),
|
|
"access_type": "offline", # refresh_token 발급
|
|
"prompt": "consent", # 매번 동의 → refresh_token 보장
|
|
"state": state,
|
|
}
|
|
return f"{AUTHORIZATION_URL}?{urlencode(params)}"
|
|
|
|
|
|
async def exchange_code(code: str) -> dict:
|
|
"""인가코드 → {access_token, refresh_token, expires_in}."""
|
|
data = {
|
|
"client_id": settings.YOUTUBE_CLIENT_ID,
|
|
"client_secret": settings.YOUTUBE_CLIENT_SECRET,
|
|
"code": code,
|
|
"grant_type": "authorization_code",
|
|
"redirect_uri": settings.YOUTUBE_REDIRECT_URI,
|
|
}
|
|
async with aiohttp.ClientSession() as s:
|
|
async with s.post(TOKEN_URL, data=data) as r:
|
|
res = await r.json()
|
|
if "error" in res:
|
|
raise HTTPException(400, f"유튜브 토큰 교환 실패: {res.get('error_description', res.get('error'))}")
|
|
return res
|
|
|
|
|
|
async def refresh_access_token(refresh_token: str) -> str:
|
|
data = {
|
|
"client_id": settings.YOUTUBE_CLIENT_ID,
|
|
"client_secret": settings.YOUTUBE_CLIENT_SECRET,
|
|
"refresh_token": refresh_token,
|
|
"grant_type": "refresh_token",
|
|
}
|
|
async with aiohttp.ClientSession() as s:
|
|
async with s.post(TOKEN_URL, data=data) as r:
|
|
res = await r.json()
|
|
if "error" in res:
|
|
raise HTTPException(401, f"유튜브 토큰 갱신 실패: {res.get('error_description', res.get('error'))}")
|
|
return res["access_token"]
|
|
|
|
|
|
async def get_channel_info(access_token: str) -> dict:
|
|
"""내 유튜브 채널 → {id, title}."""
|
|
headers = {"Authorization": f"Bearer {access_token}"}
|
|
params = {"part": "snippet", "mine": "true"}
|
|
async with aiohttp.ClientSession() as s:
|
|
async with s.get(CHANNELS_URL, headers=headers, params=params) as r:
|
|
data = await r.json()
|
|
items = data.get("items") or []
|
|
if not items:
|
|
return {"id": "", "title": "내 채널"}
|
|
return {"id": items[0]["id"], "title": items[0]["snippet"]["title"]}
|
|
|
|
|
|
async def upload_video(video_path: str, access_token: str, title: str,
|
|
description: str = "", tags: list[str] | None = None,
|
|
privacy: str = "public") -> dict:
|
|
"""영상 업로드(resumable). → {video_id, url}."""
|
|
import os
|
|
if not os.path.exists(video_path):
|
|
raise HTTPException(404, f"업로드할 파일이 없습니다: {video_path}")
|
|
file_size = os.path.getsize(video_path)
|
|
body = {
|
|
"snippet": {
|
|
"title": title[:100] or "썰박스",
|
|
"description": (description or "").replace("<", "").replace(">", ""),
|
|
"tags": tags or ["썰박스", "쇼츠", "홍보"],
|
|
"categoryId": "22",
|
|
},
|
|
"status": {"privacyStatus": privacy, "selfDeclaredMadeForKids": False},
|
|
}
|
|
init_headers = {
|
|
"Authorization": f"Bearer {access_token}",
|
|
"Content-Type": "application/json; charset=utf-8",
|
|
"X-Upload-Content-Type": "video/*",
|
|
"X-Upload-Content-Length": str(file_size),
|
|
}
|
|
async with aiohttp.ClientSession() as s:
|
|
# 1) resumable 세션 시작
|
|
async with s.post(UPLOAD_URL, params={"uploadType": "resumable", "part": "snippet,status"},
|
|
headers=init_headers, json=body) as r:
|
|
if r.status != 200:
|
|
err = await r.json()
|
|
raise HTTPException(502, f"유튜브 업로드 세션 실패: {err.get('error', {}).get('message', r.status)}")
|
|
upload_url = r.headers.get("Location")
|
|
if not upload_url:
|
|
raise HTTPException(502, "유튜브 업로드 URL을 받지 못했습니다.")
|
|
# 2) 파일 업로드(단일 PUT — 쇼츠는 소용량)
|
|
with open(video_path, "rb") as f:
|
|
content = f.read()
|
|
put_headers = {
|
|
"Authorization": f"Bearer {access_token}",
|
|
"Content-Type": "video/*",
|
|
"Content-Length": str(file_size),
|
|
}
|
|
async with s.put(upload_url, headers=put_headers, data=content) as r:
|
|
res = await r.json()
|
|
vid = res.get("id")
|
|
if not vid:
|
|
raise HTTPException(502, f"유튜브 업로드 실패: {res.get('error', {}).get('message', '알 수 없는 오류')}")
|
|
return {"video_id": vid, "url": f"https://www.youtube.com/watch?v={vid}"}
|