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

# OAuth 2.0 scopes
SCOPES = ["https://www.googleapis.com/auth/youtube.readonly"]

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
    
    if "--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")
    
    if creds:
        print("Authentication successful!")
    else:
        print("Authentication failed!")
        sys.exit(1)