youtube_manager.py

import os
import logging
import time
from pathlib import Path
from typing import Optional

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
    from googleapiclient.errors import HttpError
except ImportError:
    pass

# Same scope as responder to share token
SCOPES = ['https://www.googleapis.com/auth/youtube.force-ssl']

class YouTubeManager:
    """
    Manages YouTube Broadcast Metadata (Title, Description, Tags).
    Uses the same token.json as YouTubeResponder.
    """
    def __init__(self, token_file='token.json', client_secret='client_secret.json', mock=False):
        self.logger = logging.getLogger(__name__)
        self.base_dir = Path(__file__).parent
        self.token_path = self.base_dir / token_file
        self.client_secrets_path = self.base_dir / client_secret
        self.youtube = None
        self.credentials = None
        self.mock = mock
        
        # Rate Limiting
        self.last_update_time = 0
        self.update_cooldown = 300 # 5 minutes

    def authenticate(self) -> bool:
        """Load credentials or start OAuth flow"""
        if self.mock:
            self.logger.info("🔐 MOCK: YouTube Auth successful")
            return True

        try:
            if self.token_path.exists():
                self.credentials = Credentials.from_authorized_user_file(
                    str(self.token_path), SCOPES
                )
            
            if not self.credentials or not self.credentials.valid:
                if self.credentials and self.credentials.expired and self.credentials.refresh_token:
                    self.logger.info("🔄 Refreshing access token...")
                    self.credentials.refresh(Request())
                else:
                    self.logger.info("🔐 Starting new OAuth flow for YouTube Manager...")
                    flow = InstalledAppFlow.from_client_secrets_file(
                        str(self.client_secrets_path), SCOPES
                    )
                    self.credentials = flow.run_local_server(port=0)
                
                # Save
                with open(self.token_path, 'w') as token:
                    token.write(self.credentials.to_json())
            
            self.youtube = build('youtube', 'v3', credentials=self.credentials)
            return True
            
        except Exception as e:
            self.logger.error(f"❌ Auth Failed: {e}")
            return False

    def update_broadcast(self, video_id: str, title: str, description: str = None) -> bool:
        """
        Update the title/description of a live broadcast.
        Costs ~50 units. Use sparingly.
        """
        if not self.youtube: 
            if not self.authenticate(): return False

        # Rate Limit Check
        # For mock, we might want to bypass cooldown for testing?
        # But let's keep it to verify logic, unless forced.
        # Actually, for dress rehearsal, we want to see it happen fast.
        # So if mock, override cooldown? No, dress rehearsal waits 10s+.
        if not self.mock and time.time() - self.last_update_time < self.update_cooldown:
            self.logger.warning(f"⏳ Update skipped: Cooldown active ({self.update_cooldown}s)")
            return False
            
        try:
            self.logger.info(f"📡 Updating Broadcast Metadata for {video_id}...")
            
            if self.mock:
                self.logger.info(f"✅ YouTube Updated (MOCK): '{title}'")
                self.last_update_time = time.time()
                return True

            # 1. Get current broadcast details (to get snippet)
            # We need the 'snippet' to update it.
            # Actually, liveBroadcasts.update requires 'id' and 'snippet'.
            
            # Note: video_id is often same as broadcast_id, need to verifying.
            # Usually we need to look up broadcast ID from video ID?
            # Or assume video_id == broadcast_id for simple cases?
            # Let's try to find the broadcast ID associated with this video ID.
            
            request = self.youtube.liveBroadcasts().list(
                part="id,snippet",
                id=video_id 
            )
            response = request.execute()
            
            if not response.get("items"):
                # Maybe video_id != broadcast_id? 
                # Try searching by status=active? No, that returns all.
                self.logger.error(f"❌ Broadcast not found for ID: {video_id}")
                return False
                
            broadcast = response["items"][0]
            snippet = broadcast["snippet"]
            
            # Update fields
            snippet["title"] = title
            if description:
                snippet["description"] = description
            
            # specific call to update
            update_request = self.youtube.liveBroadcasts().update(
                part="snippet",
                body={
                    "id": video_id,
                    "snippet": snippet
                }
            )
            update_response = update_request.execute()
            
            self.logger.info(f"✅ YouTube Updated: '{title}'")
            self.last_update_time = time.time()
            return True
            
        except HttpError as e:
            self.logger.error(f"❌ YouTube API Error: {e}")
            return False
        except Exception as e:
            self.logger.error(f"❌ Update Failed: {e}")
            return False

if __name__ == "__main__":
    # Test CLI
    import sys
    logging.basicConfig(level=logging.INFO)
    mgr = YouTubeManager()
    
    if len(sys.argv) > 2:
        vid = sys.argv[1]
        new_title = sys.argv[2]
        mgr.update_broadcast(vid, new_title)
    else:
        print("Usage: python3 youtube_manager.py <video_id> <new_title>")