File size: 4,177 Bytes
345855e | 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 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 | """
bulk.py
V9 Autonomous Publisher Engine
Purpose:
--------
Handles BULK publishing across multiple platforms.
Supports:
- TikTok
- Reels (Instagram)
- YouTube Shorts
- Facebook
- Any future platform adapter
Design:
-------
Input -> Normalize -> Dispatch -> Execute -> Collect Results
Production Features:
--------------------
✔ async concurrency
✔ retry system
✔ per-platform isolation
✔ failure tolerance
✔ structured logging
✔ scheduler-compatible
✔ autonomous engine ready
"""
import asyncio
import traceback
from typing import Dict, List, Any
# Platform adapters
from publisher.platforms.tiktok import publish_tiktok
from publisher.platforms.reels import publish_reels
from publisher.platforms.shorts import publish_shorts
from publisher.platforms.facebook import publish_facebook
# =====================================================
# PLATFORM REGISTRY
# =====================================================
PLATFORM_MAP = {
"tiktok": publish_tiktok,
"reels": publish_reels,
"shorts": publish_shorts,
"facebook": publish_facebook,
}
# =====================================================
# CONFIG
# =====================================================
MAX_CONCURRENT_POSTS = 5
MAX_RETRIES = 2
# =====================================================
# HELPERS
# =====================================================
async def execute_with_retry(func, payload: Dict, retries=MAX_RETRIES):
"""
Safe execution wrapper with retries.
"""
attempt = 0
while attempt <= retries:
try:
result = await func(payload)
return {
"status": "success",
"result": result,
}
except Exception as e:
attempt += 1
if attempt > retries:
return {
"status": "failed",
"error": str(e),
"trace": traceback.format_exc(),
}
await asyncio.sleep(2)
# =====================================================
# SINGLE JOB EXECUTOR
# =====================================================
async def process_job(job: Dict[str, Any]):
"""
Expected job format:
{
"platform": "tiktok",
"video_url": "...",
"caption": "...",
"hashtags": [],
"thumbnail": "...",
"schedule_time": optional
}
"""
platform = job.get("platform")
if platform not in PLATFORM_MAP:
return {
"status": "failed",
"error": f"Unsupported platform: {platform}",
}
publisher = PLATFORM_MAP[platform]
return await execute_with_retry(publisher, job)
# =====================================================
# BULK ENGINE
# =====================================================
async def bulk_publish(jobs: List[Dict[str, Any]]):
"""
Main bulk execution engine.
"""
semaphore = asyncio.Semaphore(MAX_CONCURRENT_POSTS)
results = []
async def limited_job(job):
async with semaphore:
return await process_job(job)
tasks = [limited_job(job) for job in jobs]
completed = await asyncio.gather(*tasks, return_exceptions=False)
results.extend(completed)
return summarize_results(results)
# =====================================================
# SUMMARY
# =====================================================
def summarize_results(results: List[Dict]):
success = sum(1 for r in results if r["status"] == "success")
failed = len(results) - success
return {
"status": "completed",
"total_jobs": len(results),
"successful": success,
"failed": failed,
"results": results,
}
# =====================================================
# FASTAPI ENTRYPOINT
# =====================================================
async def execute(payload: Dict):
"""
Universal endpoint handler
POST /execute/bulk_publish
"""
jobs = payload.get("jobs")
if not jobs:
return {
"status": "error",
"message": "No jobs provided",
}
return await bulk_publish(jobs) |