foodie_dev
/
foodie-automator
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

try

Browse Source
main
Shane 7 months ago
parent 5ea9f20dd8
commit 83e69a35b7
2 changed files with 117 additions and 105 deletions
Whitespace
Show all changes Ignore whitespace when comparing lines Ignore changes in amount of whitespace Ignore changes in whitespace at EOL
Split View
Diff Options
Show Stats Download Patch File Download Diff File
  1. 205
      foodie_utils.py
  2. 1
      requirements.txt

205
foodie_utils.py
Unescape View File

@ -12,6 +12,7 @@ import shutil
import requests
import time
import openai
from requests_oauthlib import OAuth1
from dotenv import load_dotenv
from datetime import datetime, timezone, timedelta
from openai import OpenAI
@ -161,79 +162,68 @@ def generate_article_tweet(author, post, persona):
logging.info(f"Generated tweet: {tweet}")
return tweet
def post_tweet(author, tweet, reply_to_id=None):
def post_tweet(author, content, media_ids=None, reply_to_id=None):
"""
Post a tweet after checking real-time X API rate limits.
Updates rate_limit_info.json with API-provided data.
Post a tweet for an author using X API v2.
Returns (tweet_id, tweet_data) if successful, (None, None) if rate-limited or failed.
"""
from foodie_config import X_API_CREDENTIALS
import tweepy
logger = logging.getLogger(__name__)
credentials = X_API_CREDENTIALS.get(author["username"])
username = author['username']
credentials = X_API_CREDENTIALS.get(username)
if not credentials:
logger.error(f"No X credentials found for {author['username']}")
return False
logger.error(f"No X API credentials for {username}")
return None, None
# Check rate limit before posting
if check_author_rate_limit(author):
logger.error(f"Cannot post tweet for {author['username']}: Rate limit exceeded")
return False
# Check rate limit
can_post, remaining, reset = check_author_rate_limit(author)
if not can_post:
reset_time = datetime.fromtimestamp(reset, tz=timezone.utc).strftime('%Y-%m-%d %H:%M:%S')
logger.info(f"Cannot post tweet for {username}: rate-limited. Remaining: {remaining}, Reset at: {reset_time}")
return None, None
logger.debug(f"Attempting to post tweet for {author['username']} (handle: {credentials['x_username']})")
logger.debug(f"Tweet content: {tweet}")
oauth = OAuth1(
client_key=credentials['api_key'],
client_secret=credentials['api_secret'],
resource_owner_key=credentials['access_token'],
resource_owner_secret=credentials['access_token_secret']
)
url = 'https://api.x.com/2/tweets'
payload = {'text': content}
if media_ids:
payload['media'] = {'media_ids': media_ids}
if reply_to_id:
logger.debug(f"Replying to tweet ID: {reply_to_id}")
rate_limit_file = '/home/shane/foodie_automator/rate_limit_info.json'
rate_limit_info = load_json_file(rate_limit_file, default={})
username = author["username"]
payload['reply'] = {'in_reply_to_tweet_id': reply_to_id}
try:
client = tweepy.Client(
consumer_key=credentials["api_key"],
consumer_secret=credentials["api_secret"],
access_token=credentials["access_token"],
access_token_secret=credentials["access_token_secret"]
)
response = client.create_tweet(
text=tweet,
in_reply_to_tweet_id=reply_to_id
)
tweet_id = response.data['id']
logger.info(f"Successfully posted tweet {tweet_id} for {author['username']} (handle: {credentials['x_username']}): {tweet}")
response = requests.post(url, json=payload, auth=oauth)
headers = response.headers
# Update rate limit info with fresh API data
remaining, reset = get_x_rate_limit_status(author)
if remaining is not None and reset is not None:
rate_limit_info[username] = {
'tweet_remaining': max(0, remaining - 1), # Account for this tweet
'tweet_reset': reset
}
# Update rate limit info
rate_limit_file = '/home/shane/foodie_automator/rate_limit_info.json'
rate_limit_info = load_json_file(rate_limit_file, default={})
remaining = int(headers.get('x-user-limit-24hour-remaining', remaining))
reset = int(headers.get('x-user-limit-24hour-reset', reset))
rate_limit_info[username] = {'tweet_remaining': remaining, 'tweet_reset': reset}
save_json_file(rate_limit_file, rate_limit_info)
logger.info(f"Updated rate limit for {username}: {rate_limit_info[username]['tweet_remaining']} remaining, reset at {datetime.fromtimestamp(reset, tz=timezone.utc)}")
else:
logger.warning(f"Failed to update rate limit info for {username} after posting")
return {"id": tweet_id}
except tweepy.TweepyException as e:
logger.error(f"Failed to post tweet for {author['username']} (handle: {credentials['x_username']}): {e}")
if hasattr(e, 'response') and e.response and e.response.status_code == 429:
remaining, reset = get_x_rate_limit_status(author)
if remaining is None:
remaining = 0
reset = time.time() + 86400
rate_limit_info[username] = {
'tweet_remaining': remaining,
'tweet_reset': reset
}
save_json_file(rate_limit_file, rate_limit_info)
if response.status_code == 201:
tweet_data = response.json()
tweet_id = tweet_data.get('data', {}).get('id')
logger.info(f"Successfully tweeted for {username}: {content[:50]}... (ID: {tweet_id})")
return tweet_id, tweet_data
elif response.status_code == 429:
logger.info(f"Rate limit exceeded for {username}: {remaining} remaining, reset at {datetime.fromtimestamp(reset, tz=timezone.utc)}")
return False
return None, None
elif response.status_code == 403:
logger.error(f"403 Forbidden for {username}: {response.text}")
return None, None
else:
logger.error(f"Failed to tweet for {username}: {response.status_code} - {response.text}")
return None, None
except Exception as e:
logger.error(f"Unexpected error posting tweet for {author['username']} (handle: {credentials['x_username']}): {e}", exc_info=True)
return False
logger.error(f"Unexpected error posting tweet for {username}: {e}", exc_info=True)
return None, None
def select_best_persona(interest_score, content=""):
logging.info("Using select_best_persona with interest_score and content")
@ -1173,15 +1163,16 @@ def check_rate_limit(response):
def check_author_rate_limit(author, max_tweets=17, tweet_window_seconds=86400):
"""
Check if an author is rate-limited for tweets using real-time X API data.
Returns (can_post, remaining, reset_timestamp) where can_post is False if rate-limited.
Caches API results in memory for the current script run.
Check if an author is rate-limited for tweets using real-time X API v2 data.
Returns (can_post, remaining, reset_timestamp) where can_post is True if tweets are available.
Caches API results in memory for 1 minute.
Falls back to rate_limit_info.json or assumes 1 tweet remaining if API fails.
"""
logger = logging.getLogger(__name__)
rate_limit_file = '/home/shane/foodie_automator/rate_limit_info.json'
current_time = time.time()
# In-memory cache for rate limit status (reset per script run)
# In-memory cache
if not hasattr(check_author_rate_limit, "cache"):
check_author_rate_limit.cache = {}
@ -1194,15 +1185,15 @@ def check_author_rate_limit(author, max_tweets=17, tweet_window_seconds=86400):
else:
remaining, reset = get_x_rate_limit_status(author)
if remaining is None or reset is None:
# Fallback: Load from rate_limit_info.json or assume rate-limited
# Fallback: Load from rate_limit_info.json or assume 1 tweet remaining
rate_limit_info = load_json_file(rate_limit_file, default={})
if username not in rate_limit_info or current_time >= rate_limit_info.get(username, {}).get('tweet_reset', 0):
rate_limit_info[username] = {
'tweet_remaining': 0, # Conservative assumption
'tweet_remaining': 1, # Allow one tweet to avoid blocking
'tweet_reset': current_time + tweet_window_seconds
}
save_json_file(rate_limit_file, rate_limit_info)
remaining = rate_limit_info[username].get('tweet_remaining', 0)
remaining = rate_limit_info[username].get('tweet_remaining', 1)
reset = rate_limit_info[username].get('tweet_reset', current_time + tweet_window_seconds)
logger.warning(f"X API rate limit check failed for {username}, using fallback: {remaining} remaining")
check_author_rate_limit.cache[cache_key] = (remaining, reset)
@ -1214,6 +1205,11 @@ def check_author_rate_limit(author, max_tweets=17, tweet_window_seconds=86400):
else:
logger.info(f"Rate limit for {username}: {remaining}/{max_tweets} tweets remaining")
# Update rate_limit_info.json
rate_limit_info = load_json_file(rate_limit_file, default={})
rate_limit_info[username] = {'tweet_remaining': remaining, 'tweet_reset': reset}
save_json_file(rate_limit_file, rate_limit_info)
return can_post, remaining, reset
def get_next_author_round_robin():
@ -1238,45 +1234,60 @@ def get_next_author_round_robin():
def get_x_rate_limit_status(author):
"""
Query X API for the user's tweet rate limit status.
Returns (remaining, reset_timestamp) or (None, None) if the query fails.
Check the X API v2 rate limit status for an author by attempting a test tweet.
Returns (remaining, reset) where remaining is the number of tweets left in the 24-hour window,
and reset is the Unix timestamp when the limit resets.
Returns (None, None) if the check fails.
"""
from foodie_config import X_API_CREDENTIALS
import tweepy
logger = logging.getLogger(__name__)
credentials = X_API_CREDENTIALS.get(author["username"])
username = author['username']
credentials = X_API_CREDENTIALS.get(username)
if not credentials:
logger.error(f"No X credentials for {author['username']}")
logger.error(f"No X API credentials found for {username}")
return None, None
try:
client = tweepy.Client(
consumer_key=credentials["api_key"],
consumer_secret=credentials["api_secret"],
access_token=credentials["access_token"],
access_token_secret=credentials["access_token_secret"]
)
# Tweepy v2 doesn't directly expose rate limit status, so use API v1.1 for rate limit check
api = tweepy.API(
tweepy.OAuth1UserHandler(
consumer_key=credentials["api_key"],
consumer_secret=credentials["api_secret"],
access_token=credentials["access_token"],
access_token_secret=credentials["access_token_secret"]
)
oauth = OAuth1(
client_key=credentials['api_key'],
client_secret=credentials['api_secret'],
resource_owner_key=credentials['access_token'],
resource_owner_secret=credentials['access_token_secret']
)
rate_limits = api.rate_limit_status()
tweet_limits = rate_limits["resources"]["statuses"]["/statuses/update"]
remaining = tweet_limits["remaining"]
reset = tweet_limits["reset"]
logger.info(f"X API rate limit for {author['username']}: {remaining} remaining, reset at {datetime.fromtimestamp(reset, tz=timezone.utc)}")
return remaining, reset
except tweepy.TweepyException as e:
logger.error(f"Failed to fetch X rate limit for {author['username']}: {e}")
url = 'https://api.x.com/2/tweets'
payload = {'text': f'Test tweet to check rate limits for {username} - please ignore'}
try:
response = requests.post(url, json=payload, auth=oauth)
headers = response.headers
# Extract rate limit info from headers
remaining = int(headers.get('x-user-limit-24hour-remaining', 0))
reset = int(headers.get('x-user-limit-24hour-reset', 0))
if response.status_code == 201:
# Tweet posted successfully, delete it
tweet_id = response.json().get('data', {}).get('id')
if tweet_id:
delete_url = f'https://api.x.com/2/tweets/{tweet_id}'
delete_response = requests.delete(delete_url, auth=oauth)
if delete_response.status_code == 200:
logger.info(f"Successfully deleted test tweet {tweet_id} for {username}")
else:
logger.warning(f"Failed to delete test tweet {tweet_id} for {username}: {delete_response.status_code}")
elif response.status_code == 429:
# Rate limit exceeded
logger.info(f"Rate limit exceeded for {username}: {remaining} remaining, reset at {datetime.fromtimestamp(reset, tz=timezone.utc)}")
elif response.status_code == 403:
# Forbidden (e.g., account restrictions), but headers may still provide rate limit info
logger.warning(f"403 Forbidden for {username}, but rate limit info available: {remaining} remaining, reset at {datetime.fromtimestamp(reset, tz=timezone.utc)}")
else:
logger.error(f"Unexpected response for {username}: {response.status_code} - {response.text}")
return None, None
logger.info(f"Rate limit for {username}: {remaining} remaining, reset at {datetime.fromtimestamp(reset, tz=timezone.utc)}")
return remaining, reset
except Exception as e:
logger.error(f"Unexpected error fetching X rate limit for {author['username']}: {e}", exc_info=True)
logger.error(f"Unexpected error fetching X rate limit for {username}: {e}", exc_info=True)
return None, None
def prepare_post_data(summary, title, main_topic=None):

1
requirements.txt
Unescape View File

@ -12,3 +12,4 @@ tweepy==4.14.0
python-dotenv==1.0.1
flickr-api==0.7.1
filelock==3.16.1
requests-oauthlib==2.0.0
Write Preview
Loading…
Cancel
Save