nodes/socialmedia/x.py

import logging
from typing import Union, Dict
import tweepy
from tenacity import retry, stop_after_attempt, wait_exponential

# Configure logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

class TwitterError(Exception):
    """Custom exception for Twitter API errors"""
    pass

def validate_credentials(login: Dict[str, str]) -> bool:
    """Validate Twitter API credentials"""
    required_keys = ['api_key', 'secret_key', 'bearer_key', 'access_token', 'access_token_secret']
    return all(key in login and login[key] for key in required_keys)

@retry(
    stop=stop_after_attempt(3),
    wait=wait_exponential(multiplier=1, min=4, max=10),
    reraise=True
)
def create_tweet(text: str, login: Dict[str, str]) -> Union[Dict[str, str], Dict[str, str]]:
    """
    Create a tweet with improved error handling and retries
    
    Args:
        text (str): Tweet content
        login (dict): Twitter API credentials
        
    Returns:
        dict: Tweet information if successful
        dict: Error information if failed
    """
    try:
        if not text:
            return {"error": "No tweet content provided"}

        if not validate_credentials(login):
            return {"error": "Invalid or missing Twitter credentials"}

        if len(text) > 280:
            return {"error": "Tweet exceeds 280 characters"}

        # Initialize Twitter client
        client = tweepy.Client(
            bearer_token=login['bearer_key'],
            consumer_key=login['api_key'],
            consumer_secret=login['secret_key'],
            access_token=login['access_token'],
            access_token_secret=login['access_token_secret']
        )

        # Create tweet
        response = client.create_tweet(text=text)

        if response and response.data:
            tweet_id = response.data['id']
            return {
                "success": True,
                "tweet_id": tweet_id,
                "url": f"https://twitter.com/user/status/{tweet_id}"
            }
        else:
            return {"error": "Failed to create tweet: No response data"}

    except tweepy.TweepyException as e:
        logger.error(f"Twitter API error: {str(e)}")
        return {"error": f"Twitter API error: {str(e)}"}
    except Exception as e:
        logger.error(f"Unexpected error creating tweet: {str(e)}")
        return {"error": f"Failed to create tweet: {str(e)}"}