merge posting x into main files
This commit is contained in:
+23
-145
@@ -6,12 +6,10 @@ import time
|
||||
import sys
|
||||
import signal
|
||||
import os
|
||||
from datetime import datetime, timedelta, timezone
|
||||
from datetime import datetime, timezone
|
||||
from openai import OpenAI
|
||||
import tweepy
|
||||
from foodie_config import OPENAI_API_KEY, AUTHORS, LIGHT_TASK_MODEL
|
||||
from foodie_utils import load_json_file
|
||||
from foodie_x_config import X_API_CREDENTIALS, X_PERSONA_PROMPTS, AUTHOR_BACKGROUNDS_FILE, X_POST_COUNTS_FILE, RECENT_POSTS_FILE
|
||||
from foodie_config import OPENAI_API_KEY, AUTHORS, LIGHT_TASK_MODEL, PERSONA_CONFIGS, AUTHOR_BACKGROUNDS_FILE
|
||||
from foodie_utils import load_json_file, post_tweet
|
||||
from dotenv import load_dotenv
|
||||
|
||||
load_dotenv()
|
||||
@@ -49,24 +47,6 @@ except Exception as e:
|
||||
logging.error(f"Failed to load author_backgrounds.json: {e}")
|
||||
sys.exit(1)
|
||||
|
||||
def load_post_counts():
|
||||
counts = load_json_file(X_POST_COUNTS_FILE)
|
||||
if not counts:
|
||||
counts = [{"username": author["username"], "count": 0, "month": datetime.now(timezone.utc).strftime("%Y-%m")} for author in AUTHORS]
|
||||
current_month = datetime.now(timezone.utc).strftime("%Y-%m")
|
||||
for entry in counts:
|
||||
if entry["month"] != current_month:
|
||||
entry["count"] = 0
|
||||
entry["month"] = current_month
|
||||
return counts
|
||||
|
||||
def save_post_counts(counts):
|
||||
with open(X_POST_COUNTS_FILE, 'w') as f:
|
||||
for item in counts:
|
||||
json.dump(item, f)
|
||||
f.write('\n')
|
||||
logging.info(f"Saved post counts to {X_POST_COUNTS_FILE}")
|
||||
|
||||
is_posting = False
|
||||
|
||||
def signal_handler(sig, frame):
|
||||
@@ -80,36 +60,28 @@ def signal_handler(sig, frame):
|
||||
signal.signal(signal.SIGTERM, signal_handler)
|
||||
signal.signal(signal.SIGINT, signal_handler)
|
||||
|
||||
def get_recent_posts_for_author(username):
|
||||
posts = load_json_file(RECENT_POSTS_FILE)
|
||||
return [post for post in posts if post["author_username"] == username]
|
||||
|
||||
def delete_used_post(post_title):
|
||||
posts = load_json_file(RECENT_POSTS_FILE)
|
||||
posts = [post for post in posts if post["title"] != post_title]
|
||||
with open(RECENT_POSTS_FILE, 'w') as f:
|
||||
for item in posts:
|
||||
json.dump(item, f)
|
||||
f.write('\n')
|
||||
logging.info(f"Deleted post '{post_title}' from recent_posts.json")
|
||||
|
||||
def generate_article_tweet(author, post, persona):
|
||||
# Format the prompt using description and tone
|
||||
persona_config = X_PERSONA_PROMPTS[persona]
|
||||
base_prompt = persona_config["prompt"].format(
|
||||
def generate_engagement_tweet(author, persona):
|
||||
background = next((bg for bg in AUTHOR_BACKGROUNDS if bg["username"] == author["username"]), {})
|
||||
if not background or "engagement_themes" not in background:
|
||||
logging.warning(f"No background or engagement themes found for {author['username']}")
|
||||
return "What food trends are you loving right now? Share your thoughts! #FoodieTrends"
|
||||
|
||||
theme = random.choice(background["engagement_themes"])
|
||||
persona_config = PERSONA_CONFIGS[persona]
|
||||
base_prompt = persona_config["x_prompt"].format(
|
||||
description=persona_config["description"],
|
||||
tone=persona_config["tone"]
|
||||
)
|
||||
prompt = base_prompt.replace(
|
||||
"For article tweets, include the article title, a quirky hook, and the URL.",
|
||||
f"Generate an article tweet including the title '{post['title']}', a quirky hook, and the URL '{post['url']}'."
|
||||
"For engagement tweets, ask a question about food trends, foods, or articles to engage the public.",
|
||||
f"Generate an engagement tweet asking a question about {theme} to engage the public."
|
||||
)
|
||||
try:
|
||||
response = client.chat.completions.create(
|
||||
model=LIGHT_TASK_MODEL,
|
||||
messages=[
|
||||
{"role": "system", "content": prompt},
|
||||
{"role": "user", "content": f"Generate tweet for {post['title']}."}
|
||||
{"role": "user", "content": f"Generate engagement tweet for {author['username']} about {theme}."}
|
||||
],
|
||||
max_tokens=100,
|
||||
temperature=0.9
|
||||
@@ -117,115 +89,21 @@ def generate_article_tweet(author, post, persona):
|
||||
tweet = response.choices[0].message.content.strip()
|
||||
if len(tweet) > 280:
|
||||
tweet = tweet[:277] + "..."
|
||||
logging.info(f"Generated article tweet for {author['username']}: {tweet}")
|
||||
logging.info(f"Generated engagement tweet for {author['username']}: {tweet}")
|
||||
return tweet
|
||||
except Exception as e:
|
||||
logging.error(f"Failed to generate article tweet for {author['username']}: {e}")
|
||||
return f"This trend is fire! Check out {post['title']} at {post['url']} #Foodie"
|
||||
|
||||
def generate_personal_tweet(author, persona):
|
||||
background = next((bg for bg in AUTHOR_BACKGROUNDS if bg["username"] == author["username"]), {})
|
||||
if not background:
|
||||
logging.warning(f"No background found for {author['username']}")
|
||||
return f"Loving my gig at InsiderFoodie, dishing out food trends! #FoodieLife"
|
||||
|
||||
# Get DOB and calculate age
|
||||
dob = author.get('dob', '1980-01-01')
|
||||
current_year = datetime.now().year
|
||||
birth_year = int(dob.split('-')[0])
|
||||
age = current_year - birth_year
|
||||
|
||||
is_role_reflection = random.choice([True, False])
|
||||
if is_role_reflection:
|
||||
content = f"Reflect on your role at InsiderFoodie as {author['persona']}. Mention you're {age} years old."
|
||||
else:
|
||||
content = (
|
||||
f"Share a personal story about your background, considering you were born on {dob} and are {age} years old. "
|
||||
f"Hometown: {background['hometown']}, Cultural influences: {background['cultural_influences']}, "
|
||||
f"Early memory: {background['early_memory']}, Career path: {background['career_path']}."
|
||||
)
|
||||
|
||||
# Format the prompt using description and tone
|
||||
persona_config = X_PERSONA_PROMPTS[persona]
|
||||
base_prompt = persona_config["prompt"].format(
|
||||
description=persona_config["description"],
|
||||
tone=persona_config["tone"]
|
||||
)
|
||||
prompt = base_prompt.replace(
|
||||
"For personal tweets, reflect on your role at InsiderFoodie or background.",
|
||||
content
|
||||
)
|
||||
try:
|
||||
response = client.chat.completions.create(
|
||||
model="gpt-4o-mini",
|
||||
messages=[
|
||||
{"role": "system", "content": prompt},
|
||||
{"role": "user", "content": f"Generate personal tweet for {author['username']}."}
|
||||
],
|
||||
max_tokens=100,
|
||||
temperature=0.9
|
||||
)
|
||||
tweet = response.choices[0].message.content.strip()
|
||||
if len(tweet) > 280:
|
||||
tweet = tweet[:277] + "..."
|
||||
logging.info(f"Generated personal tweet for {author['username']}: {tweet}")
|
||||
return tweet
|
||||
except Exception as e:
|
||||
logging.error(f"Failed to generate personal tweet for {author['username']}: {e}")
|
||||
return f"Loving my gig at InsiderFoodie, dishing out food trends! #FoodieLife"
|
||||
|
||||
def post_tweet(author, tweet):
|
||||
global is_posting
|
||||
credentials = next((cred for cred in X_API_CREDENTIALS if cred["username"] == author["username"]), None)
|
||||
if not credentials:
|
||||
logging.error(f"No X credentials found for {author['username']}")
|
||||
return False
|
||||
|
||||
post_counts = load_post_counts()
|
||||
author_count = next((entry for entry in post_counts if entry["username"] == author["username"]), None)
|
||||
if author_count["count"] >= 450:
|
||||
logging.warning(f"Post limit reached for {author['username']} this month")
|
||||
return False
|
||||
|
||||
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"]
|
||||
)
|
||||
is_posting = True
|
||||
response = client.create_tweet(text=tweet)
|
||||
is_posting = False
|
||||
author_count["count"] += 1
|
||||
save_post_counts(post_counts)
|
||||
logging.info(f"Posted tweet for {author['username']}: {tweet}")
|
||||
return True
|
||||
except Exception as e:
|
||||
is_posting = False
|
||||
logging.error(f"Failed to post tweet for {author['username']}: {e}")
|
||||
return False
|
||||
logging.error(f"Failed to generate engagement tweet for {author['username']}: {e}")
|
||||
return f"What’s your take on {theme}? Let’s talk! #FoodieTrends"
|
||||
|
||||
def main():
|
||||
global is_posting
|
||||
logging.info("***** X Poster Launched *****")
|
||||
for author in AUTHORS:
|
||||
posts = get_recent_posts_for_author(author["username"])
|
||||
if not posts:
|
||||
logging.info(f"No recent posts for {author['username']}, skipping")
|
||||
continue
|
||||
|
||||
article_tweets = 0
|
||||
for post in posts[:2]:
|
||||
tweet = generate_article_tweet(author, post, author["persona"])
|
||||
if post_tweet(author, tweet):
|
||||
delete_used_post(post["title"])
|
||||
article_tweets += 1
|
||||
time.sleep(random.uniform(3600, 7200))
|
||||
if article_tweets >= 2:
|
||||
break
|
||||
|
||||
tweet = generate_personal_tweet(author, author["persona"])
|
||||
is_posting = True
|
||||
tweet = generate_engagement_tweet(author, author["persona"])
|
||||
post_tweet(author, tweet)
|
||||
is_posting = False
|
||||
time.sleep(random.uniform(3600, 7200))
|
||||
|
||||
logging.info("X posting completed")
|
||||
return random.randint(600, 1800)
|
||||
|
||||
Reference in New Issue
Block a user