使用YouTube Live Streaming API创建Python直播无视频源问题求助

你的脚本提示直播启动成功，但YouTube端显示“等待广播”，核心原因是未将指定的已上传视频与直播流/广播关联，同时存在API调用逻辑的几个错误，以下是具体排查和修复步骤：

核心排查点与修复措施 #

• 未关联指定视频到直播流
  你的代码中get_video_id()获取了视频ID，但create_live_broadcast函数接收该参数后完全未使用，且创建直播流(liveStreams.insert)时未绑定视频ID。要将已上传视频作为直播源，必须在创建liveStream时，在contentDetails中添加videoId字段指定目标视频。

• 直播时间与状态切换冲突
  你设置了scheduledStartTime为当前时间+1分钟，但立即调用update接口将状态改为live，这会触发YouTube的时间校验逻辑，导致直播无法正常启动。建议移除scheduledStartTime设置，或者等待到指定时间后再切换状态。

• 直播绑定接口参数错误
  调用liveBroadcasts.bind时，part参数需要包含contentDetails，否则绑定关系不会被正确记录。

• 状态切换方式不符合API规范
  YouTube官方推荐使用liveBroadcasts.transition接口切换直播状态，而非直接update状态字段，这样能确保API执行完整的状态流转逻辑。

修正后的完整代码 #

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

# OAuth认证
def authenticate():
    SCOPES = [
        'https://www.googleapis.com/auth/youtube.force-ssl',
        'https://www.googleapis.com/auth/youtube',
        'https://www.googleapis.com/auth/youtubepartner'
    ]

    credentials_file = 'token.pickle'
    credentials = None

    if os.path.exists(credentials_file):
        with open(credentials_file, 'rb') as token:
            credentials = pickle.load(token)
    
    if not credentials or not credentials.valid:
        if credentials and credentials.expired and credentials.refresh_token:
            credentials.refresh(Request())
        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

# 创建YouTube API客户端
def create_youtube_client(credentials):
    return build('youtube', 'v3', credentials=credentials)

# 获取要作为直播源的视频ID
def get_video_id():
    # 替换为你频道内的已上传视频ID
    return "YOUR_VIDEO_ID"

# 创建绑定了指定视频的直播流
def create_live_stream(youtube, video_id):
    request = youtube.liveStreams().insert(
        part="snippet,cdn,contentDetails,status",
        body={
            "snippet": {
                "title": "预录直播流",
                "description": "基于已上传视频的直播流"
            },
            "cdn": {
                "frameRate": "variable",
                "ingestionType": "rtmp",
                "resolution": "variable"
            },
            "contentDetails": {
                "enableAutoStart": True,
                "isReusable": True,
                "videoId": video_id  # 关键：绑定要播放的已上传视频
            },
            "status": {
                "privacyStatus": "private"
            }
        }
    )
    response = request.execute()
    return response["id"]

# 创建直播广播并绑定直播流
def create_live_broadcast(youtube, stream_id):
    request = youtube.liveBroadcasts().insert(
        part="snippet,status,contentDetails",
        body={
            "snippet": {
                "title": "我的预录直播",
                "description": "通过API创建的预录直播"
            },
            "status": {
                "privacyStatus": "private"
            },
            "contentDetails": {
                "enableAutoStart": True,
                "enableAutoStop": True
            }
        }
    )
    response = request.execute()
    broadcast_id = response["id"]

    # 绑定直播流到广播，part必须包含contentDetails
    bind_request = youtube.liveBroadcasts().bind(
        part="id,snippet,contentDetails",
        id=broadcast_id,
        streamId=stream_id
    )
    bind_request.execute()

    return broadcast_id

# 切换直播状态为直播中
def start_live_broadcast(youtube, broadcast_id):
    request = youtube.liveBroadcasts().transition(
        part="status",
        id=broadcast_id,
        broadcastStatus="live"
    )
    response = request.execute()
    return response["id"]

def main():
    credentials = authenticate()
    youtube = create_youtube_client(credentials)
    
    video_id = get_video_id()
    print(f"使用视频ID: {video_id} 创建直播流")

    stream_id = create_live_stream(youtube, video_id)
    print(f"创建直播流成功，ID: {stream_id}")

    broadcast_id = create_live_broadcast(youtube, stream_id)
    print(f"创建直播广播成功，ID: {broadcast_id}")

    # 等待流初始化（建议延长到30秒，确保视频加载完成）
    time.sleep(30)

    start_live_broadcast(youtube, broadcast_id)
    print("直播已成功启动，可前往YouTube后台查看播放状态")

if __name__ == "__main__":
    main()

额外注意事项 #

• 确保使用的视频ID属于当前认证的频道，且视频状态为公开或未列出（私有视频无法作为直播源）
• 首次运行需要确保client_secrets.json文件路径正确，且已在Google Cloud Console中启用YouTube Data API v3
• 如果仍出现问题，可通过API返回的响应信息排查（建议在每个API调用后打印response，查看是否有错误提示）

内容的提问来源于stack exchange，提问作者lommer