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

fix

Browse Source
Shane 7 months ago
parent 8a2ae4ebf5
commit 96bba2398c
1 changed files with 52 additions and 40 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. 92
      foodie_engagement_tweet.py

92
foodie_engagement_tweet.py
Unescape View File

@ -6,19 +6,21 @@ import signal
import sys
import fcntl
import os
import time
from datetime import datetime, timedelta, timezone
from openai import OpenAI
from foodie_utils import post_tweet, AUTHORS, SUMMARY_MODEL, load_post_counts, save_post_counts
from foodie_config import X_API_CREDENTIALS, AUTHOR_BACKGROUNDS_FILE
from foodie_utils import post_tweet, load_post_counts, save_post_counts
from foodie_config import (
AUTHORS, SUMMARY_MODEL, X_API_CREDENTIALS, AUTHOR_BACKGROUNDS_FILE,
PERSONA_CONFIGS, ENGAGEMENT_REFERENCE_DATE_FILE
)
from dotenv import load_dotenv
from foodie_config import ENGAGEMENT_REFERENCE_DATE_FILE
load_dotenv()
REFERENCE_DATE_FILE = ENGAGEMENT_REFERENCE_DATE_FILE
LOCK_FILE = "/home/shane/foodie_automator/locks/foodie_engagement_tweet.lock"
LOG_FILE = "/home/shane/foodie_automator/logs/foodie_engagement_tweet.log"
REFERENCE_DATE_FILE = "/home/shane/foodie_automator/engagement_reference_date.json"
LOG_PRUNE_DAYS = 30
MAX_RETRIES = 3
RETRY_BACKOFF = 2
@ -116,7 +118,6 @@ def get_reference_date():
except (json.JSONDecodeError, KeyError, ValueError) as e:
logging.error(f"Failed to load reference date from {REFERENCE_DATE_FILE}: {e}. Initializing new date.")
# Initialize with current date (start of day)
reference_date = datetime.now(timezone.utc).replace(hour=0, minute=0, second=0, microsecond=0)
try:
with open(REFERENCE_DATE_FILE, 'w') as f:
@ -127,26 +128,35 @@ def get_reference_date():
return reference_date
def generate_engagement_tweet(author):
"""Generate an engagement tweet using author background themes."""
credentials = next((cred for cred in X_API_CREDENTIALS if cred["username"] == author["username"]), None)
"""Generate an engagement tweet using author background themes and persona."""
username = author["username"]
credentials = X_API_CREDENTIALS.get(username)
if not credentials:
logging.error(f"No X credentials found for {author['username']}")
logging.error(f"No X credentials found for {username}")
return None
author_handle = credentials["x_username"]
persona = author["persona"]
persona_config = PERSONA_CONFIGS.get(persona, PERSONA_CONFIGS["Visionary Editor"])
background = next((bg for bg in AUTHOR_BACKGROUNDS if bg["username"] == author["username"]), {})
background = next((bg for bg in AUTHOR_BACKGROUNDS if bg["username"] == username), {})
if not background or "engagement_themes" not in background:
logging.warning(f"No background or engagement themes found for {author['username']}")
logging.warning(f"No background or engagement themes found for {username}, using default theme")
theme = "food trends"
else:
theme = random.choice(background["engagement_themes"])
base_prompt = persona_config["x_prompt"].format(
description=persona_config["description"],
tone=persona_config["tone"]
)
prompt = (
f"Generate a concise tweet (under 280 characters) for {author_handle}. "
f"Create an engaging question or statement about {theme} to spark interaction. "
f"{base_prompt}\n\n"
f"Generate an engagement tweet for {author_handle} asking a question about {theme} to engage the public. "
f"Keep it under 280 characters, using {persona_config['tone']}. "
f"Include a call to action to follow {author_handle} or like the tweet, and mention InsiderFoodie.com with a link to https://insiderfoodie.com. "
f"Avoid using the word 'elevate'—use more humanized language like 'level up' or 'bring to life'. "
f"Do not include emojis, hashtags, or reward-driven incentives (e.g., giveaways)."
f"Do not include emojis, hashtags, or reward-driven incentives (e.g., giveaways). "
f"Return only the tweet text."
)
for attempt in range(MAX_RETRIES):
@ -163,23 +173,20 @@ def generate_engagement_tweet(author):
tweet = response.choices[0].message.content.strip()
if len(tweet) > 280:
tweet = tweet[:277] + "..."
logging.debug(f"Generated engagement tweet: {tweet}")
logging.debug(f"Generated engagement tweet for {username}: {tweet}")
return tweet
except Exception as e:
logging.warning(f"Failed to generate engagement tweet for {author['username']} (attempt {attempt + 1}): {e}")
logging.warning(f"Failed to generate engagement tweet for {username} (attempt {attempt + 1}): {e}")
if attempt < MAX_RETRIES - 1:
time.sleep(RETRY_BACKOFF * (2 ** attempt))
else:
logging.error(f"Failed to generate engagement tweet after {MAX_RETRIES} attempts")
engagement_templates = [
f"What's the most mouthwatering {theme} you've seen this week? Share below and follow {author_handle} for more on InsiderFoodie.com! Link: https://insiderfoodie.com",
f"{theme.capitalize()} lovers unite! What's your go-to pick? Tell us and like this tweet for more from {author_handle} on InsiderFoodie.com! Link: https://insiderfoodie.com",
f"Ever tried a {theme} that blew your mind? Share your favorites and follow {author_handle} for more on InsiderFoodie.com! Link: https://insiderfoodie.com",
f"What {theme} trend are you loving right now? Let us know and like this tweet to keep up with {author_handle} on InsiderFoodie.com! Link: https://insiderfoodie.com"
]
template = random.choice(engagement_templates)
logging.info(f"Using fallback engagement tweet: {template}")
return template
fallback = (
f"What's the hottest {theme} you're into? Share and follow {author_handle} for more on InsiderFoodie.com! "
f"Link: https://insiderfoodie.com"
)
logging.info(f"Using fallback engagement tweet: {fallback}")
return fallback
def post_engagement_tweet():
"""Post engagement tweets for authors every 2 days."""
@ -187,52 +194,57 @@ def post_engagement_tweet():
logging.info("Starting foodie_engagement_tweet.py")
print("Starting foodie_engagement_tweet.py")
# Get reference date
reference_date = get_reference_date()
current_date = datetime.now(timezone.utc)
days_since_reference = (current_date - reference_date).days
logging.info(f"Days since reference date ({reference_date.date()}): {days_since_reference}")
print(f"Days since reference date ({reference_date.date()}): {days_since_reference}")
# Post only if the number of days since the reference date is divisible by 2
if days_since_reference % 2 == 0:
logging.info("Today is an engagement tweet day (every 2 days). Posting...")
print("Today is an engagement tweet day (every 2 days). Posting...")
# Load post counts to check limits
post_counts = load_post_counts()
for author in AUTHORS:
username = author["username"]
try:
# Check post limits
author_count = next((entry for entry in post_counts if entry["username"] == author["username"]), None)
author_count = next((entry for entry in post_counts if entry["username"] == username), None)
if not author_count:
logging.error(f"No post count entry for {author['username']}, skipping")
continue
logging.warning(f"No post count entry for {username}, initializing new entry")
author_count = {
"username": username,
"month": datetime.now(timezone.utc).strftime("%Y-%m"),
"monthly_count": 0,
"day": datetime.now(timezone.utc).strftime("%Y-%m-%d"),
"daily_count": 0
}
post_counts.append(author_count)
save_post_counts(post_counts)
if author_count["monthly_count"] >= 500:
logging.warning(f"Monthly post limit (500) reached for {author['username']}, skipping")
logging.warning(f"Monthly post limit (500) reached for {username}, skipping")
continue
if author_count["daily_count"] >= 20:
logging.warning(f"Daily post limit (20) reached for {author['username']}, skipping")
logging.warning(f"Daily post limit (20) reached for {username}, skipping")
continue
tweet = generate_engagement_tweet(author)
if not tweet:
logging.error(f"Failed to generate engagement tweet for {author['username']}, skipping")
logging.error(f"Failed to generate engagement tweet for {username}, skipping")
continue
logging.info(f"Posting engagement tweet for {author['username']}: {tweet}")
print(f"Posting engagement tweet for {author['username']}: {tweet}")
logging.info(f"Posting engagement tweet for {username}: {tweet}")
print(f"Posting engagement tweet for {username}: {tweet}")
if post_tweet(author, tweet):
logging.info(f"Successfully posted engagement tweet for {author['username']}")
# Update post counts
logging.info(f"Successfully posted engagement tweet for {username}")
author_count["monthly_count"] += 1
author_count["daily_count"] += 1
save_post_counts(post_counts)
else:
logging.warning(f"Failed to post engagement tweet for {author['username']}")
logging.warning(f"Failed to post engagement tweet for {username}")
except Exception as e:
logging.error(f"Error posting engagement tweet for {author['username']}: {e}", exc_info=True)
logging.error(f"Error posting engagement tweet for {username}: {e}", exc_info=True)
continue
else:
logging.info(f"Today is not an engagement tweet day (every 2 days). Days since reference: {days_since_reference}. Skipping...")

Write Preview
Loading…
Cancel
Save