63 lines
2.3 KiB
Python
63 lines
2.3 KiB
Python
import requests
|
|
from urllib.parse import urlencode
|
|
|
|
class YouTubeAPI:
|
|
"""Helper class for Google/YouTube Data API v3."""
|
|
|
|
AUTH_URL = "https://accounts.google.com/o/oauth2/v2/auth"
|
|
TOKEN_URL = "https://oauth2.googleapis.com/token"
|
|
|
|
def __init__(self, client_id, client_secret, redirect_uri):
|
|
self.client_id = client_id
|
|
self.client_secret = client_secret
|
|
self.redirect_uri = redirect_uri
|
|
|
|
def get_auth_url(self, state):
|
|
params = {
|
|
'client_id': self.client_id,
|
|
'redirect_uri': self.redirect_uri,
|
|
'response_type': 'code',
|
|
'scope': 'https://www.googleapis.com/auth/youtube.upload https://www.googleapis.com/auth/youtube.readonly',
|
|
'access_type': 'offline',
|
|
'prompt': 'consent',
|
|
'state': state
|
|
}
|
|
return f"{self.AUTH_URL}?{urlencode(params)}"
|
|
|
|
def exchange_code_for_token(self, code):
|
|
data = {
|
|
'grant_type': 'authorization_code',
|
|
'code': code,
|
|
'client_id': self.client_id,
|
|
'client_secret': self.client_secret,
|
|
'redirect_uri': self.redirect_uri
|
|
}
|
|
response = requests.post(self.TOKEN_URL, data=data, timeout=10)
|
|
response.raise_for_status()
|
|
return response.json()
|
|
|
|
def upload_video(self, access_token, title, description, video_data, category_id="22", privacy_status="public"):
|
|
"""Upload video to YouTube via Data API v3 multipart upload."""
|
|
import json
|
|
url = "https://www.googleapis.com/upload/youtube/v3/videos?uploadType=multipart&part=snippet,status"
|
|
headers = {
|
|
'Authorization': f'Bearer {access_token}',
|
|
}
|
|
metadata = {
|
|
"snippet": {
|
|
"title": title,
|
|
"description": description,
|
|
"categoryId": category_id
|
|
},
|
|
"status": {
|
|
"privacyStatus": privacy_status
|
|
}
|
|
}
|
|
files = {
|
|
'resource': (None, json.dumps(metadata), 'application/json; charset=UTF-8'),
|
|
'media': ('video.mp4', video_data, 'video/mp4')
|
|
}
|
|
response = requests.post(url, headers=headers, files=files, timeout=120)
|
|
response.raise_for_status()
|
|
return response.json()
|