| 1 | """ |
| 2 | Upload-Post API integration for cross-posting videos to TikTok, Instagram and YouTube Shorts. |
| 3 | |
| 4 | Docs: https://docs.upload-post.com |
| 5 | """ |
| 6 | import os |
| 7 | from typing import Optional |
| 8 | |
| 9 | import requests |
| 10 | from loguru import logger |
| 11 | from app.config import config |
| 12 | |
| 13 | |
| 14 | class UploadPostService: |
| 15 | API_BASE = "https://api.upload-post.com" |
| 16 | |
| 17 | def __init__(self): |
| 18 | self.api_key = config.app.get("upload_post_api_key", "") |
| 19 | self.username = config.app.get("upload_post_username", "") |
| 20 | self.enabled = config.app.get("upload_post_enabled", False) |
| 21 | self.platforms = config.app.get("upload_post_platforms", ["tiktok", "instagram"]) |
| 22 | self.auto_upload = config.app.get("upload_post_auto_upload", False) |
| 23 | self.youtube_privacy_status = config.app.get("upload_post_youtube_privacy_status", "public") |
| 24 | |
| 25 | def is_configured(self) -> bool: |
| 26 | return bool(self.api_key and self.username and self.enabled) |
| 27 | |
| 28 | def upload_video( |
| 29 | self, |
| 30 | video_path: str, |
| 31 | title: str, |
| 32 | platforms: Optional[list] = None, |
| 33 | privacy_level: str = "PUBLIC_TO_EVERYONE", |
| 34 | youtube_extra: Optional[dict] = None, |
| 35 | ) -> dict: |
| 36 | if not self.is_configured(): |
| 37 | logger.warning("Upload-Post is not configured. Skipping cross-post.") |
| 38 | return {"success": False, "error": "Upload-Post not configured"} |
| 39 | |
| 40 | if platforms is None: |
| 41 | platforms = self.platforms |
| 42 | |
| 43 | if not os.path.exists(video_path): |
| 44 | logger.error(f"Video file not found: {video_path}") |
| 45 | return {"success": False, "error": f"Video file not found: {video_path}"} |
| 46 | |
| 47 | logger.info(f"Cross-posting video to {', '.join(platforms)} via Upload-Post...") |
| 48 | |
| 49 | try: |
| 50 | with open(video_path, 'rb') as video_file: |
| 51 | files = {'video': video_file} |
| 52 | |
| 53 | data = [ |
| 54 | ('user', self.username), |
| 55 | ('title', title[:2200]), |
| 56 | ('privacy_level', privacy_level), |
| 57 | ] |
| 58 | |
| 59 | for platform in platforms: |
| 60 | data.append(('platform[]', platform)) |
| 61 | |
| 62 | if youtube_extra and any(p.startswith("youtube") for p in platforms): |
| 63 | if "youtube_title" in youtube_extra: |
| 64 | data.append(('youtube_title', youtube_extra["youtube_title"][:100])) |
| 65 | if "youtube_description" in youtube_extra: |
| 66 | data.append(('youtube_description', youtube_extra["youtube_description"])) |
| 67 | for tag in youtube_extra.get("tags", []): |
| 68 | data.append(('tags[]', tag)) |
| 69 | data.append(('privacyStatus', youtube_extra.get("privacyStatus", "public"))) |
| 70 | data.append(('containsSyntheticMedia', "true")) |
| 71 | |
| 72 | headers = {'Authorization': f'Apikey {self.api_key}'} |
| 73 | |
| 74 | response = requests.post( |
| 75 | f"{self.API_BASE}/api/upload", |
| 76 | headers=headers, |
| 77 | data=data, |
| 78 | files=files, |
| 79 | timeout=300, |
| 80 | ) |
| 81 | |
| 82 | response.raise_for_status() |
| 83 | result = response.json() |
| 84 | |
| 85 | if result.get('success'): |
| 86 | logger.info(f"✅ Video cross-posted successfully! Request ID: {result.get('request_id')}") |
| 87 | else: |
| 88 | logger.warning(f"Cross-post failed: {result.get('message', 'Unknown error')}") |
| 89 | |
| 90 | return result |
| 91 | |
| 92 | except requests.exceptions.RequestException as e: |
| 93 | logger.error(f"Failed to cross-post video: {str(e)}") |
| 94 | return {"success": False, "error": str(e)} |
| 95 | |
| 96 | def check_status(self, request_id: str) -> dict: |
| 97 | """ |
| 98 | Check the status of an upload request. |
| 99 | |
| 100 | Args: |
| 101 | request_id (str): The request ID from upload |
| 102 | |
| 103 | Returns: |
| 104 | dict: Status information |
| 105 | """ |
| 106 | try: |
| 107 | headers = { |
| 108 | 'Authorization': f'Apikey {self.api_key}' |
| 109 | } |
| 110 | |
| 111 | response = requests.get( |
| 112 | f"{self.API_BASE}/api/uploadposts/status", |
| 113 | params={'request_id': request_id}, |
| 114 | headers=headers, |
| 115 | timeout=30 |
| 116 | ) |
| 117 | |
| 118 | response.raise_for_status() |
| 119 | return response.json() |
| 120 | |
| 121 | except requests.exceptions.RequestException as e: |
| 122 | logger.error(f"Failed to check status: {str(e)}") |
| 123 | return {"success": False, "error": str(e)} |
| 124 | |
| 125 | |
| 126 | # Singleton instance |
| 127 | upload_post_service = UploadPostService() |
| 128 | |
| 129 | |
| 130 | def cross_post_video( |
| 131 | video_path: str, |
| 132 | title: str, |
| 133 | platforms: Optional[list] = None, |
| 134 | youtube_extra: Optional[dict] = None, |
| 135 | ) -> dict: |
| 136 | return upload_post_service.upload_video(video_path, title, platforms, youtube_extra=youtube_extra) |
| 137 |