foodie-automator/check_rate_limits.py

import logging

logging.basicConfig(
    filename='/home/shane/foodie_automator/logs/check_x_capacity.log',
    level=logging.DEBUG,
    format='%(asctime)s - %(levelname)s - %(message)s'
)

import requests
from requests_oauthlib import OAuth1
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()

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