equilink-site/server/connectors/twitter_connector.py

114 lines
4.1 KiB
Python
Raw Normal View History

2025-02-12 17:38:06 +05:30
#
# Module: twitter_connector
#
# This module implements the TwitterConnector class for interacting with Twitter APIs.
#
# Title: Twitter Connector
# Summary: Twitter connector implementation.
# Authors:
# - @TheBlockRhino
# Created: 2024-12-31
# Last edited by: @TheBlockRhino
# Last edited date: 2025-01-04
# URLs:
# - https://arai-ai.io
# - https://github.com/ARAI-DevHub/arai-ai-agents
# - https://x.com/TheBlockRhino
import os
import time
import tweepy
from dotenv import load_dotenv
load_dotenv()
class TwitterConnector:
"""A connector for interacting with the Twitter API via Tweepy.
This class handles OAuth 1.0a and OAuth 2.0 authentication for
posting tweets and retrieving data from Twitter.
Attributes:
api_key (str): The API key for the Twitter API.
api_secret_key (str): The API secret key for the Twitter API.
access_token (str): The access token for the Twitter API.
access_token_secret (str): The access token secret for the Twitter API.
bearer_token (str): The bearer token for the Twitter API.
bot_username (str): The username of the bot.
bot_id (str): The ID of the bot.
"""
def __init__(self):
"""Initializes the Twitter connector.
Raises:
ValueError: If Twitter API credentials are not present in the environment variables.
"""
self.api_key = os.getenv("TWITTER_API_KEY")
# Part of the OAuth 1.0a credentials identifying the application (required for user-based authentication).
self.api_secret_key = os.getenv("TWITTER_API_KEY_SECRET")
# Secret counterpart to the API key, used in signing OAuth 1.0a requests.
self.access_token = os.getenv("TWITTER_ACCESS_TOKEN")
# Represents the users OAuth 1.0a credentials, required for user-level actions (e.g., posting tweets).
self.access_token_secret = os.getenv("TWITTER_ACCESS_TOKEN_SECRET")
# Secret counterpart to the access token, used in signing user-level requests under OAuth 1.0a.
self.bearer_token = os.getenv("TWITTER_BEARER_TOKEN")
# Used for OAuth 2.0 app-only authentication in Twitter API v2, often for read-only access to public data.
if not all([self.api_key, self.api_secret_key, self.access_token, self.access_token_secret]):
raise ValueError("Twitter API credentials are not set in .env file")
try:
# Initialize v2 client only
self.client = tweepy.Client(
bearer_token=self.bearer_token,
consumer_key=self.api_key,
consumer_secret=self.api_secret_key,
access_token=self.access_token,
access_token_secret=self.access_token_secret,
wait_on_rate_limit=True
)
# Get the bot's info using v2 API
me = self.client.get_me()
self.bot_username = me.data.username.lower()
self.bot_id = me.data.id
print(f"Successfully authenticated as @{self.bot_username}")
self.last_mention_id = None
except Exception as e:
print(f"Twitter authentication failed: {str(e)}")
raise
def post_tweet(self, message: str) -> str:
"""Posts a tweet to the bot's Twitter account.
Args:
message (str): The message to be posted as a tweet. Must be shorter than 280 characters.
Returns:
str: A string message indicating success or failure.
Raises:
Exception: If there's an error posting the tweet.
Example:
>>> twitter_connector.post_tweet("Hello, world!")
"""
try:
if not message:
return "Error: Tweet message is empty"
# Truncate if too long
if len(message) > 280:
message = message[:277] + "..."
self.client.create_tweet(text=message)
return "Tweeted: " + message
except Exception as e:
return "Error posting tweet: " + str(e)