parent
01bab56eb6
commit
447bfb0087
7 changed files with 286 additions and 197 deletions
@ -0,0 +1,125 @@ |
||||
import requests |
||||
from requests_oauthlib import OAuth1 |
||||
import logging |
||||
from datetime import datetime, timezone |
||||
from dotenv import load_dotenv |
||||
import os |
||||
import time |
||||
from foodie_config import X_API_CREDENTIALS |
||||
|
||||
# Load environment variables from .env file |
||||
load_dotenv() |
||||
|
||||
# Set up logging |
||||
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s') |
||||
|
||||
# Function to delete a tweet |
||||
def delete_tweet(tweet_id, auth): |
||||
try: |
||||
response = requests.delete(f"https://api.x.com/2/tweets/{tweet_id}", auth=auth) |
||||
response.raise_for_status() |
||||
logging.info(f"Successfully deleted tweet {tweet_id}") |
||||
return True |
||||
except Exception as e: |
||||
logging.error(f"Failed to delete tweet {tweet_id}: {e}") |
||||
return False |
||||
|
||||
# Function to check rate limits for a given author |
||||
def check_rate_limits_for_author(username, credentials, retry=False): |
||||
logging.info(f"{'Retrying' if retry else 'Checking'} rate limits for {username} (handle: {credentials['x_username']})") |
||||
|
||||
# Retrieve OAuth 1.0a credentials for the author |
||||
consumer_key = credentials["api_key"] |
||||
consumer_secret = credentials["api_secret"] |
||||
access_token = credentials["access_token"] |
||||
access_token_secret = credentials["access_token_secret"] |
||||
|
||||
# Validate credentials |
||||
if not all([consumer_key, consumer_secret, access_token, access_token_secret]): |
||||
logging.error(f"Missing OAuth credentials for {username} in X_API_CREDENTIALS.") |
||||
return None |
||||
|
||||
# Set up OAuth 1.0a authentication |
||||
auth = OAuth1(consumer_key, consumer_secret, access_token, access_token_secret) |
||||
|
||||
# Add delay to avoid IP-based rate limiting |
||||
logging.info(f"Waiting 5 seconds before attempting to post for {username}") |
||||
time.sleep(5) |
||||
|
||||
# Try posting a test tweet to get v2 rate limit headers |
||||
tweet_id = None |
||||
try: |
||||
tweet_data = {"text": f"Test tweet to check rate limits for {username} - please ignore"} |
||||
response = requests.post("https://api.x.com/2/tweets", json=tweet_data, auth=auth) |
||||
response.raise_for_status() |
||||
tweet_id = response.json()['data']['id'] |
||||
logging.info("Successfully posted test tweet for %s: %s", username, response.json()) |
||||
logging.info("Response Headers for %s: %s", username, response.headers) |
||||
# Extract rate limit headers if present |
||||
app_limit = response.headers.get('x-app-limit-24hour-limit', 'N/A') |
||||
app_remaining = response.headers.get('x-app-limit-24hour-remaining', 'N/A') |
||||
app_reset = response.headers.get('x-app-limit-24hour-reset', 'N/A') |
||||
logging.info("App 24-Hour Tweet Limit for %s: %s", username, app_limit) |
||||
logging.info("App 24-Hour Tweets Remaining for %s: %s", username, app_remaining) |
||||
if app_reset != 'N/A': |
||||
reset_time = datetime.fromtimestamp(int(app_reset), timezone.utc).strftime('%Y-%m-%d %H:%M:%S UTC') |
||||
logging.info("App 24-Hour Reset (Readable) for %s: %s", username, reset_time) |
||||
return tweet_id |
||||
except requests.exceptions.HTTPError as e: |
||||
logging.info("Test Tweet Response Status Code for %s: %s", username, e.response.status_code) |
||||
logging.info("Test Tweet Response Headers for %s: %s", username, e.response.headers) |
||||
if e.response.status_code == 429: |
||||
logging.info("Rate Limit Exceeded for /2/tweets for %s", username) |
||||
|
||||
# Extract user-specific 24-hour limits |
||||
user_limit = e.response.headers.get('x-user-limit-24hour-limit', 'N/A') |
||||
user_remaining = e.response.headers.get('x-user-limit-24hour-remaining', 'N/A') |
||||
user_reset = e.response.headers.get('x-user-limit-24hour-reset', 'N/A') |
||||
logging.info("User 24-Hour Tweet Limit for %s: %s", username, user_limit) |
||||
logging.info("User 24-Hour Tweets Remaining for %s: %s", username, user_remaining) |
||||
logging.info("User 24-Hour Reset (Timestamp) for %s: %s", username, user_reset) |
||||
if user_reset != 'N/A': |
||||
reset_time = datetime.fromtimestamp(int(user_reset), timezone.utc).strftime('%Y-%m-%d %H:%M:%S UTC') |
||||
logging.info("User 24-Hour Reset (Readable) for %s: %s", username, reset_time) |
||||
|
||||
# Extract app-specific 24-hour limits |
||||
app_limit = e.response.headers.get('x-app-limit-24hour-limit', 'N/A') |
||||
app_remaining = e.response.headers.get('x-app-limit-24hour-remaining', 'N/A') |
||||
app_reset = e.response.headers.get('x-app-limit-24hour-reset', 'N/A') |
||||
logging.info("App 24-Hour Tweet Limit for %s: %s", username, app_limit) |
||||
logging.info("App 24-Hour Tweets Remaining for %s: %s", username, app_remaining) |
||||
logging.info("App 24-Hour Reset (Timestamp) for %s: %s", username, app_reset) |
||||
if app_reset != 'N/A': |
||||
reset_time = datetime.fromtimestamp(int(app_reset), timezone.utc).strftime('%Y-%m-%d %H:%M:%S UTC') |
||||
logging.info("App 24-Hour Reset (Readable) for %s: %s", username, reset_time) |
||||
return None |
||||
except Exception as e: |
||||
logging.error("Failed to post test tweet for %s: %s", username, e) |
||||
return None |
||||
|
||||
# Main loop to check rate limits for all authors |
||||
if __name__ == "__main__": |
||||
# First pass: Attempt to post for all authors |
||||
successful_tweets = {} |
||||
for username, credentials in X_API_CREDENTIALS.items(): |
||||
tweet_id = check_rate_limits_for_author(username, credentials) |
||||
if tweet_id: |
||||
successful_tweets[username] = (tweet_id, credentials) |
||||
logging.info("-" * 50) |
||||
|
||||
# Delete successful tweets to free up quota |
||||
for username, (tweet_id, credentials) in successful_tweets.items(): |
||||
auth = OAuth1( |
||||
credentials["api_key"], |
||||
credentials["api_secret"], |
||||
credentials["access_token"], |
||||
credentials["access_token_secret"] |
||||
) |
||||
delete_tweet(tweet_id, auth) |
||||
|
||||
# Second pass: Retry for authors that failed |
||||
logging.info("Retrying for authors that initially failed...") |
||||
for username, credentials in X_API_CREDENTIALS.items(): |
||||
if username not in successful_tweets: |
||||
check_rate_limits_for_author(username, credentials, retry=True) |
||||
logging.info("-" * 50) |
||||
Loading…
Reference in new issue