YouTube：用官方 Python 客户端以断点续传方式上传视频

用官方 Python 客户端以断点续传方式上传视频：OAuth 只申请 youtube.upload 权限，默认设成私享，不通知订阅者，分块上传并显示进度，遇到 5xx 错误时按指数退避重试。测试中用模拟的 HTTP 序列覆盖了“上传到一半服务器返回 503”的情况。

先看清成品

本机用 pytest 跑了 3 个测试，全部通过：分块上传加一次 503 重试、403 直接报错、请求体和参数的默认值。没有 client_secret.json 时直接运行，报了 FileNotFoundError（见调试示例）。没有做过真实的 OAuth 授权和上传。

注意：YouTube Data API 本身免费，但有每日配额，各方法消耗多少以官方配额页为准，本文不列数字。只上传你有权发布的视频，遵守 YouTube 社区准则和 API 服务条款；不要共享频道账号或 OAuth 凭据，在 YouTube 无法访问的地区也不要用技术手段绕过。

做完之后你会得到什么？

需要准备哪些环境？

文件结构

29-youtube-C-src/
├── upload.py
├── test_upload.py
└── requirements.txt

第 1 步：官方文档里的几个要点

说明：网上很多旧文章说“未验证的 API 项目上传的视频会被锁定为私享”。当前版本的 videos.insert 页面写的是 “Videos uploaded from unverified API projects are not restricted to private viewing mode”。以你查看时的官方页面为准，不要依赖平台替你设成私享。

第 2 步：脚本

upload.py：

"""断点续传方式上传视频，默认设为私享（private）。
用法：python upload.py video.mp4 "标题" "描述"
第一次运行会打开浏览器进行 OAuth 授权，token 保存在 token.json。"""
import json, os, random, sys, time
from googleapiclient.discovery import build
from googleapiclient.errors import HttpError
from googleapiclient.http import MediaFileUpload

SCOPES = ["https://www.googleapis.com/auth/youtube.upload"]
RETRY_STATUS = {500, 502, 503, 504}
CHUNK = 8 * 1024 * 1024          # 必须是 256 KB 的整数倍

def get_service():
    from google.auth.transport.requests import Request
    from google.oauth2.credentials import Credentials
    from google_auth_oauthlib.flow import InstalledAppFlow
    creds = Credentials.from_authorized_user_file("token.json", SCOPES) if os.path.exists("token.json") else None
    if not creds or not creds.valid:
        if creds and creds.expired and creds.refresh_token:
            creds.refresh(Request())
        else:
            creds = InstalledAppFlow.from_client_secrets_file("client_secret.json", SCOPES).run_local_server(port=0)
        with open("token.json", "w") as f: f.write(creds.to_json())
    return build("youtube", "v3", credentials=creds)

def make_request(yt, path, title, description, privacy="private", chunksize=CHUNK):
    body = {"snippet": {"title": title, "description": description, "categoryId": "22"},
            "status": {"privacyStatus": privacy, "selfDeclaredMadeForKids": False}}
    media = MediaFileUpload(path, mimetype="video/*", chunksize=chunksize, resumable=True)
    return yt.videos().insert(part="snippet,status", body=body, media_body=media, notifySubscribers=False)

def run_upload(req, max_retries=5, sleep=time.sleep, log=print):
    response, retries = None, 0
    while response is None:
        try:
            status, response = req.next_chunk()
            if status: log(f"已上传 {int(status.progress() * 100)}%")
            retries = 0
        except HttpError as e:
            if e.resp.status not in RETRY_STATUS: raise
            retries += 1
            if retries > max_retries: raise
            wait = random.uniform(0, 2 ** retries)
            log(f"服务器返回 {e.resp.status}，{wait:.1f} 秒后重试（第 {retries} 次）")
            sleep(wait)
    log(f"完成：video id = {response['id']}，隐私状态 = {response['status']['privacyStatus']}")
    return response

if __name__ == "__main__":
    path, title, desc = sys.argv[1], sys.argv[2], sys.argv[3] if len(sys.argv) > 3 else ""
    run_upload(make_request(get_service(), path, title, desc))

