File size: 3,850 Bytes
0396b0d | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 | import tweepy
import datetime
import os
def get_tweets_from_user_list_v2(user_list, client, tweets_per_user=10):
"""
Retrieves the latest tweets from a list of Twitter users within the last 24 hours using Tweepy API v2.
Args:
user_list (list): A list of Twitter usernames (strings).
client (tweepy.Client): An authenticated Tweepy Client object.
tweets_per_user (int): The maximum number of tweets to retrieve per user.
Returns:
list: A list of tweet objects (dict) that are within the last 24 hours. The structure of the tweet objects is different in v2.
Returns an empty list if there are errors.
"""
all_tweets = []
now = datetime.datetime.now(datetime.timezone.utc) # Important: Use UTC for consistency
yesterday = now - datetime.timedelta(days=1)
for username in user_list:
try:
# Get the user ID from the username
user = client.get_user(username=username)
if user.errors:
print(f"Error getting user ID for {username}: {user.errors}")
continue # Skip to the next user
user_id = user.data['id']
# Fetch tweets from the user's timeline
response = client.get_users_tweets(
id=user_id,
max_results=tweets_per_user,
tweet_fields=["created_at", "text"], # Specify the fields you need
)
if response.errors:
print(f"Error fetching tweets for {username} (ID: {user_id}): {response.errors}")
continue # Skip to the next user
tweets = response.data
if tweets:
for tweet in tweets:
tweet_time = tweet['created_at']
if tweet_time >= yesterday:
all_tweets.append(tweet)
else:
break # Optimization: Stop searching for this user
except Exception as e: # Catch any other unexpected errors
print(f"Unexpected error fetching tweets for {username}: {e}")
continue
return all_tweets
def authenticate_v2():
"""Authenticates with the Twitter API v2 using a Bearer Token.
Returns:
tweepy.Client: An authenticated Tweepy Client object, or None if authentication fails.
"""
bearer_token = (f"AAAAAAAAAAAAAAAAAAAAAKmVzAEAAAAAhVM%2BPoJytUvDoQ%2FgIGfLdJXdocU%3DvS5ZF10O60kLJcY6MGiErCDHV7awcUKB75QqWpIlKh84rK5CFw") # Get the Bearer Token from env
if not bearer_token:
print("Error: Missing Twitter Bearer Token. Please set the environment variable.")
return None
try:
client = tweepy.Client(bearer_token)
# No direct "verify_credentials" in v2. You can try a simple API call to check.
user = client.get_me()
if user.errors:
print(f"Authentication check failed: {user.errors}")
return None
print("Authentication successful (API v2)!")
return client
except Exception as e:
print(f"Authentication failed (API v2): {e}")
return None
if __name__ == '__main__':
# Authenticate with the Twitter API v2
client = authenticate_v2()
if client is None:
print("Exiting due to authentication failure.")
exit()
# Example Usage:
user_list = ["dacefupan","aiwangupiao"]
tweets = get_tweets_from_user_list_v2(user_list, client, tweets_per_user=5)
if tweets:
print(f"Found {len(tweets)} tweets from the last 24 hours:")
for tweet in tweets:
print(f"Text: {tweet['text']}\n") # Access tweet text
print(f"Tweeted at: {tweet['created_at']}\n") #Access tweet created at time
else:
print("No tweets found in the last 24 hours or an error occurred.") |