test

foodie_weekly_thread.py

@@ -7,6 +7,7 @@ from openai import OpenAI
 from foodie_utils import post_tweet, AUTHORS, SUMMARY_MODEL
 from foodie_config import X_API_CREDENTIALS
 from dotenv import load_dotenv
+import tweepy
 
 load_dotenv()
 
@@ -15,7 +16,6 @@ LOG_FILE = "/home/shane/foodie_automator/foodie_weekly_thread.log"
 LOG_PRUNE_DAYS = 30
 
 def setup_logging():
-    # Prune old logs
     if os.path.exists(LOG_FILE):
         with open(LOG_FILE, 'r') as f:
             lines = f.readlines()
@@ -31,10 +31,9 @@ def setup_logging():
         with open(LOG_FILE, 'w') as f:
             f.writelines(pruned_lines)
     
-    # Set up logging to file and console
     logging.basicConfig(
         filename=LOG_FILE,
-        level=logging.DEBUG,  # Set to DEBUG for detailed output
+        level=logging.DEBUG,
         format='%(asctime)s - %(levelname)s - %(message)s',
         datefmt='%Y-%m-%d %H:%M:%S'
     )
@@ -51,19 +50,49 @@ if not os.getenv("OPENAI_API_KEY"):
     logging.error("OPENAI_API_KEY is not set in environment variables")
     raise ValueError("OPENAI_API_KEY is required")
 
-# Validate X_API_CREDENTIALS
+# Validate X_API_CREDENTIALS and test API access
-if not X_API_CREDENTIALS:
+def validate_twitter_credentials():
-    logging.error("X_API_CREDENTIALS is empty in foodie_config.py")
+    logging.info("Validating Twitter API credentials for all authors")
-    raise ValueError("X_API_CREDENTIALS is required")
+    valid_credentials = []
+    for author in AUTHORS:
+        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']} in X_API_CREDENTIALS")
+            print(f"No X credentials found for {author['username']}")
+            continue
+        logging.debug(f"Testing credentials for {author['username']} (handle: {credentials['x_username']})")
+        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"]
+            )
+            # Test API access by fetching the user's profile
+            user = client.get_me()
+            logging.info(f"Credentials valid for {author['username']} (handle: {credentials['x_username']}, user_id: {user.data.id})")
+            print(f"Credentials valid for {author['username']} (handle: {credentials['x_username']})")
+            valid_credentials.append(credentials)
+        except tweepy.TweepyException as e:
+            logging.error(f"Failed to validate credentials for {author['username']} (handle: {credentials['x_username']}): {e}")
+            if hasattr(e, 'response') and e.response:
+                logging.error(f"Twitter API response: {e.response.text}")
+            print(f"Failed to validate credentials for {author['username']}: {e}")
+    if not valid_credentials:
+        logging.error("No valid Twitter credentials found for any author")
+        raise ValueError("No valid Twitter credentials found")
+    return valid_credentials
+
+# Run credential validation
+validate_twitter_credentials()
 
 RECENT_POSTS_FILE = "/home/shane/foodie_automator/recent_posts.json"
 
 def load_recent_posts():
     posts = []
-    unique_posts = {}  # To track unique posts by title, URL, and author
+    unique_posts = {}
     logging.debug(f"Attempting to load posts from {RECENT_POSTS_FILE}")
     
-    # Check if file exists and is readable
     if not os.path.exists(RECENT_POSTS_FILE):
         logging.error(f"Recent posts file {RECENT_POSTS_FILE} does not exist")
         return posts
@@ -82,18 +111,15 @@ def load_recent_posts():
                 continue
             try:
                 entry = json.loads(line.strip())
-                # Validate required fields
                 required_fields = ["title", "url", "author_username", "timestamp"]
                 if not all(key in entry for key in required_fields):
                     logging.warning(f"Skipping invalid entry at line {i}: missing fields {entry}")
                     continue
-                # Validate timestamp format
                 try:
                     datetime.fromisoformat(entry["timestamp"])
                 except ValueError:
                     logging.warning(f"Skipping entry at line {i}: invalid timestamp {entry['timestamp']}")
                     continue
-                # Deduplicate based on title, URL, and author
                 key = (entry["title"], entry["url"], entry["author_username"])
                 if key in unique_posts:
                     logging.debug(f"Skipping duplicate entry at line {i}: {entry['title']}")
@@ -173,8 +199,7 @@ def post_weekly_thread():
     print("Entering post_weekly_thread")
     
     today = datetime.now(timezone.utc)
-    # Fix week calculation to target the previous week (Monday to Sunday)
+    days_to_monday = today.weekday()
-    days_to_monday = today.weekday()  # 0 for Monday, 1 for Tuesday, etc.
     start_date = (today - timedelta(days=days_to_monday + 7)).replace(hour=0, minute=0, second=0, microsecond=0)
     end_date = start_date + timedelta(days=6, hours=23, minutes=59, seconds=59)
     
@@ -231,7 +256,7 @@ def post_weekly_thread():
         
         intro_response = post_tweet(author, intro_tweet)
         if not intro_response:
-            logging.error(f"Failed to post intro tweet for {author['username']}")
+            logging.error(f"Failed to post intro tweet for {author['username']}, skipping thread")
             print(f"Failed to post intro tweet for {author['username']}")
             continue