from google_auth_oauthlib.flow import InstalledAppFlow from google.auth.transport.requests import Request from google.oauth2 import service_account import os import pickle import json import sqlite3 import subprocess import shutil import tempfile from pathlib import Path from datetime import datetime import logging # OAuth 2.0 scopes SCOPES = ["https://www.googleapis.com/auth/youtube.readonly"] # Configure logging logger = logging.getLogger(__name__) def extract_chrome_cookies(output_file="youtube_cookies.txt"): """Extract YouTube cookies from Chrome browser.""" try: # Common Chrome cookie database locations chrome_paths = [ Path.home() / ".config/google-chrome/Default/Cookies", Path.home() / ".config/google-chrome/Profile 1/Cookies", Path.home() / ".config/chromium/Default/Cookies", Path.home() / "snap/chromium/common/chromium/Default/Cookies", Path.home() / ".var/app/com.google.Chrome/config/google-chrome/Default/Cookies", ] # Find existing Chrome cookies database chrome_cookies_db = None for path in chrome_paths: if path.exists(): chrome_cookies_db = path logger.info(f"Found Chrome cookies database: {chrome_cookies_db}") break if not chrome_cookies_db: logger.warning("Chrome cookies database not found") return False # Use yt-dlp to extract cookies directly from browser try: cmd = ['yt-dlp', '--cookies-from-browser', 'chrome', '--print-to-file', 'cookies', output_file, '--no-download', 'https://www.youtube.com'] result = subprocess.run(cmd, capture_output=True, text=True, timeout=30) if result.returncode == 0: logger.info(f"Successfully extracted Chrome cookies to {output_file}") return True else: logger.error(f"yt-dlp cookie extraction failed: {result.stderr}") return False except subprocess.TimeoutExpired: logger.error("Cookie extraction timed out") return False except Exception as e: logger.error(f"Error running yt-dlp cookie extraction: {e}") return False except Exception as e: logger.error(f"Error extracting Chrome cookies: {e}") return False def check_browser_cookies(): """Check which browsers have accessible cookies for YouTube.""" browsers = ['chrome', 'chromium', 'firefox', 'brave', 'edge'] available_browsers = [] for browser in browsers: try: # Test if browser cookies are accessible cmd = ['yt-dlp', '--cookies-from-browser', browser, '--simulate', '--quiet', 'https://www.youtube.com/watch?v=dQw4w9WgXcQ'] result = subprocess.run(cmd, capture_output=True, text=True, timeout=10) if result.returncode == 0: available_browsers.append(browser) logger.info(f"Browser {browser} has accessible cookies") else: logger.debug(f"Browser {browser} cookies not accessible: {result.stderr}") except (subprocess.TimeoutExpired, subprocess.SubprocessError) as e: logger.debug(f"Error checking {browser} cookies: {e}") continue return available_browsers def update_cookies_from_browser(browser='chrome', output_file="youtube_cookies.txt"): """Update cookies file from specified browser.""" try: logger.info(f"Updating cookies from {browser} browser...") # Use yt-dlp to extract cookies from browser cmd = ['yt-dlp', '--cookies-from-browser', browser, '--print-to-file', 'cookies', output_file, '--no-download', 'https://www.youtube.com'] result = subprocess.run(cmd, capture_output=True, text=True, timeout=30) if result.returncode == 0: logger.info(f"Successfully updated cookies from {browser} to {output_file}") # Verify the cookies file was created and has content cookies_path = Path(output_file) if cookies_path.exists() and cookies_path.stat().st_size > 0: logger.info(f"Cookies file created: {cookies_path} ({cookies_path.stat().st_size} bytes)") return True else: logger.warning("Cookies file was not created or is empty") return False else: logger.error(f"Failed to extract cookies from {browser}: {result.stderr}") return False except subprocess.TimeoutExpired: logger.error("Cookie extraction timed out") return False except Exception as e: logger.error(f"Error updating cookies from browser: {e}") return False def get_service_account_token(): """Authenticate using service account (no user interaction needed).""" service_account_file = "service_account.json" if not os.path.exists(service_account_file): print(f"Service account file {service_account_file} not found.") print("To use service account authentication:") print("1. Go to Google Cloud Console") print("2. Navigate to IAM & Admin > Service Accounts") print("3. Create a new service account") print("4. Download the JSON key file as 'service_account.json'") return None try: creds = service_account.Credentials.from_service_account_file( service_account_file, scopes=SCOPES) print("Service account authentication successful!") return creds except Exception as e: print(f"Service account authentication failed: {e}") return None def get_token_interactive(): """Perform OAuth authentication with user interaction.""" creds = None token_file = "youtube_token.pickle" # Check if token file exists if os.path.exists(token_file): with open(token_file, "rb") as token: creds = pickle.load(token) # If no valid credentials, authenticate user if not creds or not creds.valid: if creds and creds.expired and creds.refresh_token: try: creds.refresh(Request()) print("Token refreshed successfully!") except Exception as e: print(f"Token refresh failed: {e}") creds = None if not creds: print("Starting interactive authentication...") flow = InstalledAppFlow.from_client_secrets_file("client_secret.json", SCOPES) creds = flow.run_local_server(port=8080) # Save the credentials to a file with open(token_file, "wb") as token: pickle.dump(creds, token) print(f"Token saved to {token_file}") else: print("Token already exists and is valid.") return creds def get_token_non_interactive(): """Try non-interactive authentication methods first.""" # Try service account first creds = get_service_account_token() if creds: return creds # Try existing saved token token_file = "youtube_token.pickle" if os.path.exists(token_file): try: with open(token_file, "rb") as token: creds = pickle.load(token) if creds and creds.valid: print("Using existing valid token.") return creds elif creds and creds.expired and creds.refresh_token: try: creds.refresh(Request()) print("Token refreshed successfully!") # Save refreshed token with open(token_file, "wb") as token: pickle.dump(creds, token) return creds except Exception as e: print(f"Token refresh failed: {e}") except Exception as e: print(f"Error loading saved token: {e}") print("No valid non-interactive authentication method available.") print("Run with --interactive flag or set up service account authentication.") return None def get_token(): """Main authentication function - tries non-interactive first.""" return get_token_non_interactive() if __name__ == "__main__": import sys # Set up logging logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s') if "--extract-cookies" in sys.argv: # Extract cookies from Chrome browser = 'chrome' if "--browser" in sys.argv: browser_idx = sys.argv.index("--browser") + 1 if browser_idx < len(sys.argv): browser = sys.argv[browser_idx] output_file = "youtube_cookies.txt" if "--output" in sys.argv: output_idx = sys.argv.index("--output") + 1 if output_idx < len(sys.argv): output_file = sys.argv[output_idx] print(f"Extracting cookies from {browser} browser...") success = update_cookies_from_browser(browser, output_file) if success: print(f"Cookies successfully extracted to {output_file}") else: print("Cookie extraction failed") sys.exit(1) elif "--check-browsers" in sys.argv: # Check available browsers print("Checking available browsers with YouTube cookies...") browsers = check_browser_cookies() if browsers: print(f"Available browsers: {', '.join(browsers)}") else: print("No browsers with accessible YouTube cookies found") elif "--interactive" in sys.argv: creds = get_token_interactive() else: creds = get_token_non_interactive() if not creds: print("\nTo authenticate interactively, run: python3 auth.py --interactive") print("To extract cookies from browser, run: python3 auth.py --extract-cookies") print("To check available browsers, run: python3 auth.py --check-browsers") if "--extract-cookies" not in sys.argv and "--check-browsers" not in sys.argv: if creds: print("Authentication successful!") else: print("Authentication failed!") sys.exit(1)