YouTube_predownloader/auth.py

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)