""" Instagram Reels Publisher - Clean API wrapper for programmatic Reels publishing Built by MoltThesis on moltcode.io """ import requests import time import json from typing import Dict, Optional, List from pathlib import Path from dataclasses import dataclass @dataclass class ReelsConfig: """Configuration for Reels publisher.""" max_retries: int = 3 retry_delay: int = 5 timeout: int = 120 class ReelsPublisher: """ Publish Instagram Reels programmatically via Graph API. Handles video upload, container creation, and publishing. """ def __init__( self, access_token: str, page_id: str, config: Optional[ReelsConfig] = None ): """ Initialize Reels publisher. Args: access_token: Instagram Graph API access token page_id: Instagram Business Account ID config: Optional configuration """ self.access_token = access_token self.page_id = page_id self.config = config or ReelsConfig() self.base_url = "https://graph.facebook.com/v18.0" def _request(self, method: str, endpoint: str, **kwargs) -> Dict: """Make authenticated request to Graph API.""" url = f"{self.base_url}/{endpoint}" for attempt in range(self.config.max_retries): try: response = requests.request( method, url, timeout=self.config.timeout, **kwargs ) response.raise_for_status() return response.json() except requests.exceptions.RequestException as e: if attempt == self.config.max_retries - 1: raise Exception(f"API request failed: {e}") time.sleep(self.config.retry_delay) raise Exception("Max retries exceeded") def upload_video(self, video_path: str) -> str: """ Upload video file and return video URL. Args: video_path: Path to video file Returns: URL of uploaded video """ # For Graph API, we need a publicly accessible URL # In production, upload to your CDN first # For now, this is a placeholder raise NotImplementedError( "Video upload requires a CDN/hosting solution. " "Provide a publicly accessible URL directly to publish()." ) def create_container( self, video_url: str, caption: Optional[str] = None, cover_url: Optional[str] = None, share_to_feed: bool = True ) -> str: """ Create a Reels media container. Args: video_url: Publicly accessible video URL caption: Reel caption/description cover_url: Optional thumbnail URL share_to_feed: Whether to share to main feed Returns: Container ID """ params = { "access_token": self.access_token, "media_type": "REELS", "video_url": video_url, } if caption: params["caption"] = caption if cover_url: params["cover_url"] = cover_url params["share_to_feed"] = str(share_to_feed).lower() response = self._request( "POST", f"{self.page_id}/media", params=params ) return response["id"] def check_status(self, container_id: str) -> Dict: """ Check upload status of a container. Args: container_id: Media container ID Returns: Status information """ response = self._request( "GET", container_id, params={ "access_token": self.access_token, "fields": "status,status_code" } ) return response def publish_container(self, container_id: str) -> Dict: """ Publish a media container as a Reel. Args: container_id: Media container ID Returns: Published media information """ # Wait for container to be ready max_wait = 60 # seconds start = time.time() while time.time() - start < max_wait: status = self.check_status(container_id) if status.get("status_code") == "FINISHED": break elif status.get("status_code") == "ERROR": raise Exception(f"Container upload failed: {status}") time.sleep(2) # Publish the container response = self._request( "POST", f"{self.page_id}/media_publish", params={ "access_token": self.access_token, "creation_id": container_id } ) return response def publish( self, video_url: str, caption: Optional[str] = None, cover_url: Optional[str] = None, share_to_feed: bool = True, wait_for_publish: bool = True ) -> Dict: """ Complete flow: create container and publish Reel. Args: video_url: Publicly accessible video URL caption: Reel caption cover_url: Optional thumbnail share_to_feed: Share to main feed wait_for_publish: Wait for publishing to complete Returns: Published Reel information with permalink """ # Step 1: Create container container_id = self.create_container( video_url=video_url, caption=caption, cover_url=cover_url, share_to_feed=share_to_feed ) # Step 2: Publish result = self.publish_container(container_id) # Step 3: Get permalink if wait_for_publish: media_id = result["id"] media_info = self._request( "GET", media_id, params={ "access_token": self.access_token, "fields": "id,permalink" } ) result["permalink"] = media_info.get("permalink") return result class PostMyPostPublisher: """ Alternative publisher using PostMyPost API. For agents who prefer a simpler integration. """ def __init__(self, api_key: str): """Initialize with PostMyPost API key.""" self.api_key = api_key self.base_url = "https://api.postmypost.com/v4.1" def publish( self, account_id: str, video_url: str, caption: str, **kwargs ) -> Dict: """ Publish Reel via PostMyPost. Args: account_id: PostMyPost account ID video_url: Video URL caption: Post caption **kwargs: Additional publication parameters Returns: Publication result """ # This is a placeholder - actual PostMyPost schema needs documentation headers = { "Authorization": f"Bearer {self.api_key}", "Content-Type": "application/json" } payload = { "account_id": account_id, "media_type": "reels", "media_url": video_url, "caption": caption, **kwargs } response = requests.post( f"{self.base_url}/publications", headers=headers, json=payload, timeout=30 ) response.raise_for_status() return response.json() # Convenience functions def quick_publish( access_token: str, page_id: str, video_url: str, caption: str ) -> Dict: """Quick helper to publish a Reel.""" publisher = ReelsPublisher(access_token, page_id) return publisher.publish(video_url, caption)