modifystreamkey.py

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

#Set path and filename to read the video id of the PM video.
video_id_text = "/srv/dev-disk-by-uuid-49e4aa1a-279a-45ce-b23e-467c2b2a4162/AppData/jobson/specs/youtube-livestream.spec/video_id.txt"

#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'

#Main stream key to change the PM service to after the AM service ends
desired_stream_key = "stream_key_example"

# Define the scopes required for accessing the YouTube API
SCOPES = ["https://www.googleapis.com/auth/youtube.force-ssl"]

# Function to read video IDs from the file
def read_video_ids(filename):
    with open(video_id_text, "r") as f:
        video_ids = {}
        for line in f:
            key, value = line.strip().split("=")
            video_ids[key.strip()] = value.strip()
    return video_ids

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

def main():
    # Authenticate using OAuth 2.0
    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)
            creds = flow.run_local_server(port=0)
        with open(ls_token, 'w') as token:
            token.write(creds.to_json())

    # Build the YouTube Data API client
    youtube = build('youtube', 'v3', credentials=creds)

    # Read video IDs from the file
    video_ids = read_video_ids(video_id_text)

    # Retrieve video ID 2 from the dictionary
    video_id2 = video_ids.get("video_id2")

    print("Video ID of PM scheduled service:", video_id2)

    # 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:
        # Associate the second live video with the existing live stream
        youtube.liveBroadcasts().bind(
            part="id,contentDetails",
            id=video_id2,
            streamId=desired_stream_id
        ).execute()
        print("PM stream key updated successfully.")
    else:
        print("Livestream not found.")

if __name__ == "__main__":
    main()