Video upload patterns for YouTube, TikTok, and Vimeo. Use when uploading videos to platforms, managing video metadata, scheduling video releases, or handling bulk video uploads.
Provides code for uploading videos to YouTube, TikTok, and Vimeo with support for metadata templates, bulk uploads, and validation. Use when you need to programmatically upload videos to social media platforms.
/plugin marketplace add mindmorass/reflex/plugin install reflex@mindmorass-reflexThis skill inherits all available tools. When active, it can use any tool Claude has access to.
Best practices for uploading videos to major platforms.
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()
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
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()"
)
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 # 10MB
# Initialize
init_response = self.init_upload(file_size, chunk_size)
upload_url = init_response['data']['upload_url']
publish_id = init_response['data']['publish_id']
# Upload chunks
with open(file_path, 'rb') as f:
chunk_index = 0
total_chunks = (file_size + chunk_size - 1) // chunk_size
while True:
chunk = f.read(chunk_size)
if not chunk:
break
self.upload_chunk(upload_url, chunk, chunk_index, total_chunks)
chunk_index += 1
print(f"Uploaded chunk {chunk_index}/{total_chunks}")
return {"publish_id": publish_id}
import vimeo
class VimeoUploader:
def __init__(self, token: str, key: str, secret: str):
self.client = vimeo.VimeoClient(
token=token,
key=key,
secret=secret
)
def upload_video(
self,
file_path: str,
name: str,
description: str = "",
privacy: str = "nobody" # anybody, nobody, password, disable
) -> dict:
"""Upload video to Vimeo."""
uri = self.client.upload(file_path, data={
'name': name,
'description': description,
'privacy': {
'view': privacy
}
})
# Get video details
response = self.client.get(uri)
return response.json()
def upload_with_progress(
self,
file_path: str,
name: str,
progress_callback=None
) -> dict:
"""Upload with progress tracking."""
def progress_handler(uploaded, total):
if progress_callback:
progress_callback(uploaded / total, uploaded, total)
uri = self.client.upload(
file_path,
data={'name': name},
progress=progress_handler
)
return self.client.get(uri).json()
def replace_video(self, video_id: str, file_path: str) -> dict:
"""Replace existing video file."""
uri = f"/videos/{video_id}"
return self.client.replace(uri, file_path)
def set_thumbnail(self, video_id: str, time_code: float) -> dict:
"""Set thumbnail from video frame."""
response = self.client.post(
f"/videos/{video_id}/pictures",
data={'time': time_code, 'active': True}
)
return response.json()
def add_to_folder(self, video_id: str, folder_id: str):
"""Add video to folder/project."""
self.client.put(f"/me/projects/{folder_id}/videos/{video_id}")
import asyncio
from dataclasses import dataclass
from enum import Enum
from typing import Callable, Optional
from pathlib import Path
class UploadPlatform(Enum):
YOUTUBE = "youtube"
TIKTOK = "tiktok"
VIMEO = "vimeo"
class UploadStatus(Enum):
PENDING = "pending"
UPLOADING = "uploading"
PROCESSING = "processing"
COMPLETE = "complete"
FAILED = "failed"
@dataclass
class VideoUploadJob:
file_path: Path
title: str
description: str
tags: list[str]
platforms: list[UploadPlatform]
thumbnail_path: Optional[Path] = None
scheduled_time: Optional[str] = None
status: UploadStatus = UploadStatus.PENDING
results: dict = None
def __post_init__(self):
self.results = {}
class BulkUploadManager:
def __init__(
self,
youtube_creds=None,
tiktok_token=None,
vimeo_client=None
):
self.youtube_creds = youtube_creds
self.tiktok_token = tiktok_token
self.vimeo_client = vimeo_client
self.jobs: list[VideoUploadJob] = []
def add_job(self, job: VideoUploadJob):
"""Add upload job to queue."""
self.jobs.append(job)
async def process_job(self, job: VideoUploadJob):
"""Process single upload job."""
job.status = UploadStatus.UPLOADING
for platform in job.platforms:
try:
if platform == UploadPlatform.YOUTUBE and self.youtube_creds:
result = upload_video_youtube(
self.youtube_creds,
str(job.file_path),
job.title,
job.description,
job.tags
)
job.results['youtube'] = result
elif platform == UploadPlatform.VIMEO and self.vimeo_client:
result = self.vimeo_client.upload_video(
str(job.file_path),
job.title,
job.description
)
job.results['vimeo'] = result
except Exception as e:
job.results[platform.value] = {"error": str(e)}
job.status = UploadStatus.COMPLETE
async def process_all(self, concurrent_limit: int = 3):
"""Process all jobs with concurrency limit."""
semaphore = asyncio.Semaphore(concurrent_limit)
async def process_with_limit(job):
async with semaphore:
await self.process_job(job)
await asyncio.gather(
*[process_with_limit(job) for job in self.jobs]
)
def get_status_report(self) -> dict:
"""Get status of all jobs."""
return {
"total": len(self.jobs),
"pending": sum(1 for j in self.jobs if j.status == UploadStatus.PENDING),
"uploading": sum(1 for j in self.jobs if j.status == UploadStatus.UPLOADING),
"complete": sum(1 for j in self.jobs if j.status == UploadStatus.COMPLETE),
"failed": sum(1 for j in self.jobs if j.status == UploadStatus.FAILED),
"jobs": [
{
"file": str(j.file_path),
"status": j.status.value,
"results": j.results
}
for j in self.jobs
]
}
from dataclasses import dataclass
from typing import Optional
from string import Template
@dataclass
class VideoMetadataTemplate:
title_template: str
description_template: str
tags: list[str]
category: str
def render(self, **kwargs) -> dict:
"""Render template with variables."""
return {
"title": Template(self.title_template).safe_substitute(**kwargs),
"description": Template(self.description_template).safe_substitute(**kwargs),
"tags": self.tags,
"category": self.category
}
# Example templates
GAMING_TEMPLATE = VideoMetadataTemplate(
title_template="$game_name - $episode_title | Episode $episode_num",
description_template="""
$episode_title
Welcome back to $game_name! In this episode, we $episode_summary.
Timestamps:
$timestamps
Subscribe for more $game_name content!
#$game_tag #gaming #letsplay
""".strip(),
tags=["gaming", "letsplay", "gameplay"],
category="20" # Gaming
)
TUTORIAL_TEMPLATE = VideoMetadataTemplate(
title_template="$topic Tutorial - $subtitle | $series_name",
description_template="""
Learn $topic in this comprehensive tutorial!
In this video:
$outline
Resources:
$resources
Don't forget to like and subscribe!
""".strip(),
tags=["tutorial", "howto", "learn"],
category="27" # Education
)
import subprocess
import os
from dataclasses import dataclass
@dataclass
class VideoValidation:
is_valid: bool
duration: float
resolution: str
codec: str
file_size: int
errors: list[str]
def validate_video(file_path: str, platform: str = "youtube") -> VideoValidation:
"""Validate video meets platform requirements."""
errors = []
# Get video info
probe = subprocess.run([
"ffprobe", "-v", "quiet",
"-print_format", "json",
"-show_format", "-show_streams",
file_path
], capture_output=True, text=True)
import json
info = json.loads(probe.stdout)
video_stream = next(s for s in info['streams'] if s['codec_type'] == 'video')
duration = float(info['format']['duration'])
file_size = int(info['format']['size'])
resolution = f"{video_stream['width']}x{video_stream['height']}"
codec = video_stream['codec_name']
# Platform-specific validation
if platform == "youtube":
if duration > 12 * 3600: # 12 hours
errors.append("Video exceeds 12 hour limit")
if file_size > 256 * 1024 * 1024 * 1024: # 256GB
errors.append("File exceeds 256GB limit")
elif platform == "tiktok":
if duration > 10 * 60: # 10 minutes
errors.append("Video exceeds 10 minute limit")
if file_size > 4 * 1024 * 1024 * 1024: # 4GB
errors.append("File exceeds 4GB limit")
elif platform == "vimeo":
if duration > 12 * 3600:
errors.append("Video exceeds 12 hour limit")
return VideoValidation(
is_valid=len(errors) == 0,
duration=duration,
resolution=resolution,
codec=codec,
file_size=file_size,
errors=errors
)
This skill should be used when the user asks to "create a slash command", "add a command", "write a custom command", "define command arguments", "use command frontmatter", "organize commands", "create command with file references", "interactive command", "use AskUserQuestion in command", or needs guidance on slash command structure, YAML frontmatter fields, dynamic arguments, bash execution in commands, user interaction patterns, or command development best practices for Claude Code.
This skill should be used when the user asks to "create an agent", "add an agent", "write a subagent", "agent frontmatter", "when to use description", "agent examples", "agent tools", "agent colors", "autonomous agent", or needs guidance on agent structure, system prompts, triggering conditions, or agent development best practices for Claude Code plugins.
This skill should be used when the user asks to "create a hook", "add a PreToolUse/PostToolUse/Stop hook", "validate tool use", "implement prompt-based hooks", "use ${CLAUDE_PLUGIN_ROOT}", "set up event-driven automation", "block dangerous commands", or mentions hook events (PreToolUse, PostToolUse, Stop, SubagentStop, SessionStart, SessionEnd, UserPromptSubmit, PreCompact, Notification). Provides comprehensive guidance for creating and implementing Claude Code plugin hooks with focus on advanced prompt-based hooks API.