要点：

• resumable=True 时，客户端库先建立上传会话，再按块发送。每块的大小要是 256 KB 的整数倍，这是断点续传协议文档的要求。客户端库不会替你检查：本机传入 1,000,000 字节也被接受了。
• 只对 500/502/503/504 重试，等待时间随机取 0 到 2^n 秒。403、400 这类错误重试也没有用，直接抛出。
• token.json 里有 refresh token，不要提交进代码库，也不要分享给别人。

第 3 步：测试

test_upload.py：

import json, pytest
from googleapiclient.discovery import build
from googleapiclient.errors import HttpError
from googleapiclient.http import HttpMockSequence
import upload

DONE = json.dumps({"id": "fakeVid123", "status": {"privacyStatus": "private"}})
KB256 = 256 * 1024

def yt_with(seq):
    return build("youtube", "v3", http=HttpMockSequence(seq), developerKey="x", static_discovery=True)

@pytest.fixture
def video(tmp_path):
    p = tmp_path / "v.mp4"; p.write_bytes(b"\0" * (KB256 * 2 + 1000)); return str(p)

def test_chunks_and_retry(video):
    seq = [({"status": "200", "location": "https://upload.example/session1"}, ""),
           ({"status": "308", "range": f"0-{KB256 - 1}"}, ""),
           ({"status": "503"}, "busy"),
           ({"status": "308", "range": f"0-{2 * KB256 - 1}"}, ""),
           ({"status": "200"}, DONE)]
    logs, waits = [], []
    req = upload.make_request(yt_with(seq), video, "t", "d", chunksize=KB256)
    r = upload.run_upload(req, sleep=waits.append, log=logs.append)
    assert r["id"] == "fakeVid123" and len(waits) == 1
    assert any("503" in l for l in logs) and logs[-1].endswith("private")

def test_non_retryable(video):
    seq = [({"status": "200", "location": "https://upload.example/s"}, ""),
           ({"status": "403"}, json.dumps({"error": {"errors": [{"reason": "forbidden"}], "code": 403, "message": "no"}}))]
    req = upload.make_request(yt_with(seq), video, "t", "d", chunksize=KB256)
    with pytest.raises(HttpError):
        upload.run_upload(req, sleep=lambda s: None, log=lambda s: None)

def test_body_defaults(video):
    req = upload.make_request(yt_with([]), video, "标题", "描述")
    body = json.loads(req.body)
    assert body["status"]["privacyStatus"] == "private"
    assert "notifySubscribers=false" in req.uri and "uploadType=resumable" in req.uri

pip install -r requirements.txt
python -m pytest -q     # 3 passed

第一个测试的模拟序列：建立会话（返回 location）→ 308（已收到第一块）→ 503 → 308（已收到第二块）→ 200（返回 video 资源）。断言只等待了 1 次，最后一条日志以 private 结尾。

第 4 步：真实上传（本机未执行）

python upload.py demo.mp4 "测试上传" "这是一条私享测试视频"

第一次运行会打开浏览器，用你自己的频道账号授权，之后会输出进度和 video id。去 YouTube Studio 确认视频是“私享”状态，再手动决定是否公开。

调试示例

直接运行，报 FileNotFoundError

FileNotFoundError: [Errno 2] No such file or directory: 'client_secret.json'

InstalledAppFlow 需要从 Google Cloud 控制台下载的 OAuth 客户端文件。去“API 和服务 → 凭据”创建一个“桌面应用”类型的 OAuth 客户端，下载 JSON 后放在脚本同目录，并改名为 client_secret.json。不要用 API key 代替，上传必须用 OAuth。

常见问题

模板规则

• 隐私状态必须显式设置，默认用 private。
• 只申请需要的最小 scope。
• 只对 5xx 重试，使用指数退避。
• 凭据文件和 token 不要提交进代码库，也不要分享。

项目资源

上传脚本的重点在默认值和重试：设成私享、不通知订阅者，上传后再由人决定是否公开。断点续传加上只对 5xx 的退避重试，大文件在网络不稳定时也能传完。