使用Python调用YouTube Data API v3上传视频时遇ssl.SSLWantWriteError

问题背景 #

我用Python实现了Twitch VOD下载+YouTube自动上传的功能，下载环节正常，但上传时触发ssl.SSLWantWriteError错误。网络稳定、手动上传正常，无法直连路由器，未使用代理/防火墙。

实现代码 #

import os
import subprocess
import requests
from google.oauth2 import service_account
from googleapiclient.discovery import build

def download_vod(url, file_path):
    command = f'streamlink --hls-live-edge 99999 -o \"{file_path}\" \"{url}\" best'
    subprocess.call(command, shell=True)

def upload_to_youtube(video_file, title, description):
    credentials = service_account.Credentials.from_service_account_file('client_secret.json', scopes=['https://www.googleapis.com/auth/youtube.upload'])
    youtube = build('youtube', 'v3', credentials=credentials)

    request = youtube.videos().insert(
        part='snippet,status',
        body={
            'snippet': {
                'title': title,
                'description': description,
            },
            'status': {
                'privacyStatus': 'private',  # Change privacy status as desired
            }
        },
        media_body=video_file
    )
    response = request.execute()

    video_id = response['id']
    print(f"Video uploaded successfully. Video ID: {video_id}")

def main():
    while True:
        url = input("Enter the Twitch VOD URL (or 'q' to quit): ")
        if url == 'q':
            break

        file_name = input("Enter the desired file name (excluding extension): ")
        file_path = file_name  # You can specify a directory path here if desired

        download_vod(url, file_path + ".mp4")
        print("Download complete!")

        upload = input("Upload to YouTube? (y/n): ")
        if upload.lower() == 'y':
            title = input("Enter video title: ")
            description = input("Enter video description: ")
            upload_to_youtube(file_path + ".mp4", title, description)

        print()

if __name__ == '__main__':
    main()

报错信息 #

