""" YouTube Data API v3 Uploader Module. Handles OAuth2 authentication, video uploading, custom thumbnail uploading, multi-language captions (.srt), YouTube chapters embedding, and metadata publishing. """ import os import json import logging from typing import Dict, Any, Optional, List logger = logging.getLogger("YTFactory.YouTube") SCOPES = [ "https://www.googleapis.com/auth/youtube.upload", "https://www.googleapis.com/auth/youtube.force-ssl" ] class YouTubeUploader: def __init__( self, client_secrets_file: str = "client_secrets.json", token_file: str = "youtube_token.json", default_privacy: str = "private" ): self.client_secrets_file = client_secrets_file self.token_file = token_file self.default_privacy = default_privacy self.youtube_service = None def is_configured(self) -> bool: """Check if OAuth client secrets file exists.""" return os.path.exists(self.client_secrets_file) or os.path.exists(self.token_file) def get_service(self): """Authenticate and return YouTube API service client.""" if self.youtube_service: return self.youtube_service try: from google.oauth2.credentials import Credentials from google_auth_oauthlib.flow import InstalledAppFlow from google.auth.transport.requests import Request from googleapiclient.discovery import build creds = None if os.path.exists(self.token_file): creds = Credentials.from_authorized_user_file(self.token_file, SCOPES) if not creds or not creds.valid: if creds and creds.expired and creds.refresh_token: creds.refresh(Request()) else: if not os.path.exists(self.client_secrets_file): raise FileNotFoundError( f"YouTube OAuth secrets file '{self.client_secrets_file}' not found.\n" "To enable YouTube API uploads, download your client_secrets.json from Google Cloud Console." ) flow = InstalledAppFlow.from_client_secrets_file(self.client_secrets_file, SCOPES) creds = flow.run_local_server(port=0) with open(self.token_file, "w") as token: token.write(creds.to_json()) self.youtube_service = build("youtube", "v3", credentials=creds) return self.youtube_service except Exception as e: logger.error(f"[YOUTUBE] Authentication failed: {e}") raise def upload_video( self, video_path: str, title: str, description: str, tags: List[str], category_id: str = "27", # Education privacy_status: Optional[str] = None ) -> str: """Upload video file to YouTube and return video ID.""" from googleapiclient.http import MediaFileUpload service = self.get_service() privacy = privacy_status or self.default_privacy logger.info(f"[YOUTUBE] Uploading video '{title}' ({privacy})...") body = { "snippet": { "title": title[:100], "description": description, "tags": tags, "categoryId": category_id }, "status": { "privacyStatus": privacy, "selfDeclaredMadeForKids": False } } media = MediaFileUpload(video_path, chunksize=-1, resumable=True) request = service.videos().insert(part="snippet,status", body=body, media_body=media) response = None while response is None: status, response = request.next_chunk() if status: logger.info(f"[YOUTUBE] Upload progress: {int(status.progress() * 100)}%") video_id = response.get("id") logger.info(f"[YOUTUBE] Video uploaded successfully! ID: {video_id} -> https://youtu.be/{video_id}") return video_id def set_thumbnail(self, video_id: str, thumbnail_path: str) -> bool: """Upload custom thumbnail for a video.""" from googleapiclient.http import MediaFileUpload if not os.path.exists(thumbnail_path): logger.warning(f"[YOUTUBE] Thumbnail file not found: {thumbnail_path}") return False try: service = self.get_service() logger.info(f"[YOUTUBE] Setting thumbnail for video {video_id}...") media = MediaFileUpload(thumbnail_path, mimetype="image/png") service.thumbnails().set(videoId=video_id, media_body=media).execute() logger.info(f"[YOUTUBE] Thumbnail set successfully for {video_id}") return True except Exception as e: logger.error(f"[YOUTUBE] Failed to set thumbnail: {e}") return False def upload_caption(self, video_id: str, srt_path: str, lang: str = "en", name: str = "English") -> bool: """Upload subtitle/caption track for a video.""" from googleapiclient.http import MediaFileUpload if not os.path.exists(srt_path): return False try: service = self.get_service() logger.info(f"[YOUTUBE] Uploading {lang} captions for video {video_id}...") body = { "snippet": { "videoId": video_id, "language": lang, "name": name, "isDraft": False } } media = MediaFileUpload(srt_path, mimetype="application/x-subrip") service.captions().insert(part="snippet", body=body, media_body=media).execute() logger.info(f"[YOUTUBE] {lang} captions uploaded successfully.") return True except Exception as e: logger.error(f"[YOUTUBE] Failed to upload captions: {e}") return False def upload_bundle(self, bundle_meta_path: str, lang: str = "en", privacy_status: Optional[str] = None) -> Dict[str, Any]: """Upload a complete packaged video bundle (video + metadata + thumbnail + captions).""" with open(bundle_meta_path, "r", encoding="utf-8") as f: meta = json.load(f) video_path = meta.get("files", {}).get(lang) if not video_path or not os.path.exists(video_path): raise FileNotFoundError(f"Video file for language '{lang}' not found in bundle.") title = meta.get("titles", ["Documentary"])[0] if lang == "en" else meta.get("titles_es", meta.get("titles", ["Documentary"]))[0] desc = meta.get("description", "") if lang == "en" else meta.get("description_es", meta.get("description", "")) tags = meta.get("tags", []) # Select best thumbnail thumb_list = meta.get("thumbnails", {}).get(lang, []) or meta.get("thumbnails_raw", []) thumb_path = thumb_list[0] if thumb_list else None srt_path = meta.get("subtitles", {}).get(lang) # 1. Upload Video video_id = self.upload_video(video_path, title, desc, tags, privacy_status=privacy_status) # 2. Upload Thumbnail if thumb_path: self.set_thumbnail(video_id, thumb_path) # 3. Upload Captions if srt_path: caption_name = "English" if lang == "en" else "Spanish" self.upload_caption(video_id, srt_path, lang=lang, name=caption_name) return { "video_id": video_id, "url": f"https://youtu.be/{video_id}", "title": title, "language": lang, "privacy": privacy_status or self.default_privacy }