Update sync_logic.py

sync_logic.py

@@ -9,7 +9,7 @@ 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, fetch_linkedin_posts_data_from_bubble # Added fetch for log update
+from Bubble_API_Calls import bulk_upload_to_bubble, fetch_linkedin_posts_data_from_bubble, update_record_in_bubble
 # Assuming Linkedin_Data_API_Calls contains all necessary LinkedIn data fetching and processing functions
 from Linkedin_Data_API_Calls import (
     fetch_linkedin_posts_core,
@@ -268,7 +268,12 @@ def sync_linkedin_mentions(token_state):
 
 
 def sync_linkedin_follower_stats(token_state):
-    """Fetches new LinkedIn follower statistics and uploads them to Bubble, if scheduled."""
+    """
+    Fetches new/updated LinkedIn follower statistics and uploads/updates them in Bubble,
+    if scheduled by state_manager.
+    For both monthly gains and demographics, updates counts only if the new LinkedIn count is greater.
+    Creates new records if the category/month doesn't exist.
+    """
     logging.info("Starting LinkedIn follower stats sync process check.")
 
     if not token_state.get("fs_should_sync_now", False):
@@ -288,6 +293,7 @@ def sync_linkedin_follower_stats(token_state):
     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
 
@@ -298,9 +304,19 @@ def sync_linkedin_follower_stats(token_state):
             attempt_logged = True
         return "Follower Stats: Config error. ", token_state
     
+    # Ensure the BUBBLE_UNIQUE_ID_COLUMN_NAME exists in the DataFrame if it's not empty,
+    # as it's crucial for building the maps for updates.
+    if not bubble_follower_stats_df_orig.empty and BUBBLE_UNIQUE_ID_COLUMN_NAME not in bubble_follower_stats_df_orig.columns:
+        logging.error(f"Follower Stats sync: Critical error - '{BUBBLE_UNIQUE_ID_COLUMN_NAME}' column missing in bubble_follower_stats_df. Cannot proceed with updates.")
+        if org_urn:
+             token_state = _log_sync_attempt(org_urn, LOG_SUBJECT_FOLLOWER_STATS, token_state) # Log the attempt despite error
+             attempt_logged = True
+        return f"Follower Stats: Config error ({BUBBLE_UNIQUE_ID_COLUMN_NAME} missing). ", 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}.")
             follower_stats_sync_message = "Follower Stats: None found via API. "
@@ -308,84 +324,192 @@ def sync_linkedin_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:
-                # 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:
-            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']
-            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]) # Category can be various things
-                    )
-                    existing_demographics_map[key] = (
-                        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)),
-                str(demo_stat.get(FOLLOWER_STATS_TYPE_COLUMN)),
-                str(demo_stat.get(FOLLOWER_STATS_CATEGORY_COLUMN))
-            )
-            api_counts = (
-                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)
+        stats_for_bulk_upload = []
+        records_to_update_via_patch = [] # List of tuples: (bubble_id, fields_to_update_dict)
+
+        # --- Prepare maps for existing data in Bubble for efficient lookup ---
+        # Key: (org_urn, type, category_identifier), Value: (organic, paid, bubble_record_id)
+        # For monthly gains, category_identifier is the formatted date string.
+        # For demographics, category_identifier is the FOLLOWER_STATS_CATEGORY_COLUMN value.
+        existing_stats_map = {}
+        stats_required_cols = [
+            FOLLOWER_STATS_ORG_URN_COLUMN, FOLLOWER_STATS_TYPE_COLUMN,
+            FOLLOWER_STATS_CATEGORY_COLUMN, FOLLOWER_STATS_ORGANIC_COLUMN, # Assuming these apply to monthly too
+            FOLLOWER_STATS_PAID_COLUMN,     # Assuming these apply to monthly too
+            BUBBLE_UNIQUE_ID_COLUMN_NAME
+        ]
 
