equilink-site/server/connectors/twitter_connector.py
2025-02-12 17:38:06 +05:30

114 lines
4.1 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters

This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.

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