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 7950ddd0d8
commit e2fec73a72
1 changed files with 89 additions and 91 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. 180
      foodie_utils.py

180
foodie_utils.py
Unescape View File

@ -1169,96 +1169,6 @@ def select_best_author(content, interest_score):
logging.error(f"Error in select_best_author: {e}")
return random.choice([author["username"] for author in AUTHORS])
def check_author_rate_limit(author, max_tweets=17, tweet_window_seconds=86400):
"""
Check if an author can post based on their X API Free tier quota (17 tweets per 24 hours per app).
Posts a test tweet only on script restart or for new authors, then tracks tweets in rate_limit_info.json.
Returns (can_post, remaining, reset_timestamp) where can_post is True if tweets are available.
"""
logger = logging.getLogger(__name__)
rate_limit_file = '/home/shane/foodie_automator/rate_limit_info.json'
current_time = time.time()
# Load rate limit info
rate_limit_info = load_json_file(rate_limit_file, default={})
# Get script run ID
if not hasattr(check_author_rate_limit, "script_run_id"):
check_author_rate_limit.script_run_id = int(current_time)
logger.info(f"Set script_run_id to {check_author_rate_limit.script_run_id}")
username = author['username']
# Initialize or update author entry
if username not in rate_limit_info:
rate_limit_info[username] = {
'tweet_remaining': max_tweets,
'tweet_reset': current_time + tweet_window_seconds,
'tweets_posted_in_run': 0,
'script_run_id': 0 # Force test tweet for new authors
}
author_info = rate_limit_info[username]
script_run_id = author_info.get('script_run_id', 0)
# Check if quota has reset based on previous reset time
reset = author_info.get('tweet_reset', current_time + tweet_window_seconds)
if current_time >= reset:
logger.info(f"Quota reset for {username}, restoring to {max_tweets} tweets")
author_info['tweet_remaining'] = max_tweets
author_info['tweet_reset'] = current_time + tweet_window_seconds
author_info['tweets_posted_in_run'] = 0
author_info['script_run_id'] = check_author_rate_limit.script_run_id
rate_limit_info[username] = author_info
save_json_file(rate_limit_file, rate_limit_info)
# If script restarted or new author, post a test tweet to sync quota
if script_run_id != check_author_rate_limit.script_run_id:
logger.info(f"Script restart detected for {username}, posting test tweet to sync quota")
remaining, reset = get_x_rate_limit_status(author)
if remaining is None or reset is None:
# Fallback: Use last known quota or assume 0 remaining
if current_time < author_info.get('tweet_reset', 0):
remaining = author_info.get('tweet_remaining', 0)
reset = author_info.get('tweet_reset', current_time + tweet_window_seconds)
logger.warning(f"Test tweet failed for {username}, using last known quota: {remaining} remaining")
else:
remaining = max_tweets
reset = current_time + tweet_window_seconds
logger.warning(f"Test tweet failed for {username}, resetting quota to {max_tweets}")
else:
# Only update remaining if the API reset time is newer than the stored reset
if reset > author_info.get('tweet_reset', 0):
logger.info(f"Updating quota for {username} from API: {remaining} remaining, reset at {datetime.fromtimestamp(reset, tz=timezone.utc)}")
author_info['tweet_remaining'] = remaining
else:
logger.info(f"Retaining previous quota for {username}: {author_info['tweet_remaining']} remaining")
remaining = author_info['tweet_remaining']
# Ensure reset is in the future
if reset <= current_time:
logger.warning(f"Reset time {reset} is in the past for {username}. Setting to 24 hours from now.")
reset = current_time + tweet_window_seconds
# Update author info
author_info['tweet_reset'] = reset
author_info['tweets_posted_in_run'] = 0
author_info['script_run_id'] = check_author_rate_limit.script_run_id
rate_limit_info[username] = author_info
save_json_file(rate_limit_file, rate_limit_info)
# Calculate remaining tweets
remaining = author_info['tweet_remaining'] - author_info['tweets_posted_in_run']
can_post = remaining > 0
if not can_post:
reset_time = datetime.fromtimestamp(reset, tz=timezone.utc).strftime('%Y-%m-%d %H:%M:%S')
logger.info(f"Author {username} quota exhausted. Remaining: {remaining}, Reset at: {reset_time}")
else:
logger.info(f"Quota for {username}: {remaining}/{max_tweets} tweets remaining")
return can_post, remaining, reset
def get_next_author_round_robin():
"""
Select the next author using round-robin, respecting real-time X API rate limits.
@ -1340,7 +1250,7 @@ def get_x_rate_limit_status(author):
# Ensure reset is in the future
current_time = int(time.time())
if reset <= current_time or reset > current_time + 2 * 86400: # Allow up to 48 hours
logger.warning(f"Invalid reset time {reset} for {username}. Setting to 24 hours from now.")
logger.warning(f"Invalid reset time {reset} ({datetime.fromtimestamp(reset, tz=timezone.utc)}) for {username}. Setting to 24 hours from now.")
reset = current_time + 86400 # 24 hours
if response.status_code == 201:
@ -1368,6 +1278,94 @@ def get_x_rate_limit_status(author):
except Exception as e:
logger.error(f"Unexpected error fetching X rate limit for {username}: {e}", exc_info=True)
return None, None
def check_author_rate_limit(author, max_tweets=17, tweet_window_seconds=86400):
"""
Check if an author can post based on their X API Free tier quota (17 tweets per 24 hours per app).
Posts a test tweet only on script restart or for new authors, then tracks tweets in rate_limit_info.json.
Returns (can_post, remaining, reset_timestamp) where can_post is True if tweets are available.
"""
logger = logging.getLogger(__name__)
rate_limit_file = '/home/shane/foodie_automator/rate_limit_info.json'
current_time = time.time()
# Load rate limit info
rate_limit_info = load_json_file(rate_limit_file, default={})
# Get script run ID
if not hasattr(check_author_rate_limit, "script_run_id"):
check_author_rate_limit.script_run_id = int(current_time)
logger.info(f"Set script_run_id to {check_author_rate_limit.script_run_id}")
username = author['username']
# Initialize or update author entry
if username not in rate_limit_info:
rate_limit_info[username] = {
'tweet_remaining': max_tweets,
'tweet_reset': current_time + tweet_window_seconds,
'tweets_posted_in_run': 0,
'script_run_id': 0 # Force test tweet for new authors
}
author_info = rate_limit_info[username]
script_run_id = author_info.get('script_run_id', 0)
# Check if quota has reset based on previous reset time
reset = author_info.get('tweet_reset', current_time + tweet_window_seconds)
if current_time >= reset:
logger.info(f"Quota reset for {username}, restoring to {max_tweets} tweets")
author_info['tweet_remaining'] = max_tweets
author_info['tweet_reset'] = current_time + tweet_window_seconds
author_info['tweets_posted_in_run'] = 0
author_info['script_run_id'] = check_author_rate_limit.script_run_id
rate_limit_info[username] = author_info
save_json_file(rate_limit_file, rate_limit_info)
# If script restarted or new author, post a test tweet to sync quota
if script_run_id != check_author_rate_limit.script_run_id:
logger.info(f"Script restart detected for {username}, posting test tweet to sync quota")
remaining, api_reset = get_x_rate_limit_status(author)
if remaining is None or api_reset is None:
# Fallback: Use last known quota or assume 0 remaining
if current_time < author_info.get('tweet_reset', 0):
remaining = author_info.get('tweet_remaining', 0)
reset = author_info.get('tweet_reset', current_time + tweet_window_seconds)
logger.warning(f"Test tweet failed for {username}, using last known quota: {remaining} remaining")
else:
remaining = max_tweets
reset = current_time + tweet_window_seconds
logger.warning(f"Test tweet failed for {username}, resetting quota to {max_tweets}")
else:
# Only update remaining if the API reset time is newer and valid
if api_reset > author_info.get('tweet_reset', 0) and api_reset > current_time:
logger.info(f"Updating quota for {username} from API: {remaining} remaining, reset at {datetime.fromtimestamp(api_reset, tz=timezone.utc)}")
author_info['tweet_remaining'] = remaining
author_info['tweet_reset'] = api_reset
else:
logger.info(f"Retaining previous quota for {username}: {author_info['tweet_remaining']} remaining")
remaining = author_info['tweet_remaining']
# Keep the existing reset time (from quota reset or previous state)
reset = author_info['tweet_reset']
# Update author info
author_info['tweets_posted_in_run'] = 0
author_info['script_run_id'] = check_author_rate_limit.script_run_id
author_info['tweet_remaining'] = remaining
author_info['tweet_reset'] = reset
rate_limit_info[username] = author_info
save_json_file(rate_limit_file, rate_limit_info)
# Calculate remaining tweets
remaining = author_info['tweet_remaining'] - author_info['tweets_posted_in_run']
can_post = remaining > 0
if not can_post:
reset_time = datetime.fromtimestamp(reset, tz=timezone.utc).strftime('%Y-%m-%d %H:%M:%S')
logger.info(f"Author {username} quota exhausted. Remaining: {remaining}, Reset at: {reset_time}")
else:
logger.info(f"Quota for {username}: {remaining}/{max_tweets} tweets remaining")
return can_post, remaining, reset
def prepare_post_data(summary, title, main_topic=None):
try:

Write Preview
Loading…
Cancel
Save