+        if not bubble_follower_stats_df_orig.empty and all(col in bubble_follower_stats_df_orig.columns for col in stats_required_cols):
+            for _, row in bubble_follower_stats_df_orig.iterrows():
+                category_identifier = str(row[FOLLOWER_STATS_CATEGORY_COLUMN])
+                # For monthly gains, ensure category (date) is consistently formatted if needed
+                if row[FOLLOWER_STATS_TYPE_COLUMN] == 'follower_gains_monthly':
+                    try:
+                        category_identifier = pd.to_datetime(row[FOLLOWER_STATS_CATEGORY_COLUMN], errors='coerce').strftime('%Y-%m-%d')
+                        if category_identifier == 'NaT': # Handle parsing errors
+                            logging.warning(f"Could not parse date for existing monthly gain: {row[FOLLOWER_STATS_CATEGORY_COLUMN]}. Skipping this entry for map.")
+                            continue
+                    except Exception: # Catch any other parsing issues
+                        logging.warning(f"Error parsing date for existing monthly gain: {row[FOLLOWER_STATS_CATEGORY_COLUMN]}. Skipping this entry for map.")
+                        continue
+                
+                key = (
+                    str(row[FOLLOWER_STATS_ORG_URN_COLUMN]),
+                    str(row[FOLLOWER_STATS_TYPE_COLUMN]),
+                    category_identifier
+                )
+                existing_stats_map[key] = (
+                    row[FOLLOWER_STATS_ORGANIC_COLUMN], # Assuming monthly gains have this
+                    row[FOLLOWER_STATS_PAID_COLUMN],   # Assuming monthly gains have this
+                    row[BUBBLE_UNIQUE_ID_COLUMN_NAME]
+                )
+        elif not bubble_follower_stats_df_orig.empty:
+            logging.warning(f"Follower Stats: Data in Bubble is missing one or more required columns for update logic: {stats_required_cols}. Will treat all API stats as new if not matched by key elements.")
+
+
+        # --- Process all stats from API (monthly gains and demographics) ---
+        for stat_from_api in api_follower_stats:
+            api_type = str(stat_from_api.get(FOLLOWER_STATS_TYPE_COLUMN))
+            api_category_raw = stat_from_api.get(FOLLOWER_STATS_CATEGORY_COLUMN)
+            
+            api_category_identifier = str(api_category_raw)
+            if api_type == 'follower_gains_monthly':
+                try:
+                    api_category_identifier = pd.to_datetime(api_category_raw, errors='coerce').strftime('%Y-%m-%d')
+                    if api_category_identifier == 'NaT':
+                        logging.warning(f"Could not parse date from API for monthly gain: {api_category_raw}. Skipping this API stat.")
+                        continue
+                except Exception:
+                    logging.warning(f"Error parsing date from API for monthly gain: {api_category_raw}. Skipping this API stat.")
+                    continue
 
