Dateien über die Kommandozeile zu YouTube exportieren

(
  set -eu
  command -v python3.14 >/dev/null
  mkdir youtube-cli
  python3.14 -m venv youtube-cli/.venv
  youtube-cli/.venv/bin/python -m pip --isolated install \
    google-api-python-client==2.201.0 google-auth-oauthlib==1.5.0 \
    google-auth==2.62.1 httplib2==0.32.1 requests==2.34.2
)

{
  "title": "My command-line upload",
  "description": "A completed local video.\nUploaded privately for review.",
  "tags": ["demo", "command-line"],
  "madeForKids": false
}

import argparse
import contextlib
import json
import logging
import os
import re
import signal
import stat
import sys
from urllib.parse import urlsplit

import httplib2
import requests
from google_auth_oauthlib.flow import InstalledAppFlow
from googleapiclient.discovery import build
from googleapiclient.http import MediaIoBaseUpload
from requests.adapters import HTTPAdapter

UPLOAD_SCOPE = 'https://www.googleapis.com/auth/youtube.upload'
MAX_BYTES = 64 * 1024 * 1024


class UserError(Exception):
    pass


def stop_at_deadline(*_):
    raise UserError('The five-minute deadline expired')


class UploadTransport:
    def __init__(self, credentials):
        self.session = requests.Session()
        self.session.trust_env = False
        self.session.mount('https://', HTTPAdapter(max_retries=0))
        credentials.apply(self.session.headers)

    def request(self, uri, method='GET', body=None, headers=None, **_):
        target = urlsplit(uri)
        if (target.scheme != 'https' or target.hostname not in
                ('www.googleapis.com', 'youtube.googleapis.com') or
                target.username is not None or target.port not in (None, 443)):
            raise UserError('Unexpected upload destination')
        response = self.session.request(
            method, uri, data=body, headers=headers,
            timeout=(10, 30), allow_redirects=False)
        if 300 <= response.status_code < 400 and response.status_code != 308:
            raise UserError('Upload redirect refused')
        return (httplib2.Response(dict(response.headers,
                                      status=str(response.status_code))),
                response.content)

    def close(self):
        self.session.close()


class TimedFlow(InstalledAppFlow):
    def fetch_token(self, **kwargs):
        return super().fetch_token(timeout=(10, 30), **kwargs)


def authorize(config):
    installed = config.get('installed', {})
    if (installed.get('auth_uri') != 'https://accounts.google.com/o/oauth2/auth' or
            installed.get('token_uri') != 'https://oauth2.googleapis.com/token'):
        raise UserError('Use the downloaded Desktop client JSON')
    flow = TimedFlow.from_client_config(
        config, scopes=[UPLOAD_SCOPE], autogenerate_code_verifier=True)
    flow.oauth2session.trust_env = False
    flow.oauth2session.mount('https://', HTTPAdapter(max_retries=0))

    def refuse_token_redirect(url, headers, request_kwargs):
        return url, headers, dict(request_kwargs, allow_redirects=False)

    flow.oauth2session.register_compliance_hook('access_token_request', refuse_token_redirect)
    try:
        if flow.oauth2session.token:
            raise UserError('Expected a fresh authorization flow')
        with contextlib.redirect_stdout(sys.stderr):
            credentials = flow.run_local_server(
                host='127.0.0.1', port=0, open_browser=False, timeout_seconds=180,
                prompt='consent select_account', include_granted_scopes='false',
                access_type='online')
        with requests.Session() as check:
            check.trust_env = False
            response = check.get('https://oauth2.googleapis.com/tokeninfo',
                                    params={'access_token': credentials.token},
                                    timeout=(10, 30), allow_redirects=False)
            if response.status_code != 200:
                raise UserError('Could not check the new grant')
            info = response.json()
        if (info.get('aud') != installed['client_id'] or
                set(info.get('scope', '').split()) != {UPLOAD_SCOPE} or
                int(info.get('expires_in', 0)) < 600):
            raise UserError('The new grant must be upload-only and unexpired')
        return credentials
    finally:
        flow.oauth2session.close()


