""" 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)