-        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 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
+            key = (
+                str(stat_from_api.get(FOLLOWER_STATS_ORG_URN_COLUMN)),
+                api_type,
+                api_category_identifier
+            )
+            
+            # Assuming monthly gains also have organic/paid counts.
+            # If they have different count fields, these need to be specified.
+            # For simplicity, using FOLLOWER_STATS_ORGANIC_COLUMN and FOLLOWER_STATS_PAID_COLUMN.
+            # If monthly gains only have a single 'count' field, adjust logic accordingly.
+            api_organic_count = stat_from_api.get(FOLLOWER_STATS_ORGANIC_COLUMN, 0)
+            api_paid_count = stat_from_api.get(FOLLOWER_STATS_PAID_COLUMN, 0)
+
+            if key not in existing_stats_map:
+                # This stat category/month is entirely new, add for bulk creation
+                stats_for_bulk_upload.append(stat_from_api)
+            else:
+                # Stat category/month exists, check if counts need updating
+                existing_organic, existing_paid, bubble_id = existing_stats_map[key]
+                fields_to_update_in_bubble = {}
+
+                if api_organic_count != existing_organic:
+                    fields_to_update_in_bubble[FOLLOWER_STATS_ORGANIC_COLUMN] = api_organic_count
+                
+                if api_paid_count != existing_paid:
+                    fields_to_update_in_bubble[FOLLOWER_STATS_PAID_COLUMN] = api_paid_count
+                
+                if fields_to_update_in_bubble: # If there's at least one field to update
+                    records_to_update_via_patch.append((bubble_id, fields_to_update_in_bubble))
+        
+        # --- Perform Bubble Operations ---
+        num_bulk_uploaded = 0
+        if stats_for_bulk_upload:
+            if bulk_upload_to_bubble(stats_for_bulk_upload, BUBBLE_FOLLOWER_STATS_TABLE_NAME):
+                num_bulk_uploaded = len(stats_for_bulk_upload)
+                logging.info(f"Successfully bulk-uploaded {num_bulk_uploaded} new follower stat entries to Bubble for org {org_urn}.")
+            else:
+                logging.error(f"Failed to bulk-upload {len(stats_for_bulk_upload)} new follower stat entries for org {org_urn}.")
+        
+        num_patched_updated = 0
+        if records_to_update_via_patch:
+            for bubble_id, fields_to_update in records_to_update_via_patch:
+                if update_record_in_bubble(BUBBLE_FOLLOWER_STATS_TABLE_NAME, bubble_id, fields_to_update):
+                    num_patched_updated += 1
+                else:
+                    logging.error(f"Failed to update record {bubble_id} via PATCH for follower stats for org {org_urn}.")
+            logging.info(f"Attempted to update {len(records_to_update_via_patch)} follower stat entries via PATCH, {num_patched_updated} succeeded for org {org_urn}.")
+
+        if not stats_for_bulk_upload and not records_to_update_via_patch:
+            logging.info(f"Follower Stats sync: Data for org {org_urn} is up-to-date or no changes met update criteria.")
+            follower_stats_sync_message = "Follower Stats: Data up-to-date or no qualifying changes. "
+        else:
+            follower_stats_sync_message = f"Follower Stats: Synced (New: {num_bulk_uploaded}, Updated: {num_patched_updated}). "
+
+        # --- Update token_state's follower stats DataFrame ---
+        current_data_for_state_df = bubble_follower_stats_df_orig.copy()
+
+        if records_to_update_via_patch and num_patched_updated > 0:
+            # Create a temporary map of successful updates for quick lookup
+            successful_updates_map = {
+                bubble_id: fields for i, (bubble_id, fields) in enumerate(records_to_update_via_patch) if i < num_patched_updated
+            }
+            if successful_updates_map: # only proceed if there were successful updates to reflect
+                for index, row in current_data_for_state_df.iterrows():
+                    bubble_id_from_df = row.get(BUBBLE_UNIQUE_ID_COLUMN_NAME)
+                    if bubble_id_from_df in successful_updates_map:
+                        fields_updated = successful_updates_map[bubble_id_from_df]
+                        for col, value in fields_updated.items():
+                            current_data_for_state_df.loc[index, col] = value
+        
+        if stats_for_bulk_upload and num_bulk_uploaded > 0:
+            # Only consider successfully uploaded new records
+            successfully_created_stats = [s for i, s in enumerate(stats_for_bulk_upload) if i < num_bulk_uploaded]
+            if successfully_created_stats:
+                newly_created_df = pd.DataFrame(successfully_created_stats)
+                if not newly_created_df.empty:
+                    for col in current_data_for_state_df.columns:
+                        if col not in newly_created_df.columns:
+                            newly_created_df[col] = pd.NA # Use pd.NA for missing values
+                    # Align columns before concat to avoid issues with differing column orders or types
+                    aligned_newly_created_df = newly_created_df.reindex(columns=current_data_for_state_df.columns).fillna(pd.NA)
+                    current_data_for_state_df = pd.concat([current_data_for_state_df, aligned_newly_created_df], ignore_index=True)
+
+        if not current_data_for_state_df.empty:
+            # Deduplication logic (important after combining original, patched, and new data)
+            # Ensure consistent primary key for deduplication across types
+            # For monthly gains, primary key is (org_urn, type='follower_gains_monthly', category=date_str)
+            # For demographics, primary key is (org_urn, type, category)
+            
+            # To handle this, we can sort by a hypothetical 'last_modified_indicator' if we had one,
+            # or rely on 'keep=last' after ensuring data is ordered such that API data (potentially newer) comes later.
+            # The concat order (original, then new) and then drop_duplicates with keep='last' on identifying keys is standard.
 