def read_metadata(filename):
    with open(filename, encoding='utf-8') as source:
        metadata = json.load(source)
    if not isinstance(metadata, dict) or set(metadata) != {
            'title', 'description', 'tags', 'madeForKids'}:
        raise UserError('Metadata needs title, description, tags, and madeForKids only')
    title, description, tags = (metadata[key] for key in ('title', 'description', 'tags'))
    if not isinstance(title, str) or not title.strip() or len(title) > 100:
        raise UserError('Title must contain 1–100 characters')
    if not isinstance(description, str) or not isinstance(tags, list) or not all(
            isinstance(tag, str) for tag in tags):
        raise UserError('Description must be text and tags must be a list of strings')
    if type(metadata['madeForKids']) is not bool:
        raise UserError('madeForKids must be a JSON boolean')
    return {'snippet': {'title': title, 'description': description, 'tags': tags},
            'status': {'privacyStatus': 'private',
                       'selfDeclaredMadeForKids': metadata['madeForKids']}}


def main():
    parser = argparse.ArgumentParser(description='Upload one private MP4; no automatic retry')
    parser.add_argument('--file', required=True)
    parser.add_argument('--metadata', required=True)
    parser.add_argument('--channel-id', required=True)
    parser.add_argument('--client-secrets', required=True,
                        help='Desktop JSON filename, or - for one private JSON line on stdin')
    args = parser.parse_args()
    if not re.fullmatch(r'UC[A-Za-z0-9_-]{22}', args.channel_id):
        raise UserError('Expected a UC channel ID')
    for filename in (args.file, args.metadata, args.client_secrets):
        os.fsencode(filename).decode('utf-8', errors='strict')
    body = read_metadata(args.metadata)
    with open(args.file, 'rb') as media:
        details = os.fstat(media.fileno())
        if (not stat.S_ISREG(details.st_mode) or not 0 < details.st_size <= MAX_BYTES or
                not args.file.lower().endswith('.mp4')):
            raise UserError('Use a completed, nonempty MP4 of at most 64 MiB')
        if args.client_secrets == '-':
            config = json.loads(sys.stdin.readline())
        else:
            with open(args.client_secrets, encoding='utf-8') as source:
                config = json.load(source)
        credentials = authorize(config)
        print(f'Confirm the channel selected in your browser: {args.channel_id}', file=sys.stderr)
        print('Type that channel ID to insert the private video:', file=sys.stderr, flush=True)
        if sys.stdin.readline().strip() != args.channel_id:
            raise UserError('Channel confirmation did not match; no insert attempted')
        transport = UploadTransport(credentials)
        try:
            youtube = build('youtube', 'v3', http=transport,
                            static_discovery=True, cache_discovery=False)
            request = youtube.videos().insert(
                part='snippet,status', body=body, notifySubscribers=False,
                media_body=MediaIoBaseUpload(media, mimetype='video/mp4',
                                            chunksize=8 * 1024 * 1024, resumable=True))
            print('Insert starting; inspect the channel before retrying any failure.',
                  file=sys.stderr, flush=True)
            result = None
            while result is None:
                _, result = request.next_chunk(num_retries=0)
            video_id = result.get('id')
            if not isinstance(video_id, str) or not re.fullmatch(r'[A-Za-z0-9_-]{11}', video_id):
                raise UserError('YouTube did not return a native video ID')
            print(f'Returned video ID: {video_id}', file=sys.stderr, flush=True)
            if (result.get('snippet', {}).get('channelId') != args.channel_id or
                    result.get('snippet', {}).get('title') != body['snippet']['title'] or
                    result.get('status', {}).get('privacyStatus') != 'private'):
                raise UserError('Returned metadata did not match; inspect the returned ID')
            print(json.dumps({'id': video_id, 'channelId': args.channel_id,
                              'status': 'accepted_processing_unchecked'}), flush=True)
        finally:
            transport.close()


if __name__ == '__main__':
    logging.disable(logging.CRITICAL)
    signal.signal(signal.SIGALRM, stop_at_deadline)
    signal.alarm(300)
    try:
        main()
    except (Exception, KeyboardInterrupt) as error:
        if isinstance(error, UserError):
            print(str(error), file=sys.stderr)
        print('Upload not confirmed. If insertion started, the video may exist. '
              'Inspect your channel in YouTube Studio before another upload. '
              'Check local files, metadata, authorization, and quota.', file=sys.stderr)
        sys.exit(1)
    finally:
        signal.alarm(0)

youtube-cli/.venv/bin/python youtube-cli/upload.py \
  --file './video.mp4' \
  --metadata youtube-cli/metadata.json \
  --client-secrets youtube-cli/client_secrets.json \
  --channel-id YOUR_CHANNEL_ID