#!/usr/bin/env python3 """ YouTube OAuth 2.0 Authentication for Genesis Handles OAuth flow for accessing YouTube Data API v3, specifically for watch history access which requires user authentication. Setup Instructions: 1. Go to https://console.cloud.google.com/ 2. Create a new project or select existing 3. Enable YouTube Data API v3 4. Create OAuth 2.0 credentials (Desktop App type) 5. Download client_secret.json to config/youtube_client_secret.json 6. Run this script to complete OAuth flow Usage: python youtube_oauth.py --setup # Initial OAuth setup python youtube_oauth.py --refresh # Refresh expired token python youtube_oauth.py --test # Test API access """ import os import sys import json import argparse import logging from pathlib import Path from datetime import datetime, timedelta from typing import Optional, Dict, Any # Add genesis-system to path sys.path.insert(0, '/mnt/e/genesis-system') logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s') logger = logging.getLogger(__name__) # Paths GENESIS_ROOT = Path("/mnt/e/genesis-system") CONFIG_DIR = GENESIS_ROOT / "config" SECRETS_ENV = CONFIG_DIR / "secrets.env" CLIENT_SECRET_PATH = CONFIG_DIR / "youtube_client_secret.json" TOKEN_PATH = CONFIG_DIR / "youtube_token.json" # OAuth Scopes for watch history SCOPES = [ 'https://www.googleapis.com/auth/youtube.readonly', 'https://www.googleapis.com/auth/youtube.force-ssl' ] class YouTubeOAuth: """Handles YouTube OAuth 2.0 authentication flow.""" def __init__(self): self.credentials = None self._load_existing_token() def _load_existing_token(self) -> bool: """Load existing token if available.""" if TOKEN_PATH.exists(): try: with open(TOKEN_PATH, 'r') as f: token_data = json.load(f) # Check if token is still valid expiry = datetime.fromisoformat(token_data.get('expiry', '2000-01-01')) if expiry > datetime.now(): self.credentials = token_data logger.info("Loaded existing valid token") return True else: logger.info("Token expired, refresh required") return False except Exception as e: logger.warning(f"Failed to load token: {e}") return False return False def setup_oauth(self) -> bool: """ Run OAuth setup flow. Requires google-auth-oauthlib package: pip install google-auth-oauthlib google-auth-httplib2 google-api-python-client """ try: from google_auth_oauthlib.flow import InstalledAppFlow from google.auth.transport.requests import Request from google.oauth2.credentials import Credentials except ImportError: logger.error("Required packages not installed. Run:") logger.error("pip install google-auth-oauthlib google-auth-httplib2 google-api-python-client") return False # Check for client secret if not CLIENT_SECRET_PATH.exists(): logger.error(f"Client secret not found at: {CLIENT_SECRET_PATH}") logger.error("Download from Google Cloud Console > APIs & Services > Credentials") self._print_setup_instructions() return False try: # Run OAuth flow flow = InstalledAppFlow.from_client_secrets_file( str(CLIENT_SECRET_PATH), scopes=SCOPES ) logger.info("Opening browser for OAuth consent...") credentials = flow.run_local_server(port=8080) # Save token token_data = { 'token': credentials.token, 'refresh_token': credentials.refresh_token, 'token_uri': credentials.token_uri, 'client_id': credentials.client_id, 'client_secret': credentials.client_secret, 'scopes': list(credentials.scopes), 'expiry': credentials.expiry.isoformat() if credentials.expiry else None } with open(TOKEN_PATH, 'w') as f: json.dump(token_data, f, indent=2) # Also update secrets.env self._update_secrets_env(token_data) logger.info(f"OAuth setup complete! Token saved to: {TOKEN_PATH}") self.credentials = token_data return True except Exception as e: logger.error(f"OAuth setup failed: {e}") return False def refresh_token(self) -> bool: """Refresh an expired access token.""" try: from google.oauth2.credentials import Credentials from google.auth.transport.requests import Request except ImportError: logger.error("Required packages not installed") return False if not TOKEN_PATH.exists(): logger.error("No token to refresh. Run --setup first") return False try: with open(TOKEN_PATH, 'r') as f: token_data = json.load(f) credentials = Credentials( token=token_data['token'], refresh_token=token_data['refresh_token'], token_uri=token_data['token_uri'], client_id=token_data['client_id'], client_secret=token_data['client_secret'], scopes=token_data['scopes'] ) # Refresh credentials.refresh(Request()) # Save updated token token_data['token'] = credentials.token token_data['expiry'] = credentials.expiry.isoformat() if credentials.expiry else None with open(TOKEN_PATH, 'w') as f: json.dump(token_data, f, indent=2) self._update_secrets_env(token_data) logger.info("Token refreshed successfully!") self.credentials = token_data return True except Exception as e: logger.error(f"Token refresh failed: {e}") return False def _update_secrets_env(self, token_data: Dict[str, Any]): """Update secrets.env with YouTube OAuth tokens.""" secrets = {} # Read existing secrets if SECRETS_ENV.exists(): with open(SECRETS_ENV, 'r') as f: for line in f: line = line.strip() if line and not line.startswith('#') and '=' in line: key, value = line.split('=', 1) secrets[key] = value # Update YouTube tokens secrets['YOUTUBE_ACCESS_TOKEN'] = token_data['token'] secrets['YOUTUBE_REFRESH_TOKEN'] = token_data['refresh_token'] secrets['YOUTUBE_CLIENT_ID'] = token_data['client_id'] secrets['YOUTUBE_CLIENT_SECRET'] = token_data['client_secret'] # Write back with open(SECRETS_ENV, 'w') as f: f.write("# Genesis System Secrets Configuration\n") f.write("# NEVER commit this file to version control!\n\n") for key, value in sorted(secrets.items()): f.write(f"{key}={value}\n") logger.info(f"Updated {SECRETS_ENV} with YouTube credentials") def get_credentials(self): """Get credentials object for API calls.""" try: from google.oauth2.credentials import Credentials except ImportError: return None if not self.credentials: if not self._load_existing_token(): return None return Credentials( token=self.credentials['token'], refresh_token=self.credentials['refresh_token'], token_uri=self.credentials['token_uri'], client_id=self.credentials['client_id'], client_secret=self.credentials['client_secret'], scopes=self.credentials['scopes'] ) def test_api_access(self) -> bool: """Test YouTube API access by fetching channel info.""" try: from googleapiclient.discovery import build except ImportError: logger.error("google-api-python-client not installed") return False credentials = self.get_credentials() if not credentials: logger.error("No valid credentials. Run --setup first") return False try: youtube = build('youtube', 'v3', credentials=credentials) # Test: Get authenticated user's channel request = youtube.channels().list( part='snippet', mine=True ) response = request.execute() if response.get('items'): channel = response['items'][0]['snippet'] logger.info(f"API Access OK! Connected as: {channel.get('title')}") return True else: logger.warning("API responded but no channel found") return False except Exception as e: logger.error(f"API test failed: {e}") return False def get_watch_history(self, max_results: int = 50) -> list: """ Fetch watch history from YouTube. Note: YouTube API v3 doesn't directly expose watch history. This uses the 'myRating' endpoint as a proxy, or we need to use the YouTube Data API with the 'playlistItems' for the special 'HL' (History) playlist. """ try: from googleapiclient.discovery import build except ImportError: logger.error("google-api-python-client not installed") return [] credentials = self.get_credentials() if not credentials: logger.error("No valid credentials") return [] try: youtube = build('youtube', 'v3', credentials=credentials) # Method 1: Try to access watch history playlist # The watch history playlist ID is 'HL' (History List) # But this may not be accessible via API # Method 2: Get liked videos as proxy for engagement # This shows videos the user has interacted with request = youtube.videos().list( part='snippet,contentDetails', myRating='like', maxResults=max_results ) response = request.execute() videos = [] for item in response.get('items', []): videos.append({ 'video_id': item['id'], 'title': item['snippet']['title'], 'channel': item['snippet']['channelTitle'], 'published_at': item['snippet']['publishedAt'], 'duration': item['contentDetails']['duration'] }) logger.info(f"Retrieved {len(videos)} videos from history/liked") return videos except Exception as e: logger.error(f"Failed to get watch history: {e}") return [] def _print_setup_instructions(self): """Print detailed setup instructions.""" print(""" ╔══════════════════════════════════════════════════════════════════════╗ ║ YouTube Data API Setup Instructions ║ ╠══════════════════════════════════════════════════════════════════════╣ ║ ║ ║ 1. Go to: https://console.cloud.google.com/ ║ ║ ║ ║ 2. Create a new project (or select existing): ║ ║ - Click project dropdown → "New Project" ║ ║ - Name: "Genesis YouTube Pipeline" ║ ║ ║ ║ 3. Enable YouTube Data API v3: ║ ║ - Go to: APIs & Services → Library ║ ║ - Search "YouTube Data API v3" ║ ║ - Click "Enable" ║ ║ ║ ║ 4. Create OAuth Credentials: ║ ║ - Go to: APIs & Services → Credentials ║ ║ - Click "Create Credentials" → "OAuth client ID" ║ ║ - Application type: "Desktop app" ║ ║ - Name: "Genesis Desktop Client" ║ ║ ║ ║ 5. Configure OAuth Consent Screen: ║ ║ - User type: "External" ║ ║ - Add your email as test user ║ ║ - Scopes: youtube.readonly ║ ║ ║ ║ 6. Download client secret: ║ ║ - Click download icon on your OAuth client ║ ║ - Save as: config/youtube_client_secret.json ║ ║ ║ ║ 7. Run this script again: ║ ║ python youtube_oauth.py --setup ║ ║ ║ ╚══════════════════════════════════════════════════════════════════════╝ """) def main(): parser = argparse.ArgumentParser(description="YouTube OAuth Setup for Genesis") parser.add_argument('--setup', action='store_true', help='Run OAuth setup flow') parser.add_argument('--refresh', action='store_true', help='Refresh expired token') parser.add_argument('--test', action='store_true', help='Test API access') parser.add_argument('--history', action='store_true', help='Fetch watch history') args = parser.parse_args() oauth = YouTubeOAuth() if args.setup: success = oauth.setup_oauth() sys.exit(0 if success else 1) elif args.refresh: success = oauth.refresh_token() sys.exit(0 if success else 1) elif args.test: success = oauth.test_api_access() sys.exit(0 if success else 1) elif args.history: videos = oauth.get_watch_history() print(json.dumps(videos, indent=2)) sys.exit(0 if videos else 1) else: # Check current status if oauth.credentials: print("✓ YouTube OAuth configured") print(f" Token expires: {oauth.credentials.get('expiry', 'unknown')}") print("\nRun --test to verify API access") else: print("✗ YouTube OAuth not configured") print("\nRun --setup to begin OAuth flow") oauth._print_setup_instructions() if __name__ == "__main__": main()