import snscrape.modules.twitter as sntwitter
import pandas as pd
import datetime as dt
from tqdm import tqdm
import requests
from scripts import sentiment
import tweepy
import configparser
import os
import pandas as pd
from datetime import datetime, date, timedelta
def get_latest_account_tweets(handle):
try:
if os.path.exists("tweepy_auth.ini"):
config = configparser.ConfigParser()
config.read("tweepy_auth.ini")
# Get the authentication details
authentication_section = config["AUTHENTICATION"]
consumer_key = authentication_section["twitter_consumer_key"]
consumer_secret = authentication_section["twitter_consumer_secret"]
access_token = authentication_section["twitter_access_token"]
access_token_secret = authentication_section["twitter_access_token_secret"]
else:
consumer_key = os.environ["twitter_consumer_key"]
consumer_secret = os.environ["twitter_consumer_secret"]
access_token = os.environ["twitter_access_token"]
access_token_secret = os.environ["twitter_access_token_secret"]
auth = tweepy.OAuthHandler(consumer_key, consumer_secret)
auth.set_access_token(access_token, access_token_secret)
# create the API object
api = tweepy.API(auth)
# load the tweets from a specific user
tweets = api.user_timeline(
screen_name=handle, count=10000000, tweet_mode="extended"
)
df_tweets = pd.DataFrame(data=[t._json for t in tweets])
df_tweets["created_at"] = pd.to_datetime(df_tweets["created_at"])
df_tweets = df_tweets.sort_values("created_at")
# print the tweet texts
tweets_txt = []
for tweet in tweets:
tweets_txt.append(sentiment.tweet_cleaner(tweet.full_text))
df_tweets["clean_text"] = tweets_txt
df_tweets["handle"] = df_tweets.user.iloc[0]["screen_name"]
return df_tweets
except tweepy.TweepError as e:
# Handle specific error conditions
if e.api_code == 63:
print("User has been suspended.")
elif e.api_code == 88:
print("Rate limit exceeded. Please try again later.")
else:
print("Error occurred during API call:", str(e))
return str(e)
except Exception as e:
print("An error occurred:", str(e))
return str(e)
return None
def get_tweets(
handle: str,
):
"""
Fetches tweets from Twitter based on a given query and returns a list of extracted tweet information.
Args:
query (str): The query to search for tweets on Twitter.
Returns:
A list of extracted tweet information.
"""
# Get the current date
today = datetime.today()
two_months_ago = today - timedelta(days=2 * 30)
start_date = two_months_ago.strftime("%Y-%m-%d")
end_date = today.strftime("%Y-%m-%d")
query = f"from:{handle} since:{start_date} until:{end_date} -filter:replies -filter:retweets"
fetched_tweets = sntwitter.TwitterSearchScraper(query).get_items()
tweets = [extract_tweet_info(tweet) for tweet in tqdm(fetched_tweets)]
df_tweets = pd.DataFrame(tweets)
df_tweets["full_text"] = df_tweets["content"]
df_tweets["clean_text"] = df_tweets["full_text"].apply(
lambda r: sentiment.tweet_cleaner(r)
)
df_tweets["handle"] = df_tweets["username"]
df_tweets["created_at"] = df_tweets["date"]
return df_tweets
def get_replies(username: str, conversation_id: str, max_tweets: int) -> list:
"""
Fetches the replies for a given Twitter user and conversation, and returns a list of extracted tweet information.
Args:
username (str): The username of the Twitter user whose replies are to be fetched.
conversation_id (str): The ID of the conversation for which replies are to be fetched.
Returns:
A list of extracted tweet information for the replies.
"""
print(
f"Fetching replies for username {username} and conversation {conversation_id}"
)
query = f"to:{username} since_id:{conversation_id} filter:safe"
tweets_list = []
for i, tweet in tqdm(enumerate(sntwitter.TwitterSearchScraper(query).get_items())):
if i > max_tweets:
break
else:
tweets_list.append(extract_tweet_info(tweet))
return tweets_list
def get_tweet_by_id_and_username(username: str, tweet_id: str):
"""
Fetches a tweet from Twitter based on the given username and tweet ID.
Args:
username (str): The username of the Twitter user who posted the tweet.
tweet_id (str): The ID of the tweet to fetch.
Returns:
The fetched tweet.
"""
tweet_url = f"https://twitter.com/{username}/status/{tweet_id}"
return sntwitter.TwitterSearchScraper(tweet_url).get_items()
def extract_tweet_info(tweet):
"""
Extracts relevant information from a tweet object and returns a dictionary with the extracted values.
Args:
tweet: A tweet object.
Returns:
A dictionary with the extracted tweet information.
"""
return {
"date": tweet.date,
"username": tweet.user.username,
"content": tweet.rawContent,
"retweet_count": tweet.retweetCount,
"tweet_id": tweet.id,
"like_count": tweet.likeCount,
"reply_count": tweet.replyCount,
"in_reply_to_tweet_id": tweet.inReplyToTweetId,
"conversation_id": tweet.conversationId,
"view_count": tweet.viewCount,
}
def get_follower_ids(username: str, limit: int = 20):
"""
Retrieves a list of Twitter IDs for users who follow a given Twitter handle.
Args:
username (str): The Twitter handle to retrieve follower IDs for.
limit (int): The maximum number of follower IDs to retrieve.
Returns:
A list of Twitter user IDs (as strings).
"""
# Construct the search query using snscrape
query = f"from:{username} replies:True"
start_date = dt.date(year=2023, month=3, day=10)
end_date = dt.date(year=2023, month=3, day=22)
query = f"from:{username} since:{start_date} until:{end_date}"
tweets = get_tweets(query=query)
one_tweet = tweets[-1]
one_tweet_id = one_tweet["tweet_id"]
replies = get_replies(
username=username, conversation_id=one_tweet_id, max_tweets=1000
)
return one_tweet, replies
def get_twitter_account_info(twitter_handle: str) -> dict:
"""
Extracts the name, username, follower count, and last tweet of a Twitter user using snscrape.
Args:
twitter_handle (str): The Twitter username to retrieve information for.
Returns:
dict: A dictionary containing the name, username, follower count, and last tweet of the Twitter user.
"""
# Create a TwitterUserScraper object
user_scraper = sntwitter.TwitterUserScraper(twitter_handle)
# Get the user's profile information
user_profile = user_scraper.entity
check_string = lambda s: "false" if str(s).lower() == "false" else "true"
return {
"name": user_profile.displayname,
"username": user_profile.username,
"user_id": user_profile.id,
"follower_count": user_profile.followersCount,
"friends_count": user_profile.friendsCount,
"verified": check_string(user_profile.verified),
}
if __name__ == "__main__":
## Testing extracting tweets from an account
# Set the search variables (dates for when account tweeted. Does not take into account replies)
account = "taylorlorenz"
start_date = dt.date(year=2023, month=2, day=1)
end_date = dt.date(year=2023, month=3, day=11)
# Format the query string
query = f"from:{account} since:{start_date} until:{end_date}"
print(f"query: {query}")
tweets = get_tweets(query=query)
df_tweets = pd.DataFrame(data=tweets)
df_tweets = df_tweets.sort_values("in_reply_to_tweet_id")
# Uncomment to save output
df_tweets.to_csv("df_tweets.csv")
print(df_tweets.head(2))
print(df_tweets.tail(2))
print(f"Total Tweets: {len(tweets)}")
## Testing extracting conversatin threeds from conversation Id
conversation_id = (
1620650202305798144 # A tweet from elon musk about turbulent times
)
max_tweets = 3000
tweets = get_replies(
username="elonmusk", conversation_id=conversation_id, max_tweets=max_tweets
)
df_replies = pd.DataFrame(data=tweets)
# Uncomment to save output
# df_replies.to_csv("df_replies.csv")
print(
f"Number of extracted tweets from conversation_id: {conversation_id}, {len(tweets)}"
)