scheduler.py

from google.oauth2.credentials import Credentials
from google_auth_oauthlib.flow import InstalledAppFlow
from google.auth.transport.requests import Request
from googleapiclient.discovery import build
import os

# Define the scopes required for managing live broadcasts
SCOPES = ['https://www.googleapis.com/auth/youtube',
          'https://www.googleapis.com/auth/youtube.force-ssl']

#Text files with metadata
metadata_text_AM = "/srv/dev-disk-by-uuid-49e4aa1a-279a-45ce-b23e-467c2b2a4162/AppData/jobson/specs/youtube-livestream.spec/metadata-AM.txt"
metadata_text_PM = "/srv/dev-disk-by-uuid-49e4aa1a-279a-45ce-b23e-467c2b2a4162/AppData/jobson/specs/youtube-livestream.spec/metadata-PM.txt"

# Function to read metadata from the AM file
def read_metadata_AM(filename):
    with open(filename, "r") as f:
        metadata_AM = {}
        for line in f:
            key, value = line.strip().split("=")
            metadata_AM[key.strip()] = value.strip().replace("\\n", "\n")
    return metadata_AM

# Function to read metadata from the PM file
def read_metadata_PM(filename):
    with open(metadata_text_PM, "r") as f:
        metadata_PM = {}
        for line in f:
            key, value = line.strip().split("=")
            metadata_PM[key.strip()] = value.strip().replace("\\n", "\n")
    return metadata_PM

# Read metadata from the files
metadata_AM = read_metadata_AM(metadata_text_AM)
metadata_PM = read_metadata_PM(metadata_text_PM)

#authentication filepaths and files
ls_token = '/srv/dev-disk-by-uuid-49e4aa1a-279a-45ce-b23e-467c2b2a4162/AppData/youtubelivestream/token.json'
ls_credentials = '/srv/dev-disk-by-uuid-49e4aa1a-279a-45ce-b23e-467c2b2a4162/AppData/youtubelivestream/credentials.json'

#video ids filepaths and files
video_id_text = "/srv/dev-disk-by-uuid-49e4aa1a-279a-45ce-b23e-467c2b2a4162/AppData/jobson/specs/youtube-livestream.spec/video_id.txt"

#AM service metadata
ls_title_AM = metadata_AM.get("ls_title_AM")
ls_description_AM = metadata_AM.get("ls_description_AM")
ls_date_AM = metadata_AM.get("ls_date_AM")
ls_privacy_AM = "public"
thumbnail_AM = "/srv/dev-disk-by-uuid-49e4aa1a-279a-45ce-b23e-467c2b2a4162/WorkFolder/Images/Thumbnails/AMServiceThumbnail.png"
ls_category_AM = "29" # Category ID for Nonprofits & Activism

#PM service metadata
ls_title_PM = metadata_PM.get("ls_title_PM")
ls_description_PM = metadata_PM.get("ls_description_PM")
ls_date_PM = metadata_PM.get("ls_date_PM")
ls_privacy_PM = "public"
thumbnail_PM = "/srv/dev-disk-by-uuid-49e4aa1a-279a-45ce-b23e-467c2b2a4162/WorkFolder/Images/Thumbnails/PMServiceThumbnail.png"
ls_category_PM = "29" # Category ID for Nonprofits & Activism

#temporary stream key to change the PM service to in order to not conflict with the AM service
desired_stream_key = "main_stream_key_example"

# Function to authenticate and build the YouTube API client
def get_authenticated_service():
    creds = None
    if os.path.exists(ls_token):
        creds = Credentials.from_authorized_user_file(ls_token)
    if not creds or not creds.valid:
        if creds and creds.expired and creds.refresh_token:
            creds.refresh(Request())
        else:
            flow = InstalledAppFlow.from_client_secrets_file(
                ls_credentials, SCOPES, access_type='offline')
            creds = flow.run_local_server(port=0)
        with open(ls_token, 'w') as token:
            token.write(creds.to_json())
    return build('youtube', 'v3', credentials=creds)