-        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)
-        follower_stats_sync_message = f"Follower Stats: Synced {len(new_stats_to_upload)} entries. "
+            # We need to define unique keys for each type to drop duplicates correctly.
+            # The current deduplication splits by type and then applies different subsets. This should still work.
+            
+            monthly_part = current_data_for_state_df[current_data_for_state_df[FOLLOWER_STATS_TYPE_COLUMN] == 'follower_gains_monthly']
+            if not monthly_part.empty:
+                # Ensure category is consistently formatted for monthly gains before deduplication
+                monthly_part_copy = monthly_part.copy() # To avoid SettingWithCopyWarning
+                monthly_part_copy[FOLLOWER_STATS_CATEGORY_COLUMN] = pd.to_datetime(monthly_part_copy[FOLLOWER_STATS_CATEGORY_COLUMN], errors='coerce').dt.strftime('%Y-%m-%d')
+                monthly_part = monthly_part_copy.drop_duplicates(
+                    subset=[FOLLOWER_STATS_ORG_URN_COLUMN, FOLLOWER_STATS_TYPE_COLUMN, FOLLOWER_STATS_CATEGORY_COLUMN],
+                    keep='last'
+                )
+
+            demographics_part = current_data_for_state_df[current_data_for_state_df[FOLLOWER_STATS_TYPE_COLUMN] != 'follower_gains_monthly']
+            if not demographics_part.empty:
+                demo_subset_cols = [FOLLOWER_STATS_ORG_URN_COLUMN, FOLLOWER_STATS_TYPE_COLUMN, FOLLOWER_STATS_CATEGORY_COLUMN]
+                if all(col in demographics_part.columns for col in demo_subset_cols):
+                    demographics_part = demographics_part.drop_duplicates(
+                        subset=demo_subset_cols,
+                        keep='last'
+                    )
+                else:
+                    logging.warning("Follower Stats: Missing columns for demographic deduplication in token_state update. Skipping.")
+            
+            if monthly_part.empty and demographics_part.empty:
+                token_state["bubble_follower_stats_df"] = pd.DataFrame(columns=bubble_follower_stats_df_orig.columns)
+            elif monthly_part.empty:
+                token_state["bubble_follower_stats_df"] = demographics_part.reset_index(drop=True) if not demographics_part.empty else pd.DataFrame(columns=bubble_follower_stats_df_orig.columns)
+            elif demographics_part.empty:
+                token_state["bubble_follower_stats_df"] = monthly_part.reset_index(drop=True) if not monthly_part.empty else pd.DataFrame(columns=bubble_follower_stats_df_orig.columns)
+            else:
+                token_state["bubble_follower_stats_df"] = pd.concat([monthly_part, demographics_part], ignore_index=True)
+        else:
+            token_state["bubble_follower_stats_df"] = pd.DataFrame(columns=bubble_follower_stats_df_orig.columns)
 
     except ValueError as ve:
         logging.error(f"ValueError during follower stats sync for {org_urn}: {ve}", exc_info=True)
@@ -395,7 +519,8 @@ def sync_linkedin_follower_stats(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)
+            token_state = _log_sync_attempt(org_urn, LOG_SUBJECT_FOLLOWER_STATS, token_state)
+            
     return follower_stats_sync_message, token_state