Update sync_logic.py

sync_logic.py

@@ -1,15 +1,15 @@
 # sync_logic.py
 """
 Handles the logic for syncing LinkedIn data: posts, mentions, and follower statistics.
-Fetches data from LinkedIn APIs and uploads to Bubble.
+Fetches data from LinkedIn APIs, uploads to Bubble, and logs sync attempts.
 """
 import pandas as pd
 import logging
 import html
+from datetime import timezone # Python's datetime
 
 # Assuming Bubble_API_Calls contains bulk_upload_to_bubble
-from Bubble_API_Calls import bulk_upload_to_bubble
-
+from Bubble_API_Calls import bulk_upload_to_bubble, fetch_linkedin_posts_data_from_bubble # Added fetch for log update
 # Assuming Linkedin_Data_API_Calls contains all necessary LinkedIn data fetching and processing functions
 from Linkedin_Data_API_Calls import (
     fetch_linkedin_posts_core,
@@ -33,25 +33,98 @@ from config import (
     DEFAULT_MENTIONS_INITIAL_FETCH_COUNT, DEFAULT_MENTIONS_UPDATE_FETCH_COUNT,
     BUBBLE_FOLLOWER_STATS_TABLE_NAME, FOLLOWER_STATS_TYPE_COLUMN, FOLLOWER_STATS_CATEGORY_COLUMN,
     FOLLOWER_STATS_ORG_URN_COLUMN, FOLLOWER_STATS_ORGANIC_COLUMN, FOLLOWER_STATS_PAID_COLUMN,
-    LINKEDIN_CLIENT_ID_ENV_VAR
+    LINKEDIN_CLIENT_ID_ENV_VAR, # Though client_id is usually passed in token_state
+    # NEW constants for logging
+    BUBBLE_OPERATIONS_LOG_TABLE_NAME, BUBBLE_OPERATIONS_LOG_DATE_COLUMN,
+    BUBBLE_OPERATIONS_LOG_SUBJECT_COLUMN, BUBBLE_OPERATIONS_LOG_ORG_URN_COLUMN,
+    LOG_SUBJECT_POSTS, LOG_SUBJECT_MENTIONS, LOG_SUBJECT_FOLLOWER_STATS
 )
 
+def _log_sync_attempt(org_urn, subject, token_state):
+    """
+    Logs a sync attempt to the Bubble operations log table and updates
+    the operations log DataFrame in token_state.
+    """
+    logging.info(f"Logging sync attempt for subject: {subject}, org_urn: {org_urn}")
+    if not org_urn:
+        logging.warning("Cannot log sync attempt: org_urn is missing.")
+        return token_state
+    try:
+        log_entry_data = {
+            BUBBLE_OPERATIONS_LOG_DATE_COLUMN: pd.Timestamp.now(tz='UTC').isoformat(),
+            BUBBLE_OPERATIONS_LOG_SUBJECT_COLUMN: subject,
+            BUBBLE_OPERATIONS_LOG_ORG_URN_COLUMN: org_urn
+        }
+        
+        # Ensure data types are what Bubble expects, e.g., date as string
+        # bulk_upload_to_bubble should handle dicts with basic types.
+        upload_payload = [log_entry_data]
+
+        bulk_upload_to_bubble(upload_payload, BUBBLE_OPERATIONS_LOG_TABLE_NAME)
+        logging.info(f"Successfully logged sync attempt for {subject} to Bubble table '{BUBBLE_OPERATIONS_LOG_TABLE_NAME}'.")
+
+        # Update token_state with the new log entry to keep it fresh
+        current_log_df = token_state.get("bubble_operations_log_df", pd.DataFrame())
+        new_log_entry_df = pd.DataFrame(upload_payload) # DataFrame from the same data we uploaded
+
+        # Ensure date column is datetime before concat if it exists and is not empty
+        if not new_log_entry_df.empty and BUBBLE_OPERATIONS_LOG_DATE_COLUMN in new_log_entry_df.columns:
+             new_log_entry_df[BUBBLE_OPERATIONS_LOG_DATE_COLUMN] = pd.to_datetime(new_log_entry_df[BUBBLE_OPERATIONS_LOG_DATE_COLUMN], errors='coerce', utc=True)
+        
+        if not current_log_df.empty and BUBBLE_OPERATIONS_LOG_DATE_COLUMN in current_log_df.columns:
+            # Ensure existing log df date column is also datetime
+            if not pd.api.types.is_datetime64_any_dtype(current_log_df[BUBBLE_OPERATIONS_LOG_DATE_COLUMN]):
+                current_log_df[BUBBLE_OPERATIONS_LOG_DATE_COLUMN] = pd.to_datetime(current_log_df[BUBBLE_OPERATIONS_LOG_DATE_COLUMN], errors='coerce', utc=True)
+        
+        updated_log_df = pd.concat([current_log_df, new_log_entry_df], ignore_index=True)
+        # To ensure the get_last_sync_attempt_date always gets the absolute latest,
+        # we can sort and drop duplicates, keeping the last.
+        # However, simply appending and letting max() find the latest is also fine.
+        # For robustness, let's sort and keep the latest for each subject/org combo if multiple logs were made rapidly.
+        if not updated_log_df.empty and all(col in updated_log_df.columns for col in [BUBBLE_OPERATIONS_LOG_DATE_COLUMN, BUBBLE_OPERATIONS_LOG_SUBJECT_COLUMN, BUBBLE_OPERATIONS_LOG_ORG_URN_COLUMN]):
+            updated_log_df = updated_log_df.sort_values(by=BUBBLE_OPERATIONS_LOG_DATE_COLUMN).drop_duplicates(
+                subset=[BUBBLE_OPERATIONS_LOG_SUBJECT_COLUMN, BUBBLE_OPERATIONS_LOG_ORG_URN_COLUMN],
+                keep='last'
+            )
+        token_state["bubble_operations_log_df"] = updated_log_df
+        logging.info(f"Updated 'bubble_operations_log_df' in token_state after logging {subject}.")
+
+    except Exception as e:
+        logging.error(f"Failed to log sync attempt for {subject} or update token_state: {e}", exc_info=True)
+    return token_state
+
 
 def _sync_linkedin_posts_internal(token_state, fetch_count_for_posts_api):
     """Internal logic for syncing LinkedIn posts."""
+    # This function is called by orchestrator only if fetch_count_for_posts_api > 0
+    # So, an attempt to sync posts is indeed happening.
+
     logging.info(f"Posts sync: Starting fetch for {fetch_count_for_posts_api} posts.")
     client_id = token_state.get("client_id")
     token_dict = token_state.get("token")
     org_urn = token_state.get('org_urn')
     bubble_posts_df_orig = token_state.get("bubble_posts_df", pd.DataFrame()).copy()
     posts_sync_message = ""
+    attempt_logged = False # Flag to ensure log happens once
 
     try:
+        # Basic checks before API call
+        if not all([client_id, token_dict, org_urn]):
+            posts_sync_message = "Posts: Config error (client_id, token, or org_urn missing). "
+            logging.error(f"Posts sync: Prerequisite missing - client_id: {'OK' if client_id else 'Missing'}, token: {'OK' if token_dict else 'Missing'}, org_urn: {'OK' if org_urn else 'Missing'}")
+            # Log attempt even if config error, as state_manager decided a sync *should* occur
+            token_state = _log_sync_attempt(org_urn, LOG_SUBJECT_POSTS, token_state)
+            attempt_logged = True
+            return posts_sync_message, token_state
+
         processed_raw_posts, stats_map, _ = fetch_linkedin_posts_core(client_id, token_dict, org_urn, count=fetch_count_for_posts_api)
 
         if not processed_raw_posts:
             posts_sync_message = "Posts: None found via API. "
             logging.info("Posts sync: No raw posts returned from API.")
+            # Log attempt as API was called
+            token_state = _log_sync_attempt(org_urn, LOG_SUBJECT_POSTS, token_state)
+            attempt_logged = True
             return posts_sync_message, token_state
 
         existing_post_urns = set()
@@ -63,15 +136,17 @@ def _sync_linkedin_posts_internal(token_state, fetch_count_for_posts_api):
         if not new_raw_posts:
             posts_sync_message = "Posts: All fetched already in Bubble. "
             logging.info("Posts sync: All fetched posts were already found in Bubble.")
+            # Log attempt as API was called and processed
+            token_state = _log_sync_attempt(org_urn, LOG_SUBJECT_POSTS, token_state)
+            attempt_logged = True
             return posts_sync_message, token_state
 
         logging.info(f"Posts sync: Processing {len(new_raw_posts)} new raw posts.")
         post_urns_to_process = [p[LINKEDIN_POST_URN_KEY] for p in new_raw_posts if p.get(LINKEDIN_POST_URN_KEY)]
 
         all_comments_data = fetch_comments(client_id, token_dict, post_urns_to_process, stats_map)
-        sentiments_per_post = analyze_sentiment(all_comments_data) # Assumes analysis of comments for posts
+        sentiments_per_post = analyze_sentiment(all_comments_data)
         detailed_new_posts = compile_detailed_posts(new_raw_posts, stats_map, sentiments_per_post)
-
         li_posts, li_post_stats, li_post_comments = prepare_data_for_bubble(detailed_new_posts, all_comments_data)
 
         if li_posts:
@@ -97,59 +172,59 @@ def _sync_linkedin_posts_internal(token_state, fetch_count_for_posts_api):
     except Exception as e:
         logging.exception("Posts sync: Unexpected error during processing.")
         posts_sync_message = f"Posts: Unexpected error ({type(e).__name__}). "
+    finally:
+        # Log the sync attempt if it hasn't been logged already (e.g. due to early exit)
+        # and if basic conditions (org_urn) for logging are met.
+        if not attempt_logged and org_urn:
+            token_state = _log_sync_attempt(org_urn, LOG_SUBJECT_POSTS, token_state)
     return posts_sync_message, token_state
 
 
 def sync_linkedin_mentions(token_state):
-    """Fetches new LinkedIn mentions and uploads them to Bubble."""
-    logging.info("Starting LinkedIn mentions sync process.")
+    """Fetches new LinkedIn mentions and uploads them to Bubble, if scheduled by state_manager."""
+    logging.info("Starting LinkedIn mentions sync process check.")
+    
+    if not token_state.get("mentions_should_sync_now", False):
+        logging.info("Mentions sync: Not scheduled by state_manager based on operations log. Skipping.")
+        return "Mentions: Sync not currently required by schedule. ", token_state
+
+    logging.info("Mentions sync: Proceeding as scheduled by state_manager.")
+    
     if not token_state or not token_state.get("token"):
         logging.error("Mentions sync: Access denied. No LinkedIn token.")
+        # Still log an attempt if org_urn is available, as a sync was scheduled
+        org_urn_for_log = token_state.get('org_urn') if token_state else None
+        if org_urn_for_log:
+             token_state = _log_sync_attempt(org_urn_for_log, LOG_SUBJECT_MENTIONS, token_state)
         return "Mentions: No token. ", token_state
 
     client_id = token_state.get("client_id")
     token_dict = token_state.get("token")
     org_urn = token_state.get('org_urn')
-    # Work with a copy, original df in token_state will be updated at the end
     bubble_mentions_df_orig = token_state.get("bubble_mentions_df", pd.DataFrame()).copy()
+    mentions_sync_message = ""
+    attempt_logged = False
 
     if not org_urn or not client_id or client_id == "ENV VAR MISSING":
         logging.error("Mentions sync: Configuration error (Org URN or Client ID missing).")
+        if org_urn: # Log if possible
+            token_state = _log_sync_attempt(org_urn, LOG_SUBJECT_MENTIONS, token_state)
+            attempt_logged = True
         return "Mentions: Config error. ", token_state
 
-    fetch_count_for_mentions_api = 0
-    mentions_sync_is_needed_now = False
-    if bubble_mentions_df_orig.empty:
-        mentions_sync_is_needed_now = True
-        fetch_count_for_mentions_api = DEFAULT_MENTIONS_INITIAL_FETCH_COUNT
-        logging.info("Mentions sync needed: Bubble DF empty. Fetching initial count.")
-    else:
-        if BUBBLE_MENTIONS_DATE_COLUMN_NAME not in bubble_mentions_df_orig.columns or \
-           bubble_mentions_df_orig[BUBBLE_MENTIONS_DATE_COLUMN_NAME].isnull().all():
-            mentions_sync_is_needed_now = True
-            fetch_count_for_mentions_api = DEFAULT_MENTIONS_INITIAL_FETCH_COUNT
-            logging.info(f"Mentions sync needed: Date column '{BUBBLE_MENTIONS_DATE_COLUMN_NAME}' missing or all null. Fetching initial count.")
-        else:
-            # Use a copy for date checks to avoid SettingWithCopyWarning if any modification were made
-            mentions_df_check = bubble_mentions_df_orig.copy()
-            mentions_df_check[BUBBLE_MENTIONS_DATE_COLUMN_NAME] = pd.to_datetime(mentions_df_check[BUBBLE_MENTIONS_DATE_COLUMN_NAME], errors='coerce', utc=True)
-            last_mention_date_utc = mentions_df_check[BUBBLE_MENTIONS_DATE_COLUMN_NAME].dropna().max()
-            if pd.isna(last_mention_date_utc) or \
-               (pd.Timestamp('now', tz='UTC').normalize() - last_mention_date_utc.normalize()).days >= 7:
-                mentions_sync_is_needed_now = True
-                fetch_count_for_mentions_api = DEFAULT_MENTIONS_UPDATE_FETCH_COUNT
-                logging.info(f"Mentions sync needed: Last mention date {last_mention_date_utc} is old or invalid. Fetching update count.")
-
-    if not mentions_sync_is_needed_now:
-        logging.info("Mentions data is fresh based on current check. No API fetch needed for mentions.")
-        return "Mentions: Up-to-date. ", token_state
-
-    logging.info(f"Mentions sync proceeding. Fetch count: {fetch_count_for_mentions_api}")
+    # Determine fetch count: initial if no mentions data, update otherwise
+    fetch_count_for_mentions_api = DEFAULT_MENTIONS_INITIAL_FETCH_COUNT \
+        if bubble_mentions_df_orig.empty else DEFAULT_MENTIONS_UPDATE_FETCH_COUNT
+    logging.info(f"Mentions sync: Fetch count set to {fetch_count_for_mentions_api}.")
+
     try:
         processed_raw_mentions = fetch_linkedin_mentions_core(client_id, token_dict, org_urn, count=fetch_count_for_mentions_api)
         if not processed_raw_mentions:
             logging.info("Mentions sync: No new mentions found via API.")
-            return "Mentions: None found via API. ", token_state
+            mentions_sync_message = "Mentions: None found via API. "
+            token_state = _log_sync_attempt(org_urn, LOG_SUBJECT_MENTIONS, token_state)
+            attempt_logged = True
+            return mentions_sync_message, token_state
 
         existing_mention_ids = set()
         if not bubble_mentions_df_orig.empty and BUBBLE_MENTIONS_ID_COLUMN_NAME in bubble_mentions_df_orig.columns:
@@ -164,7 +239,10 @@ def sync_linkedin_mentions(token_state):
 
         if not new_compiled_mentions_to_upload:
             logging.info("Mentions sync: All fetched mentions are already in Bubble.")
-            return "Mentions: All fetched already in Bubble. ", token_state
+            mentions_sync_message = "Mentions: All fetched already in Bubble. "
+            token_state = _log_sync_attempt(org_urn, LOG_SUBJECT_MENTIONS, token_state)
+            attempt_logged = True
+            return mentions_sync_message, token_state
 
         bubble_ready_mentions = prepare_mentions_for_bubble(new_compiled_mentions_to_upload)
         if bubble_ready_mentions:
@@ -172,108 +250,103 @@ def sync_linkedin_mentions(token_state):
             logging.info(f"Successfully uploaded {len(bubble_ready_mentions)} new mentions to Bubble.")
             updated_mentions_df = pd.concat([bubble_mentions_df_orig, pd.DataFrame(bubble_ready_mentions)], ignore_index=True)
             token_state["bubble_mentions_df"] = updated_mentions_df.drop_duplicates(subset=[BUBBLE_MENTIONS_ID_COLUMN_NAME], keep='last')
-            return f"Mentions: Synced {len(bubble_ready_mentions)} new. ", token_state
+            mentions_sync_message = f"Mentions: Synced {len(bubble_ready_mentions)} new. "
         else:
             logging.info("Mentions sync: No new mentions were prepared for Bubble upload.")
-            return "Mentions: No new ones to upload. ", token_state
+            mentions_sync_message = "Mentions: No new ones to upload. "
+            
     except ValueError as ve:
         logging.error(f"ValueError during mentions sync: {ve}", exc_info=True)
-        return f"Mentions Error: {html.escape(str(ve))}. ", token_state
+        mentions_sync_message = f"Mentions Error: {html.escape(str(ve))}. "
     except Exception as e:
         logging.exception("Unexpected error in sync_linkedin_mentions.")
-        return f"Mentions: Unexpected error ({type(e).__name__}). ", token_state
+        mentions_sync_message = f"Mentions: Unexpected error ({type(e).__name__}). "
+    finally:
+        if not attempt_logged and org_urn:
+            token_state = _log_sync_attempt(org_urn, LOG_SUBJECT_MENTIONS, token_state)
+    return mentions_sync_message, token_state
 
 
 def sync_linkedin_follower_stats(token_state):
-    """Fetches new LinkedIn follower statistics and uploads them to Bubble."""
-    logging.info("Starting LinkedIn follower stats sync process.")
+    """Fetches new LinkedIn follower statistics and uploads them to Bubble, if scheduled."""
+    logging.info("Starting LinkedIn follower stats sync process check.")
+
+    if not token_state.get("fs_should_sync_now", False):
+        logging.info("Follower Stats sync: Not scheduled by state_manager. Skipping.")
+        return "Follower Stats: Sync not currently required by schedule. ", token_state
+        
+    logging.info("Follower Stats sync: Proceeding as scheduled by state_manager.")
+
     if not token_state or not token_state.get("token"):
         logging.error("Follower Stats sync: Access denied. No LinkedIn token.")
+        org_urn_for_log = token_state.get('org_urn') if token_state else None
+        if org_urn_for_log:
+            token_state = _log_sync_attempt(org_urn_for_log, LOG_SUBJECT_FOLLOWER_STATS, token_state)
         return "Follower Stats: No token. ", token_state
 
     client_id = token_state.get("client_id")
     token_dict = token_state.get("token")
     org_urn = token_state.get('org_urn')
     bubble_follower_stats_df_orig = token_state.get("bubble_follower_stats_df", pd.DataFrame()).copy()
+    follower_stats_sync_message = ""
+    attempt_logged = False
 
     if not org_urn or not client_id or client_id == "ENV VAR MISSING":
         logging.error("Follower Stats sync: Configuration error (Org URN or Client ID missing).")
+        if org_urn:
+            token_state = _log_sync_attempt(org_urn, LOG_SUBJECT_FOLLOWER_STATS, token_state)
+            attempt_logged = True
         return "Follower Stats: Config error. ", token_state
-
-    follower_stats_sync_is_needed_now = False
-    if bubble_follower_stats_df_orig.empty:
-        follower_stats_sync_is_needed_now = True
-        logging.info("Follower stats sync needed: Bubble DF is empty.")
-    else:
-        monthly_gains_df_check = bubble_follower_stats_df_orig[bubble_follower_stats_df_orig[FOLLOWER_STATS_TYPE_COLUMN] == 'follower_gains_monthly'].copy()
-        if monthly_gains_df_check.empty or FOLLOWER_STATS_CATEGORY_COLUMN not in monthly_gains_df_check.columns:
-            follower_stats_sync_is_needed_now = True
-            logging.info("Follower stats sync needed: Monthly gains data missing or date column absent.")
-        else:
-            monthly_gains_df_check.loc[:, FOLLOWER_STATS_CATEGORY_COLUMN] = pd.to_datetime(monthly_gains_df_check[FOLLOWER_STATS_CATEGORY_COLUMN], errors='coerce').dt.normalize()
-            last_gain_date = monthly_gains_df_check[FOLLOWER_STATS_CATEGORY_COLUMN].dropna().max()
-            
-            if pd.isna(last_gain_date):
-                 follower_stats_sync_is_needed_now = True
-                 logging.info("Follower stats sync needed: No valid dates in monthly gains after conversion for check.")
-            else:
-                if last_gain_date.tzinfo is None or last_gain_date.tzinfo.utcoffset(last_gain_date) is None:
-                    last_gain_date = last_gain_date.tz_localize('UTC')
-                else:
-                    last_gain_date = last_gain_date.tz_convert('UTC')
-                
-                start_of_current_month = pd.Timestamp('now', tz='UTC').normalize().replace(day=1)
-                if last_gain_date < start_of_current_month:
-                    follower_stats_sync_is_needed_now = True
-                    logging.info(f"Follower stats sync needed: Last gain date {last_gain_date} is old or invalid.")
-        
-        if bubble_follower_stats_df_orig[bubble_follower_stats_df_orig[FOLLOWER_STATS_TYPE_COLUMN] != 'follower_gains_monthly'].empty:
-            follower_stats_sync_is_needed_now = True
-            logging.info("Follower stats sync needed: Demographic data (non-monthly) is missing.")
-
-    if not follower_stats_sync_is_needed_now:
-        logging.info("Follower stats data is fresh based on current check. No API fetch needed.")
-        return "Follower Stats: Data up-to-date. ", token_state
-
+    
     logging.info(f"Follower stats sync proceeding for org_urn: {org_urn}")
     try:
         api_follower_stats = get_linkedin_follower_stats(client_id, token_dict, org_urn)
         if not api_follower_stats:
             logging.info(f"Follower Stats sync: No stats found via API for org {org_urn}.")
-            return "Follower Stats: None found via API. ", token_state
+            follower_stats_sync_message = "Follower Stats: None found via API. "
+            token_state = _log_sync_attempt(org_urn, LOG_SUBJECT_FOLLOWER_STATS, token_state)
+            attempt_logged = True
+            return follower_stats_sync_message, token_state
 
         new_stats_to_upload = []
+        # Logic for comparing API stats with existing Bubble stats (monthly gains and demographics)
+        # Monthly Gains
         api_monthly_gains = [s for s in api_follower_stats if s.get(FOLLOWER_STATS_TYPE_COLUMN) == 'follower_gains_monthly']
         existing_monthly_gain_dates = set()
         if not bubble_follower_stats_df_orig.empty:
             bubble_monthly_df = bubble_follower_stats_df_orig[bubble_follower_stats_df_orig[FOLLOWER_STATS_TYPE_COLUMN] == 'follower_gains_monthly']
             if FOLLOWER_STATS_CATEGORY_COLUMN in bubble_monthly_df.columns:
-                existing_monthly_gain_dates = set(bubble_monthly_df[FOLLOWER_STATS_CATEGORY_COLUMN].astype(str).unique())
+                # Convert to string for consistent comparison, assuming API data also provides date as string or convertible
+                existing_monthly_gain_dates = set(pd.to_datetime(bubble_monthly_df[FOLLOWER_STATS_CATEGORY_COLUMN], errors='coerce').dt.strftime('%Y-%m-%d').dropna())
 
         for gain_stat in api_monthly_gains:
-            if str(gain_stat.get(FOLLOWER_STATS_CATEGORY_COLUMN)) not in existing_monthly_gain_dates:
+            api_date_str = pd.to_datetime(gain_stat.get(FOLLOWER_STATS_CATEGORY_COLUMN), errors='coerce').strftime('%Y-%m-%d')
+            if api_date_str != 'NaT' and api_date_str not in existing_monthly_gain_dates:
                 new_stats_to_upload.append(gain_stat)
-
+        
+        # Demographics (overwrite logic: if API has it, and it's different or not present in Bubble, upload)
         api_demographics = [s for s in api_follower_stats if s.get(FOLLOWER_STATS_TYPE_COLUMN) != 'follower_gains_monthly']
+        # Create a map of existing demographics for quick lookup
+        # Key: (org_urn, type, category), Value: (organic_count, paid_count)
         existing_demographics_map = {}
         if not bubble_follower_stats_df_orig.empty:
             bubble_demographics_df = bubble_follower_stats_df_orig[bubble_follower_stats_df_orig[FOLLOWER_STATS_TYPE_COLUMN] != 'follower_gains_monthly']
-            if not bubble_demographics_df.empty and \
-               all(col in bubble_demographics_df.columns for col in [
-                   FOLLOWER_STATS_ORG_URN_COLUMN, FOLLOWER_STATS_TYPE_COLUMN,
-                   FOLLOWER_STATS_CATEGORY_COLUMN, FOLLOWER_STATS_ORGANIC_COLUMN,
-                   FOLLOWER_STATS_PAID_COLUMN
-               ]):
+            required_cols_demo = [
+                FOLLOWER_STATS_ORG_URN_COLUMN, FOLLOWER_STATS_TYPE_COLUMN,
+                FOLLOWER_STATS_CATEGORY_COLUMN, FOLLOWER_STATS_ORGANIC_COLUMN,
+                FOLLOWER_STATS_PAID_COLUMN
+            ]
+            if not bubble_demographics_df.empty and all(col in bubble_demographics_df.columns for col in required_cols_demo):
                 for _, row in bubble_demographics_df.iterrows():
                     key = (
                         str(row[FOLLOWER_STATS_ORG_URN_COLUMN]),
                         str(row[FOLLOWER_STATS_TYPE_COLUMN]),
-                        str(row[FOLLOWER_STATS_CATEGORY_COLUMN])
+                        str(row[FOLLOWER_STATS_CATEGORY_COLUMN]) # Category can be various things
                     )
                     existing_demographics_map[key] = (
-                        row[FOLLOWER_STATS_ORGANIC_COLUMN],
-                        row[FOLLOWER_STATS_PAID_COLUMN]
+                        row[FOLLOWER_STATS_ORGANIC_COLUMN], row[FOLLOWER_STATS_PAID_COLUMN]
                     )
+        
         for demo_stat in api_demographics:
             key = (
                 str(demo_stat.get(FOLLOWER_STATS_ORG_URN_COLUMN)),
@@ -284,33 +357,46 @@ def sync_linkedin_follower_stats(token_state):
                 demo_stat.get(FOLLOWER_STATS_ORGANIC_COLUMN, 0),
                 demo_stat.get(FOLLOWER_STATS_PAID_COLUMN, 0)
             )
+            # If key not in existing OR counts are different, then it's new/changed
             if key not in existing_demographics_map or existing_demographics_map[key] != api_counts:
                 new_stats_to_upload.append(demo_stat)
 
+
         if not new_stats_to_upload:
-            logging.info(f"Follower Stats sync: Data for org {org_urn} is up-to-date or no changes found.")
-            return "Follower Stats: Data up-to-date or no changes. ", token_state
+            logging.info(f"Follower Stats sync: Data for org {org_urn} is up-to-date or no changes found after comparison.")
+            follower_stats_sync_message = "Follower Stats: Data up-to-date or no changes. "
+            token_state = _log_sync_attempt(org_urn, LOG_SUBJECT_FOLLOWER_STATS, token_state)
+            attempt_logged = True
+            return follower_stats_sync_message, token_state
 
         bulk_upload_to_bubble(new_stats_to_upload, BUBBLE_FOLLOWER_STATS_TABLE_NAME)
         logging.info(f"Successfully uploaded {len(new_stats_to_upload)} follower stat entries to Bubble for org {org_urn}.")
 
+        # Update token_state's follower stats DataFrame
         temp_df = pd.concat([bubble_follower_stats_df_orig, pd.DataFrame(new_stats_to_upload)], ignore_index=True)
+        # For monthly gains, keep last entry per org/date (category)
         monthly_part = temp_df[temp_df[FOLLOWER_STATS_TYPE_COLUMN] == 'follower_gains_monthly'].drop_duplicates(
             subset=[FOLLOWER_STATS_ORG_URN_COLUMN, FOLLOWER_STATS_CATEGORY_COLUMN],
             keep='last'
         )
+        # For demographics, keep last entry per org/type/category
         demographics_part = temp_df[temp_df[FOLLOWER_STATS_TYPE_COLUMN] != 'follower_gains_monthly'].drop_duplicates(
             subset=[FOLLOWER_STATS_ORG_URN_COLUMN, FOLLOWER_STATS_TYPE_COLUMN, FOLLOWER_STATS_CATEGORY_COLUMN],
             keep='last'
         )
         token_state["bubble_follower_stats_df"] = pd.concat([monthly_part, demographics_part], ignore_index=True)
-        return f"Follower Stats: Synced {len(new_stats_to_upload)} entries. ", token_state
+        follower_stats_sync_message = f"Follower Stats: Synced {len(new_stats_to_upload)} entries. "
+
     except ValueError as ve:
         logging.error(f"ValueError during follower stats sync for {org_urn}: {ve}", exc_info=True)
-        return f"Follower Stats Error: {html.escape(str(ve))}. ", token_state
+        follower_stats_sync_message = f"Follower Stats Error: {html.escape(str(ve))}. "
     except Exception as e:
         logging.exception(f"Unexpected error in sync_linkedin_follower_stats for {org_urn}.")
-        return f"Follower Stats: Unexpected error ({type(e).__name__}). ", token_state
+        follower_stats_sync_message = f"Follower Stats: Unexpected error ({type(e).__name__}). "
+    finally:
+        if not attempt_logged and org_urn:
+             token_state = _log_sync_attempt(org_urn, LOG_SUBJECT_FOLLOWER_STATS, token_state)
+    return follower_stats_sync_message, token_state
 
 
 def sync_all_linkedin_data_orchestrator(token_state):
@@ -321,7 +407,7 @@ def sync_all_linkedin_data_orchestrator(token_state):
         return "<p style='color:red; text-align:center;'>❌ Access denied. LinkedIn token not available.</p>", token_state
 
     org_urn = token_state.get('org_urn')
-    client_id = token_state.get("client_id") # Client ID should be in token_state from process_and_store_bubble_token
+    client_id = token_state.get("client_id")
 
     posts_sync_message = ""
     mentions_sync_message = ""
@@ -330,22 +416,31 @@ def sync_all_linkedin_data_orchestrator(token_state):
     if not org_urn:
         logging.error("Sync All: Org URN missing in token_state.")
         return "<p style='color:red;'>❌ Config error: Org URN missing.</p>", token_state
-    if not client_id or client_id == "ENV VAR MISSING": # Check client_id from token_state
+    if not client_id or client_id == "ENV VAR MISSING":
         logging.error("Sync All: Client ID missing or not set in token_state.")
         return "<p style='color:red;'>❌ Config error: Client ID missing.</p>", token_state
 
     # --- Sync Posts ---
     fetch_count_for_posts_api = token_state.get('fetch_count_for_api', 0)
     if fetch_count_for_posts_api == 0:
-        posts_sync_message = "Posts: Already up-to-date. "
-        logging.info("Posts sync: Skipped as fetch_count_for_posts_api is 0.")
+        # This means state_manager determined no post sync is needed based on log
+        posts_sync_message = "Posts: Sync not currently required by schedule. "
+        logging.info("Posts sync: Skipped as fetch_count_for_posts_api is 0 (determined by state_manager).")
+        # Log an "attempt" to sync posts which resulted in a skip due to schedule.
+        # This keeps the log fresh, indicating a check was made.
+        token_state = _log_sync_attempt(org_urn, LOG_SUBJECT_POSTS, token_state)
     else:
         posts_sync_message, token_state = _sync_linkedin_posts_internal(token_state, fetch_count_for_posts_api)
+        # _sync_linkedin_posts_internal now handles its own logging internally
 
     # --- Sync Mentions ---
+    # sync_linkedin_mentions will check token_state.get("mentions_should_sync_now")
+    # and log its attempt internally.
     mentions_sync_message, token_state = sync_linkedin_mentions(token_state)
 
     # --- Sync Follower Stats ---
+    # sync_linkedin_follower_stats will check token_state.get("fs_should_sync_now")
+    # and log its attempt internally.
     follower_stats_sync_message, token_state = sync_linkedin_follower_stats(token_state)
 
     logging.info(f"Sync process complete. Messages: Posts: [{posts_sync_message.strip()}], Mentions: [{mentions_sync_message.strip()}], Follower Stats: [{follower_stats_sync_message.strip()}]")