How to create a job in the YouTube Studio ‘Reports’ section or via the API

import os
import pickle
import google_auth_oauthlib.flow
import googleapiclient.discovery
import googleapiclient.errors
from google.auth.transport.requests import Request

# --- CONFIGURATION ---
# Need read/write scope to create jobs
SCOPES = ['https://www.googleapis.com/auth/yt-analytics.readonly']
API_SERVICE_NAME = 'youtubereporting'
API_VERSION = 'v1'
CLIENT_SECRETS_FILE = 'client_secrets.json'
TOKEN_PICKLE_FILE = 'token.pickle' # Same file as the download script

# --- AUTHENTICATION ---
def get_authenticated_service():
    """Handles OAuth 2.0 flow and returns an authenticated API service."""
    creds = None
    if os.path.exists(TOKEN_PICKLE_FILE):
        with open(TOKEN_PICKLE_FILE, 'rb') as token:
            creds = pickle.load(token)

    if not creds or not creds.valid:
        if creds and creds.expired and creds.refresh_token:
            try:
                creds.refresh(Request())
            except Exception as e:
                 print(f"Error refreshing token: {e}. Re-authenticating.")
                 creds = None # Force re-authentication if refresh fails

        if not creds: # Need to authenticate
            if not os.path.exists(CLIENT_SECRETS_FILE):
                print(f"Error: {CLIENT_SECRETS_FILE} not found.")
                exit()
            flow = google_auth_oauthlib.flow.InstalledAppFlow.from_client_secrets_file(
                CLIENT_SECRETS_FILE, SCOPES)
            creds = flow.run_local_server(port=0)

        with open(TOKEN_PICKLE_FILE, 'wb') as token:
            pickle.dump(creds, token)

    return googleapiclient.discovery.build(
        API_SERVICE_NAME, API_VERSION, credentials=creds)

# --- List Available Report Types ---
def list_report_types(service):
    """Lists available report types."""
    print("Fetching available report types...")
    try:
        results = service.reportTypes().list().execute()
        report_types = results.get('reportTypes', [])

        if not report_types:
            print("No report types found.")
            return None
        else:
            print("Available Report Types:")
            for i, r_type in enumerate(report_types):
                print(f"  [{i+1}] ID: {r_type['id']}")
                print(f"      Name: {r_type['name']}")
            return report_types
    except googleapiclient.errors.HttpError as e:
        print(f"An API error occurred: {e}")
        return None


# --- Create Job Function ---
def create_reporting_job(service, report_type_id, job_name):
    """Creates a new reporting job."""
    print(f"\nCreating job '{job_name}' for report type '{report_type_id}'...")
    job_body = {
        'reportTypeId': report_type_id,
        'name': job_name
    }
    try:
        response = service.jobs().create(body=job_body).execute()
        print("\n--- Job Created Successfully ---")
        print(f"Job ID: {response['id']}")
        print(f"Job Name: {response['name']}")
        print(f"Report Type ID: {response['reportTypeId']}")
        print(f"Create Time: {response['createTime']}")
        print("--------------------------------")
        print("Reports will start generating within 48 hours.")
        return response
    except googleapiclient.errors.HttpError as e:
        print(f"An API error occurred: {e}")
        error_details = e.resp.get('content', b'').decode('utf-8')
        if "jobs.create" in str(e) and "409" in str(e): # 409 Conflict means it likely already exists
             print("Error: A job with this reportTypeId might already exist for your channel.")
        elif "403" in str(e) and "system-managed" in error_details.lower():
             print("Error: Cannot create jobs for system-managed report types.")
        else:
             print(f"Details: {error_details}")
        return None

# --- MAIN EXECUTION ---
def main():
    service = get_authenticated_service()

    available_types = list_report_types(service)
    if not available_types:
        return

    # --- User Input ---
    while True:
        try:
            choice = int(input(f"Enter the number of the report type you want to create a job for (1-{len(available_types)}): "))
            if 1 <= choice <= len(available_types):
                selected_report_type = available_types[choice-1]
                break
            else:
                print("Invalid choice.")
        except ValueError:
            print("Please enter a number.")

    job_name_input = input(f"Enter a name for this job (e.g., 'Daily {selected_report_type['name']}'): ")
    if not job_name_input:
         job_name_input = f"My {selected_report_type['name']} Job" # Default name

    # --- Create the Job ---
    create_reporting_job(service, selected_report_type['id'], job_name_input)

if __name__ == '__main__':
    main()