133 lines
5.3 KiB
Python
133 lines
5.3 KiB
Python
import json
|
|
from datetime import datetime, timedelta
|
|
import logging
|
|
import random
|
|
from openai import OpenAI # Add this import
|
|
from foodie_utils import post_tweet, AUTHORS, SUMMARY_MODEL
|
|
|
|
# Setup logging
|
|
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
|
|
|
|
# Initialize OpenAI client
|
|
client = OpenAI(api_key=os.getenv("OPENAI_API_KEY"))
|
|
|
|
RECENT_POSTS_FILE = "/home/shane/foodie_automator/recent_posts.json"
|
|
|
|
def load_recent_posts():
|
|
posts = []
|
|
if not os.path.exists(RECENT_POSTS_FILE):
|
|
return posts
|
|
|
|
with open(RECENT_POSTS_FILE, 'r') as f:
|
|
for line in f:
|
|
if line.strip():
|
|
try:
|
|
entry = json.loads(line.strip())
|
|
posts.append(entry)
|
|
except json.JSONDecodeError as e:
|
|
logging.warning(f"Skipping invalid JSON line in {RECENT_POSTS_FILE}: {e}")
|
|
|
|
return posts
|
|
|
|
def filter_posts_for_week(posts, start_date, end_date):
|
|
filtered_posts = []
|
|
for post in posts:
|
|
timestamp = datetime.fromisoformat(post["timestamp"])
|
|
if start_date <= timestamp <= end_date:
|
|
filtered_posts.append(post)
|
|
return filtered_posts
|
|
|
|
def generate_intro_tweet(author):
|
|
author_handle = author["handle"]
|
|
prompt = (
|
|
f"Generate a concise tweet (under 280 characters) for {author_handle}. "
|
|
f"Introduce a thread of their top 10 foodie posts of the week on InsiderFoodie.com. "
|
|
f"Make it engaging, create curiosity, and include a call to action to visit InsiderFoodie.com, follow {author_handle}, or like the thread. "
|
|
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)."
|
|
)
|
|
|
|
try:
|
|
response = client.chat.completions.create(
|
|
model=SUMMARY_MODEL,
|
|
messages=[
|
|
{"role": "system", "content": "You are a social media expert crafting engaging tweets."},
|
|
{"role": "user", "content": prompt}
|
|
],
|
|
max_tokens=100,
|
|
temperature=0.7
|
|
)
|
|
tweet = response.choices[0].message.content.strip()
|
|
if len(tweet) > 280:
|
|
tweet = tweet[:277] + "..."
|
|
return tweet
|
|
except Exception as e:
|
|
logging.warning(f"Failed to generate intro tweet for {author['username']}: {e}")
|
|
# Fallback template
|
|
return (
|
|
f"This weeks top 10 foodie finds by {author_handle} Check out the best on InsiderFoodie.com "
|
|
f"Follow {author_handle} for more and like this thread to stay in the loop Visit us at https://insiderfoodie.com"
|
|
)
|
|
|
|
def post_weekly_thread():
|
|
# Determine the date range (Monday to Sunday of the past week)
|
|
today = datetime.now(timezone.utc)
|
|
days_since_monday = (today.weekday() + 1) % 7 + 7 # Go back to previous Monday
|
|
start_date = (today - timedelta(days=days_since_monday)).replace(hour=0, minute=0, second=0, microsecond=0)
|
|
end_date = start_date + timedelta(days=6, hours=23, minutes=59, seconds=59)
|
|
|
|
logging.info(f"Fetching posts from {start_date} to {end_date}")
|
|
|
|
# Load and filter posts
|
|
all_posts = load_recent_posts()
|
|
weekly_posts = filter_posts_for_week(all_posts, start_date, end_date)
|
|
|
|
# Group posts by author
|
|
posts_by_author = {}
|
|
for post in weekly_posts:
|
|
author = post["author_username"] # Updated to match the key in recent_posts.json
|
|
if author not in posts_by_author:
|
|
posts_by_author[author] = []
|
|
posts_by_author[author].append(post)
|
|
|
|
# For each author, post a thread
|
|
for author in AUTHORS:
|
|
author_posts = posts_by_author.get(author["username"], [])
|
|
if not author_posts:
|
|
logging.info(f"No posts found for {author['username']} this week")
|
|
continue
|
|
|
|
# Sort by timestamp (as a proxy for interest_score) and take top 10
|
|
author_posts.sort(key=lambda x: x.get("timestamp", ""), reverse=True)
|
|
top_posts = author_posts[:10]
|
|
|
|
if not top_posts:
|
|
continue
|
|
|
|
# First tweet: Intro with CTA (generated by GPT)
|
|
intro_tweet = generate_intro_tweet(author)
|
|
|
|
logging.info(f"Posting intro tweet for {author['username']}: {intro_tweet}")
|
|
intro_response = post_tweet(author, intro_tweet)
|
|
if not intro_response:
|
|
logging.warning(f"Failed to post intro tweet for {author['username']}")
|
|
continue
|
|
|
|
intro_tweet_id = intro_response.get("id")
|
|
|
|
# Post each top post as a reply in the thread
|
|
for i, post in enumerate(top_posts, 1):
|
|
post_tweet_content = (
|
|
f"{i}. {post['title']} Link: {post['url']}"
|
|
)
|
|
logging.info(f"Posting thread reply {i} for {author['username']}: {post_tweet_content}")
|
|
post_tweet(author, post_tweet_content, reply_to_id=intro_tweet_id)
|
|
|
|
logging.info(f"Successfully posted weekly thread for {author['username']}")
|
|
|
|
if __name__ == "__main__":
|
|
# Run only on Sundays
|
|
if datetime.now(timezone.utc).weekday() == 6: # Sunday (0 = Monday, 6 = Sunday)
|
|
post_weekly_thread()
|
|
else:
|
|
logging.info("Not Sunday - skipping weekly thread posting") |