File size: 4,893 Bytes
072ddc8 0b5d555 072ddc8 0b5d555 072ddc8 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 | from typing import Any, Optional
from smolagents.tools import Tool
import os
import upload_post
class UploadPostVideoTool(Tool):
name = "upload_post_video"
description = "Publish (or schedule) a video on social networks through Upload-Post: TikTok, Instagram, YouTube, LinkedIn, Facebook, X, Threads, Pinterest, Bluesky and more. The video must be a public http(s) URL. The upload is processed asynchronously: the returned dict contains a request_id (or a job_id when scheduled) to pass to upload_post_status."
inputs = {'video': {'type': 'string', 'description': 'Public http(s) URL of the video. Local file paths only work if the tool was created with allow_local_files=True.'}, 'platforms': {'type': 'array', 'description': "Platforms to publish to, e.g. ['tiktok', 'instagram', 'youtube']. Allowed: tiktok, instagram, youtube, linkedin, facebook, x, threads, pinterest, bluesky, discord, telegram, google_business, mastodon, wordpress."}, 'title': {'type': 'string', 'description': 'Title / caption of the video. Required for YouTube.', 'nullable': True}, 'description': {'type': 'string', 'description': 'Longer text used for LinkedIn commentary, Facebook and YouTube descriptions and Pinterest notes.', 'nullable': True}, 'profile': {'type': 'string', 'description': 'Upload-Post profile (username) to publish from. Omit to use the default profile. Use upload_post_profiles to list profiles.', 'nullable': True}, 'scheduled_date': {'type': 'string', 'description': "ISO-8601 date/time to schedule the post instead of publishing now, e.g. '2026-12-31T18:00:00Z'. Must be in the future (max 365 days).", 'nullable': True}, 'timezone': {'type': 'string', 'description': "IANA timezone (e.g. 'Europe/Madrid') for scheduled_date. Defaults to UTC.", 'nullable': True}, 'add_to_queue': {'type': 'boolean', 'description': "Put the post in the profile's next free queue slot. Not combinable with scheduled_date.", 'nullable': True}, 'first_comment': {'type': 'string', 'description': 'First comment posted right after publishing (Instagram, Facebook, Threads, Bluesky, X, YouTube, LinkedIn, TikTok).', 'nullable': True}}
output_type = "object"
def __init__(self, api_key=None, profile=None, allow_local_files=False, **kwargs):
super().__init__()
import os
self.api_key = api_key or os.environ.get("UPLOAD_POST_API_KEY")
self.profile = profile or os.environ.get("UPLOAD_POST_PROFILE")
self.allow_local_files = allow_local_files
self.client = None
def setup(self):
from upload_post import UploadPostClient
if not self.api_key:
raise ValueError(
"Missing Upload-Post API key: pass api_key=... or set UPLOAD_POST_API_KEY."
)
self.client = UploadPostClient(self.api_key)
self.is_initialized = True
def forward(
self,
video: str,
platforms: list,
title: Optional[str] = None,
description: Optional[str] = None,
profile: Optional[str] = None,
scheduled_date: Optional[str] = None,
timezone: Optional[str] = None,
add_to_queue: Optional[bool] = None,
first_comment: Optional[str] = None,
) -> Any:
import os
if not self.is_initialized:
self.setup()
user = profile or self.profile
if not user:
raise ValueError(
"No Upload-Post profile: pass profile=... or set UPLOAD_POST_PROFILE. "
"Call upload_post_profiles to list the available profiles."
)
if isinstance(platforms, str):
platforms = [p.strip() for p in platforms.split(",") if p.strip()]
if not platforms:
raise ValueError("At least one platform is required.")
if scheduled_date and add_to_queue:
raise ValueError("scheduled_date and add_to_queue cannot be used together.")
if not video.lower().startswith(("http://", "https://")):
if not self.allow_local_files:
raise ValueError(
"Local file paths are disabled (allow_local_files=False). "
"Pass a public http(s) URL instead."
)
video = os.path.expanduser(video)
if not os.path.isfile(video):
raise ValueError(f"File not found: {video}")
options = {
"description": description,
"scheduled_date": scheduled_date,
"timezone": timezone,
"add_to_queue": add_to_queue,
"first_comment": first_comment,
}
options = {k: v for k, v in options.items() if v is not None and v != ""}
return self.client.upload_video(
video,
title=title,
user=user,
platforms=list(platforms),
async_upload=True,
**options,
)
|