使用 Youtube Live Streaming Api [Python] 进行直播视频

Question

我正在尝试创建一个 Python 脚本，该脚本将使用 YouTube Live Streaming API 和 OAuth 创建直播。在代码的第一部分中，通过 OAuth 执行身份验证，效果很好。

在下一部分中，我从 YouTube 上的公共视频创建直播并启动直播。然后我将流绑定到直播并开始直播。

运行脚本后，打印“直播已成功开始”

但是，当我查看 YouTube 时，直播已创建，但其中没有播放任何内容。仍然显示正在等待广播的消息。

我完全迷失了，因为我是初学者。你能告诉我如何解决这个问题吗？

import os
import pickle
import time
from google_auth_oauthlib.flow import InstalledAppFlow
from google.auth.transport.requests import Request
from googleapiclient.discovery import build
from datetime import datetime, timedelta

# Function for OAuth authentication
def authenticate():
    SCOPES = ['https://www.googleapis.com/auth/youtube.force-ssl', 
              'https://www.googleapis.com/auth/youtube', 
              'https://www.googleapis.com/auth/youtube.readonly', 
              'https://www.googleapis.com/auth/youtubepartner', 
              'https://www.googleapis.com/auth/youtube.upload']

    credentials_file = 'token.pickle'

    if os.path.exists(credentials_file):
        with open(credentials_file, 'rb') as token:
            credentials = pickle.load(token)
    else:
        flow = InstalledAppFlow.from_client_secrets_file('client_secrets.json', scopes=SCOPES)
        credentials = flow.run_local_server(port=8080)

        with open(credentials_file, 'wb') as token:
            pickle.dump(credentials, token)

    return credentials

# Create an authorized YouTube API client
def create_youtube_client(credentials):
    return build('youtube', 'v3', credentials=credentials)

# Get the video ID from your channel
def get_video_id():
    # Here you can implement code to get the video ID, for example using the YouTube Data API
    return "my video ID from https"

# Create a live stream
def create_live_stream(youtube):
    request = youtube.liveStreams().insert(
        part="snippet,cdn,contentDetails,status",
        body={
            "snippet": {
                "title": "Your new video stream's name",
                "description": "A description of your video stream. This field is optional."
            },
            "cdn": {
                "frameRate": "variable",
                "ingestionType": "rtmp",
                "resolution": "variable"
            },
            "contentDetails": {
                "enableAutoStart": True,
                "isReusable": True
            }
        }
    )
    response = request.execute()
    return response["id"]

# Create a live broadcast
def create_live_broadcast(youtube, video_id, stream_id):
    request = youtube.liveBroadcasts().insert(
        part="snippet,status,contentDetails",
        body={
            "snippet": {
                "title": "Your live broadcast title",
                "description": "Your live broadcast description",
                "scheduledStartTime": (datetime.now() + timedelta(minutes=1)).isoformat(),  # Schedule the broadcast for 1 minutes later
                "thumbnails": {
                    "default": {
                        "url": "https://example.com/live-thumbnail.jpg"
                    }
                }
            },
            "status": {
                "privacyStatus": "private"
            },
            "contentDetails": {
                "enableAutoStart": True
            }
        }
    )
    response = request.execute()
    broadcast_id = response["id"]

    # Bind the stream to the live broadcast
    bind_request = youtube.liveBroadcasts().bind(
        part="id,snippet",
        id=broadcast_id,
        streamId=stream_id
    )
    bind_response = bind_request.execute()

    return broadcast_id

# Start the live broadcast
def start_live_broadcast(youtube, broadcast_id):
    request = youtube.liveBroadcasts().update(
        part="id,status",
        body={
            "id": broadcast_id,
            "status": {
                "lifeCycleStatus": "live"
            }
        }
    )
    response = request.execute()
    return response["id"]

# Main function
def main():
    # Authenticate using OAuth
    credentials = authenticate()

    # Create an authorized client for the YouTube API
    youtube = create_youtube_client(credentials)

    # Get the video ID
    video_id = get_video_id()

    # Create a live stream
    stream_id = create_live_stream(youtube)

    # Create a live broadcast
    broadcast_id = create_live_broadcast(youtube, video_id, stream_id)

    # Wait for 10 seconds to initialize the broadcast
    time.sleep(10)

    # Start the live broadcast
    start_live_broadcast(youtube, broadcast_id)

    print("Live broadcast has been successfully started.")

if __name__ == "__main__":
    main()

我尝试了频道中的不同视频 ID，尝试使用 youtube api 中的“videos.insert”上传新视频，但似乎没有任何效果。一定有一些我的业余眼睛无法发现的错误，但开发人员会立即发现。

直播好像收不到直播的视频源，但不知道为什么。

Answer 1

该问题源于 start_live_broadcast 函数中用于将广播状态更新为“live”的参数不正确。正确的参数是status.lifeCycleStatus而不是status。

def start_live_broadcast(youtube, broadcast_id):
request = youtube.liveBroadcasts().transition(
    broadcastStatus="live",
    id=broadcast_id,
    part="id,status"
)
response = request.execute()
return response["id"]