post_to_twitter.py

import argparse
import sys
import logging
import os
from composio import ComposioToolSet, Action
from upload_media import upload_media

# Set debug logging
logging.basicConfig(
    level=os.environ.get("COMPOSIO_LOGGING_LEVEL", "DEBUG"),
    format="%(asctime)s - %(name)s - %(levelname)s - %(message)s",
)
logger = logging.getLogger(__name__)
logger.setLevel(logging.DEBUG)


def create_post(
    text,
    media_id=None,
    card_uri=None,
    direct_message_deep_link=None,
    for_super_followers_only=False,
    geo__place__id=None,
    media__media__ids=None,
    media__tagged__user__ids=None,
    nullcast=False,
    poll__duration__minutes=None,
    poll__options=None,
    poll__reply__settings=None,
    quote_tweet_id=None,
    reply__exclude__reply__user__ids=None,
    reply__in__reply__to__tweet__id=None,
    reply_settings=None,
):
    tool_set = ComposioToolSet()
    params = {
        "text": text,
    }
    if card_uri is not None:
        params["card_uri"] = card_uri
    if direct_message_deep_link is not None:
        params["direct_message_deep_link"] = direct_message_deep_link
    if for_super_followers_only is not False:
        params["for_super_followers_only"] = for_super_followers_only
    if media_id is not None:
        params["media__media__ids"] = (
            "[media_id,]"  # Match the API schema parameter name
        )
    if media__tagged__user__ids is not None:
        params["media.tagged_user_ids"] = media__tagged__user__ids
    if nullcast is not False:
        params["nullcast"] = nullcast
    if poll__duration__minutes is not None:
        params["poll.duration_minutes"] = poll__duration__minutes
    if poll__options is not None:
        params["poll.options"] = poll__options
    if poll__reply__settings is not None:
        params["poll.reply_settings"] = poll__reply__settings
    if quote_tweet_id is not None:
        params["quote_tweet_id"] = quote_tweet_id
    if reply__exclude__reply__user__ids is not None:
        params["reply.exclude_reply_user_ids"] = reply__exclude__reply__user__ids
    if reply__in__reply__to__tweet__id is not None:
        params["reply.in_reply_to_tweet_id"] = reply__in__reply__to__tweet__id
    if reply_settings is not None:
        params["reply_settings"] = reply_settings
    tool_set.execute_action(
        action=Action.TWITTER_CREATION_OF_A_POST,
        params=params,
    )


def post_to_twitter(text, media_id=None, in_reply_to_tweet_id=None, reply_text=None, for_super_followers_only=False):
    """Post a tweet with optional media and reply in a single API call"""
    tool_set = ComposioToolSet()
    
    # Create params for the main tweet
    params = {
        "text": text,
        "for_super_followers_only": for_super_followers_only,
    }
    
    # Add media if provided
    if media_id is not None:
        params["media__media__ids"] = [str(media_id)]
    
    # Add reply settings if this is a reply
    if in_reply_to_tweet_id is not None:
        params["reply__in__reply__to__tweet__id"] = str(in_reply_to_tweet_id)
    
    # If reply text is provided, append it to the main tweet text
    if reply_text is not None:
        params["text"] = f"{text}\n\n{reply_text}"
    
    try:
        # Single API call to post the tweet
        response = tool_set.execute_action(
            action=Action.TWITTER_CREATION_OF_A_POST,
            params=params,
        )
        
        # Log the response
        logger.debug(f"Tweet API response: {response}")
        
        if not response.get("successful", response.get("successfull", False)):
            error_msg = response.get("error", "Unknown error")
            raise ValueError(f"Failed to post tweet: {error_msg}")
            
        return response
        
    except Exception as e:
        logger.error(f"Error posting tweet: {e}")
        raise


def main():
    parser = argparse.ArgumentParser(description="Post to Twitter using Composio API")
    parser.add_argument("text", type=str, help="Text content of the tweet")
    parser.add_argument("--image", type=str, help="Path to the image file to upload")
    args = parser.parse_args()

    try:
        media_id = upload_media(args.image) if args.image else None
        response = post_to_twitter(args.text, media_id=media_id)
        if response.get("successful", response.get("successfull", False)):
            print("Tweet posted successfully!")
        else:
            print(f"Error posting tweet: {response.get('error', 'Unknown error')}")
            print(f"Full response: {response}")
            sys.exit(1)
    except Exception as e:
        print(f"Error: {str(e)}")
        if hasattr(e, "response"):
            print(f"Full error response: {e.response}")
        sys.exit(1)


if __name__ == "__main__":
    main()