video-upload-patterns

What this skill does

# Video Upload Patterns

Best practices for uploading videos to major platforms.

## YouTube Upload

### Basic Upload

```python
from googleapiclient.discovery import build
from googleapiclient.http import MediaFileUpload
from google.oauth2.credentials import Credentials

def upload_video_youtube(
    credentials: Credentials,
    file_path: str,
    title: str,
    description: str,
    tags: list[str],
    category_id: str = "22",  # People & Blogs
    privacy: str = "private"
) -> dict:
    """Upload video to YouTube."""
    youtube = build('youtube', 'v3', credentials=credentials)

    body = {
        "snippet": {
            "title": title,
            "description": description,
            "tags": tags,
            "categoryId": category_id
        },
        "status": {
            "privacyStatus": privacy,
            "selfDeclaredMadeForKids": False
        }
    }

    media = MediaFileUpload(
        file_path,
        mimetype='video/*',
        resumable=True,
        chunksize=1024*1024  # 1MB chunks
    )

    request = youtube.videos().insert(
        part="snippet,status",
        body=body,
        media_body=media
    )

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

    return response


def set_thumbnail(
    credentials: Credentials,
    video_id: str,
    thumbnail_path: str
):
    """Set custom thumbnail for video."""
    youtube = build('youtube', 'v3', credentials=credentials)

    media = MediaFileUpload(thumbnail_path, mimetype='image/jpeg')

    return youtube.thumbnails().set(
        videoId=video_id,
        media_body=media
    ).execute()


def schedule_video(
    credentials: Credentials,
    video_id: str,
    publish_at: str  # ISO 8601 format
):
    """Schedule video for future publication."""
    youtube = build('youtube', 'v3', credentials=credentials)

    return youtube.videos().update(
        part="status",
        body={
            "id": video_id,
            "status": {
                "privacyStatus": "private",
                "publishAt": publish_at
            }
        }
    ).execute()
```

### Resumable Upload with Progress

```python
import os
import time
from googleapiclient.errors import HttpError

class YouTubeUploader:
    MAX_RETRIES = 10
    RETRIABLE_STATUS_CODES = [500, 502, 503, 504]

    def __init__(self, credentials):
        self.youtube = build('youtube', 'v3', credentials=credentials)

    def upload_with_retry(
        self,
        file_path: str,
        metadata: dict,
        progress_callback=None
    ) -> dict:
        """Upload with automatic retry on failure."""
        file_size = os.path.getsize(file_path)

        media = MediaFileUpload(
            file_path,
            mimetype='video/*',
            resumable=True,
            chunksize=5 * 1024 * 1024  # 5MB chunks
        )

        request = self.youtube.videos().insert(
            part="snippet,status",
            body=metadata,
            media_body=media
        )

        response = None
        retry = 0

        while response is None:
            try:
                status, response = request.next_chunk()
                if status:
                    progress = status.progress()
                    bytes_uploaded = int(file_size * progress)
                    if progress_callback:
                        progress_callback(progress, bytes_uploaded, file_size)

            except HttpError as e:
                if e.resp.status in self.RETRIABLE_STATUS_CODES:
                    retry += 1
                    if retry > self.MAX_RETRIES:
                        raise
                    sleep_seconds = 2 ** retry
                    print(f"Retry {retry}, sleeping {sleep_seconds}s")
                    time.sleep(sleep_seconds)
                else:
                    raise

        return response
```

## TikTok Upload

### Content Posting API

```python
import requests
from dataclasses import dataclass
from typing import Optional

@dataclass
class TikTokVideo:
    video_url: str  # URL to video file
    title: str
    privacy_level: str = "SELF_ONLY"  # SELF_ONLY, MUTUAL_FOLLOW_FRIENDS, FOLLOWER_OF_CREATOR, PUBLIC_TO_EVERYONE
    disable_duet: bool = False
    disable_comment: bool = False
    disable_stitch: bool = False


class TikTokUploader:
    def __init__(self, access_token: str):
        self.access_token = access_token
        self.base_url = "https://open.tiktokapis.com/v2"
        self.headers = {
            "Authorization": f"Bearer {access_token}",
            "Content-Type": "application/json"
        }

    def init_video_upload(self, video: TikTokVideo) -> dict:
        """Initialize video upload and get upload URL."""
        response = requests.post(
            f"{self.base_url}/post/publish/video/init/",
            headers=self.headers,
            json={
                "post_info": {
                    "title": video.title,
                    "privacy_level": video.privacy_level,
                    "disable_duet": video.disable_duet,
                    "disable_comment": video.disable_comment,
                    "disable_stitch": video.disable_stitch
                },
                "source_info": {
                    "source": "PULL_FROM_URL",
                    "video_url": video.video_url
                }
            }
        )
        return response.json()

    def check_publish_status(self, publish_id: str) -> dict:
        """Check status of video publishing."""
        response = requests.post(
            f"{self.base_url}/post/publish/status/fetch/",
            headers=self.headers,
            json={"publish_id": publish_id}
        )
        return response.json()

    def upload_from_file(self, file_path: str, video: TikTokVideo) -> dict:
        """Upload video from local file (requires file hosting)."""
        # TikTok requires video URL, so you need to host the file first
        # This is a placeholder for the workflow
        raise NotImplementedError(
            "TikTok requires video URL. Host file and use init_video_upload()"
        )
```

### Direct Upload (Chunk-based)

```python
class TikTokDirectUploader:
    """Direct upload using chunk-based approach."""

    def __init__(self, access_token: str):
        self.access_token = access_token
        self.base_url = "https://open.tiktokapis.com/v2"

    def init_upload(
        self,
        file_size: int,
        chunk_size: int = 10 * 1024 * 1024  # 10MB
    ) -> dict:
        """Initialize direct upload."""
        total_chunks = (file_size + chunk_size - 1) // chunk_size

        response = requests.post(
            f"{self.base_url}/post/publish/inbox/video/init/",
            headers={
                "Authorization": f"Bearer {self.access_token}",
                "Content-Type": "application/json"
            },
            json={
                "source_info": {
                    "source": "FILE_UPLOAD",
                    "video_size": file_size,
                    "chunk_size": chunk_size,
                    "total_chunk_count": total_chunks
                }
            }
        )
        return response.json()

    def upload_chunk(
        self,
        upload_url: str,
        chunk_data: bytes,
        chunk_index: int,
        total_chunks: int
    ):
        """Upload a single chunk."""
        response = requests.put(
            upload_url,
            headers={
                "Content-Type": "video/mp4",
                "Content-Range": f"bytes {chunk_index * len(chunk_data)}-{(chunk_index + 1) * len(chunk_data) - 1}/{total_chunks * len(chunk_data)}"
            },
            data=chunk_data
        )
        return response

    def upload_file(self, file_path: str) -> dict:
        """Upload complete file in chunks."""
        import os

        file_size = os.path.getsize(file_path)
        chunk_size = 10 * 1024 * 1024