Service_Orchestrator/get_live_api_broadcasts.py

67 lines
1.8 KiB
Python

from __future__ import print_function
import datetime
import os.path
import argparse
from google.auth.transport.requests import Request
from google.oauth2.credentials import Credentials
from google_auth_oauthlib.flow import InstalledAppFlow
from googleapiclient.discovery import build
from googleapiclient.errors import HttpError
from pprint import pprint
SCOPES = ["https://www.googleapis.com/auth/youtube"]
def authorize():
creds = None
if os.path.exists('token.json'):
creds = Credentials.from_authorized_user_file('token.json', SCOPES)
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('credentials.json', SCOPES)
creds = flow.run_local_server(port=7384)
with open('token.json', 'w') as token:
token.write(creds.to_json())
return creds
def construct_youtube_service(creds):
return build('youtube', 'v3', credentials=creds)
def list_all_live_broadcasts(youtube_service, broadcastStatus):
page_token = None
list_of_entries = []
while True:
broadcast_list = youtube_service.liveBroadcasts().list(
part="snippet",
broadcastStatus=broadcastStatus,
pageToken=page_token
).execute()
list_of_entries.extend(broadcast_list["items"])
page_token = broadcast_list.get("nextPageToken")
if not page_token:
break
return list_of_entries
def main():
creds = authorize()
youtube_service = construct_youtube_service(creds)
pprint(list_all_live_broadcasts(youtube_service, "upcoming"))
# pprint([method_name for method_name in dir(youtube_service) if callable(getattr(youtube_service, method_name))])
if __name__ == '__main__':
main()