# Function to schedule a live video on YouTube
def schedule_live_video(youtube, title, description, start_time, privacy_status):
    request_body = {
        "snippet": {
            "title": title,
            "description": description,
            "scheduledStartTime": start_time,
        },
        "status": {
            "privacyStatus": privacy_status
        },
        "contentDetails": {
            "enableAutoStart": True,
            "enableAutoStop": True,
            "enableLowLatency": True,
            "enableDvr": True,
            "madeForKids": "false",
            "selfDeclaredMadeForKids": "false"
        }
    }
    live_broadcast = youtube.liveBroadcasts().insert(
        part="snippet,status,contentDetails",
        body=request_body
    ).execute()

    return live_broadcast['id']

from googleapiclient.http import MediaFileUpload

def upload_thumbnail(youtube, video_id, thumbnail_filename):
    # Create a MediaFileUpload object for the thumbnail
    media = MediaFileUpload(thumbnail_filename, chunksize=-1, resumable=True)

    # Call the YouTube API to set the thumbnail for the video
    request = youtube.thumbnails().set(
        videoId=video_id,
        media_body=media
    )
    response = request.execute()
    return response

def get_live_stream_id_by_key(youtube, stream_key):
    request = youtube.liveStreams().list(
        part="id,cdn",
        mine=True,
        fields="items(id,cdn/ingestionInfo/streamName)",
        maxResults=50
    )
    response = request.execute()
    for item in response.get('items', []):
        stream_id = item.get('id')
        cdn_info = item.get('cdn', {})
        ingestion_info = cdn_info.get('ingestionInfo', {})
        if ingestion_info.get('streamName') == stream_key:
            return stream_id
    return None

# Function to modify the category of a live video
def modify_video_category(youtube, video_id, category_id):
    video = youtube.videos().list(
        part="snippet",
        id=video_id
    ).execute()
    snippet = video['items'][0]['snippet']
    snippet['categoryId'] = category_id

    request_body = {
        "id": video_id,
        "snippet": snippet
    }
    youtube.videos().update(
        part="snippet",
        body=request_body
    ).execute()

def main():
    youtube = get_authenticated_service()

    # Schedule the first live video
    video_id1 = schedule_live_video(youtube, ls_title_AM, ls_description_AM,
                                     ls_date_AM, ls_privacy_AM)
    print("First live video scheduled with ID:", video_id1)

    # Upload AMServiceThumbnail.png for the first live video
    thumbnail_filename1 = thumbnail_AM
    thumbnail_response1 = upload_thumbnail(youtube, video_id1, thumbnail_filename1)
    thumbnail_url1 = thumbnail_response1["items"][0]["default"]["url"]

    # Modify the category of the first live video
    modify_video_category(youtube, video_id1, ls_category_AM)
    print("Modified video category of AM service.")

    # Retrieve the ID of the live stream associated with the desired stream key
    desired_stream_id = get_live_stream_id_by_key(youtube, desired_stream_key)

    if desired_stream_id:
        print("Using temporary live stream key for PM service.")
    else:
        print("Temporary live stream key not found.")
        return

    # Schedule the second live video and associate it with the existing live stream
    video_id2 = schedule_live_video(youtube, ls_title_PM, ls_description_PM,
                                     ls_date_PM, ls_privacy_PM)
    print("Second live video scheduled with ID:", video_id2)

    # Associate the second live video with the existing live stream
    youtube.liveBroadcasts().bind(
        part="id,contentDetails",
        id=video_id2,
        streamId=desired_stream_id
    ).execute()

    # Upload PMServiceThumbnail.png for the second live video
    thumbnail_filename2 = thumbnail_PM
    thumbnail_response2 = upload_thumbnail(youtube, video_id2, thumbnail_filename2)
    thumbnail_url2 = thumbnail_response2["items"][0]["default"]["url"]

    # Modify the category of the second live video
    modify_video_category(youtube, video_id2, ls_category_PM)
    print("Modified video category of PM service.")

    # Write the video ids to a file to be used in modifystreamkey.py script
    with open(video_id_text, "w") as f:
        f.write(f"video_id1 = {video_id1}\n")
        f.write(f"video_id2 = {video_id2}\n")

if __name__ == "__main__":
    main()