Traceback (most recent call last):
  File "c:\Users\myrai\OneDrive\Desktop\TTV-YT\main.py", line 54, in <module>
    main()
  File "c:\Users\myrai\OneDrive\Desktop\TTV-YT\main.py", line 49, in main
    upload_to_youtube(file_path + ".mp4", title, description)
  File "c:\Users\myrai\OneDrive\Desktop\TTV-YT\main.py", line 28, in upload_to_youtube
    response = request.execute()
  File "C:\Users\myrai\AppData\Local\Programs\Python\Python310\lib\site-packages\googleapiclient\_helpers.py", line 130, in positional_wrapper
    return wrapped(*args, **kwargs)
  File "C:\Users\myrai\AppData\Local\Programs\Python\Python310\lib\site-packages\googleapiclient\http.py", line 923, in execute
    resp, content = _retry_request(
  File "C:\Users\myrai\AppData\Local\Programs\Python\Python310\lib\site-packages\googleapiclient\http.py", line 222, in _retry_request
    raise exception
  File "C:\Users\myrai\AppData\Local\Programs\Python\Python310\lib\site-packages\googleapiclient\http.py", line 191, in _retry_request
    resp, content = http.request(uri, method, *args, **kwargs)
  File "C:\Users\myrai\AppData\Local\Programs\Python\Python310\lib\site-packages\google_auth_httplib2.py", line 218, in request
    response, content = self.http.request(
  File "C:\Users\myrai\AppData\Local\Programs\Python\Python310\lib\site-packages\httplib2\__init__.py", line 1724, in request
    (response, content) = self._request(
  File "C:\Users\myrai\AppData\Local\Programs\Python\Python310\lib\site-packages\httplib2\__init__.py", line 1444, in _request
    (response, content) = self._conn_request(conn, request_uri, method, body, headers)
  File "C:\Users\myrai\AppData\Local\Programs\Python\Python310\lib\site-packages\httplib2\__init__.py", line 1367, in _conn_request
    conn.request(method, request_uri, body, headers)
  File "C:\Users\myrai\AppData\Local\Programs\Python\Python310\lib\http\client.py", line 1282, in request
    self._send_request(method, url, body, headers, encode_chunked)
  File "C:\Users\myrai\AppData\Local\Programs\Python\Python310\lib\http\client.py", line 1328, in _send_request
    self.endheaders(body, encode_chunked=encode_chunked)
  File "C:\Users\myrai\AppData\Local\Programs\Python\Python310\lib\http\client.py", line 1277, in endheaders
    self._send_output(message_body, encode_chunked=encode_chunked)
  File "C:\Users\myrai\AppData\Local\Programs\Python\Python310\lib\http\client.py", line 1076, in _send_output
    self.send(chunk)
  File "C:\Users\myrai\AppData\Local\Programs\Python\Python310\lib\http\client.py", line 998, in send
    self.sock.sendall(data)
  File "C:\Users\myrai\AppData\Local\Programs\Python\Python310\lib\ssl.py", line 1237, in sendall
    v = self.send(byte_view[count:])
  File "C:\Users\myrai\AppData\Local\Programs\Python\Python310\lib\ssl.py", line 1206, in send
    return self._sslobj.write(data)
ssl.SSLWantWriteError: The operation did not complete (write) (_ssl.c:2396)

解决方案 #

1. 启用可恢复上传+重试逻辑 #

这个错误属于临时SSL写入阻塞，用MediaFileUpload实现分块上传，并添加重试机制，能自动处理这类临时错误：

import ssl
import time
from google.oauth2 import service_account
from googleapiclient.discovery import build
from googleapiclient.http import MediaFileUpload

def upload_to_youtube(video_file, title, description):
    credentials = service_account.Credentials.from_service_account_file('client_secret.json', scopes=['https://www.googleapis.com/auth/youtube.upload'])
    youtube = build('youtube', 'v3', credentials=credentials)

    # 启用10MB分块的可恢复上传
    media = MediaFileUpload(video_file, chunksize=10*1024*1024, resumable=True)
    request = youtube.videos().insert(
        part='snippet,status',
        body={
            'snippet': {
                'title': title,
                'description': description,
            },
            'status': {
                'privacyStatus': 'private',
            }
        },
        media_body=media
    )

    response = None
    retry_limit = 3
    while response is None:
        try:
            status, response = request.next_chunk()
            if status:
                print(f"上传进度: {int(status.progress()*100)}%")
        except ssl.SSLWantWriteError:
            retry_limit -= 1
            if retry_limit <= 0:
                raise
            print(f"SSL写入失败，重试中...剩余次数: {retry_limit}")
            time.sleep(2)

    print(f"上传成功，视频ID: {response['id']}")

2. 更换HTTP客户端为requests #

httplib2在部分环境下存在SSL兼容性问题，改用requests作为HTTP客户端：
首先安装依赖：

pip install requests-oauthlib

然后修改上传函数：

from google.oauth2 import service_account
from googleapiclient.discovery import build
from googleapiclient.http import MediaFileUpload
import requests

def upload_to_youtube(video_file, title, description):
    credentials = service_account.Credentials.from_service_account_file(
        'client_secret.json',
        scopes=['https://www.googleapis.com/auth/youtube.upload']
    )

    # 使用requests会话替代httplib2
    http = requests.Session()
    http.auth = credentials

    youtube = build('youtube', 'v3', http=http)

    media = MediaFileUpload(video_file, chunksize=10*1024*1024, resumable=True)
    request = youtube.videos().insert(
        part='snippet,status',
        body={
            'snippet': {
                'title': title,
                'description': description,
            },
            'status': {
                'privacyStatus': 'private',
            }
        },
        media_body=media
    )

    response = None
    while response is None:
        status, response = request.next_chunk()
        if status:
            print(f"上传进度: {int(status.progress()*100)}%")

    print(f"上传成功，视频ID: {response['id']}")

3. 强制使用TLS 1.2协议 #

指定SSL协议版本，避免协议兼容问题：

import ssl
from httplib2 import Http
from google.oauth2 import service_account
from googleapiclient.discovery import build
from googleapiclient.http import MediaFileUpload

def upload_to_youtube(video_file, title, description):
    credentials = service_account.Credentials.from_service_account_file('client_secret.json', scopes=['https://www.googleapis.com/auth/youtube.upload'])

    # 强制使用TLS 1.2
    ssl_context = ssl.SSLContext(ssl.PROTOCOL_TLSv1_2)
    http = Http(ssl_context=ssl_context)
    http = credentials.authorize(http)

    youtube = build('youtube', 'v3', http=http)

    media = MediaFileUpload(video_file, chunksize=10*1024*1024, resumable=True)
    request = youtube.videos().insert(
        part='snippet,status',
        body={
            'snippet': {
                'title': title,
                'description': description,
            },
            'status': {
                'privacyStatus': 'private',
            }
        },
        media_body=media
    )

    response = None
    while response is None:
        status, response = request.next_chunk()
        if status:
            print(f"上传进度: {int(status.progress()*100)}%")

    print(f"上传成功，视频ID: {response['id']}")

4. 验证服务账号权限 #

确保服务账号已被添加到YouTube频道的权限列表：

1. 打开YouTube Studio，进入设置 → 权限
2. 点击邀请，输入client_secret.json里client_email字段的邮箱
3. 授予所有者或管理者权限，等待邀请生效

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