Update analytics_data_processing.py

analytics_data_processing.py

@@ -1,5 +1,5 @@
 import pandas as pd
-from datetime import datetime, timedelta, time # Added time for min.time
+from datetime import datetime, timedelta, time
 import logging
 
 # Configure logging for this module
@@ -23,8 +23,8 @@ def filter_dataframe_by_date(df, date_column, start_date, end_date):
         # Drop rows where date conversion might have failed (NaT) or was originally NaT
         df_copy.dropna(subset=[date_column], inplace=True)
         if df_copy.empty:
-             logging.info(f"Filter by date: DataFrame empty after to_datetime and dropna for column '{date_column}'.")
-             return pd.DataFrame()
+            logging.info(f"Filter by date: DataFrame empty after to_datetime and dropna for column '{date_column}'.")
+            return pd.DataFrame()
 
         # Normalize to midnight. This preserves timezone information if present.
         df_copy[date_column] = df_copy[date_column].dt.normalize()
@@ -39,45 +39,55 @@ def filter_dataframe_by_date(df, date_column, start_date, end_date):
         logging.error(f"Error processing date column '{date_column}': {e}", exc_info=True)
         return pd.DataFrame() 
 
-    df_filtered = df_copy # df_copy is now processed and potentially filtered by dropna
-    # No need for: df_filtered = df_copy.dropna(subset=[date_column]) again here.
-    if df_filtered.empty: # Check again in case all rows were dropped or some other issue.
-        logging.info(f"Filter by date: DataFrame became empty after processing date column '{date_column}'.")
-        return pd.DataFrame()
-
     # Convert start_date and end_date (which are naive Python datetime or naive Pandas Timestamp)
     # to naive pandas Timestamps and normalize them.
     start_dt_obj = pd.to_datetime(start_date, errors='coerce').normalize() if start_date else None
     end_dt_obj = pd.to_datetime(end_date, errors='coerce').normalize() if end_date else None
 
     # Perform the filtering
+    # df_filtered is already df_copy which has NaNs dropped and dates processed
     if start_dt_obj and end_dt_obj:
-        return df_filtered[(df_filtered[date_column] >= start_dt_obj) & (df_filtered[date_column] <= end_dt_obj)]
+        df_filtered_final = df_copy[(df_copy[date_column] >= start_dt_obj) & (df_copy[date_column] <= end_dt_obj)]
     elif start_dt_obj:
-        return df_filtered[df_filtered[date_column] >= start_dt_obj]
+        df_filtered_final = df_copy[df_copy[date_column] >= start_dt_obj]
     elif end_dt_obj:
-        return df_filtered[df_filtered[date_column] <= end_dt_obj]
-    return df_filtered # No date filtering if neither start_date nor end_date is provided
+        df_filtered_final = df_copy[df_copy[date_column] <= end_dt_obj]
+    else:
+        df_filtered_final = df_copy # No date filtering if neither start_date nor end_date is provided
+    
+    if df_filtered_final.empty:
+        logging.info(f"Filter by date: DataFrame became empty after applying date range to column '{date_column}'.")
+    
+    return df_filtered_final
 
 
 def prepare_filtered_analytics_data(token_state_value, date_filter_option, custom_start_date, custom_end_date):
     """
-    Retrieves data from token_state, determines date range, filters posts and mentions.
-    Returns filtered_posts_df, filtered_mentions_df, follower_stats_df (unfiltered),
-    and the determined start_dt, end_dt for messaging.
+    Retrieves data from token_state, determines date range, filters posts, mentions, and follower time-series data.
+    Merges posts with post stats.
+    Returns:
+        - filtered_merged_posts_df: Posts merged with stats, filtered by date.
+        - filtered_mentions_df: Mentions filtered by date.
+        - date_filtered_follower_stats_df: Follower stats filtered by date (for time-series plots).
+        - raw_follower_stats_df: Unfiltered follower stats (for demographic plots).
+        - start_dt_filter: Determined start date for filtering.
+        - end_dt_filter: Determined end date for filtering.
     """
     logging.info(f"Preparing filtered analytics data. Filter: {date_filter_option}, Custom Start: {custom_start_date}, Custom End: {custom_end_date}")
 
-    posts_df = token_state_value.get("bubble_posts_df", pd.DataFrame())
-    mentions_df = token_state_value.get("bubble_mentions_df", pd.DataFrame())
-    follower_stats_df = token_state_value.get("bubble_follower_stats_df", pd.DataFrame()) 
+    posts_df = token_state_value.get("bubble_posts_df", pd.DataFrame()).copy()
+    mentions_df = token_state_value.get("bubble_mentions_df", pd.DataFrame()).copy()
+    follower_stats_df = token_state_value.get("bubble_follower_stats_df", pd.DataFrame()).copy()
+    post_stats_df = token_state_value.get("bubble_post_stats_df", pd.DataFrame()).copy() # Fetch post_stats_df
 
     date_column_posts = token_state_value.get("config_date_col_posts", "published_at")
     date_column_mentions = token_state_value.get("config_date_col_mentions", "date")
+    # Assuming follower_stats_df has a 'date' column for time-series data
+    date_column_followers = token_state_value.get("config_date_col_followers", "date") 
 
-    # Determine date range for filtering posts and mentions
+    # Determine date range for filtering
     current_datetime_obj = datetime.now()
-    current_time_normalized = current_datetime_obj.replace(hour=0, minute=0, second=0, microsecond=0) # Naive Python datetime
+    current_time_normalized = current_datetime_obj.replace(hour=0, minute=0, second=0, microsecond=0) 
     
     end_dt_filter = current_time_normalized 
     start_dt_filter = None
@@ -87,10 +97,7 @@ def prepare_filtered_analytics_data(token_state_value, date_filter_option, custo
     elif date_filter_option == "Last 30 Days":
         start_dt_filter = current_time_normalized - timedelta(days=29) 
     elif date_filter_option == "Custom Range":
-        # custom_start_date and custom_end_date are strings from gr.DateTime(type="string")
-        # Convert to pandas Timestamp (which will be naive if input string is naive) then normalize using pandas method
         start_dt_filter_temp = pd.to_datetime(custom_start_date, errors='coerce')
-        # .replace() on pandas Timestamp normalizes time part
         start_dt_filter = start_dt_filter_temp.replace(hour=0, minute=0, second=0, microsecond=0) if pd.notna(start_dt_filter_temp) else None 
         
         end_dt_filter_temp = pd.to_datetime(custom_end_date, errors='coerce')
@@ -98,15 +105,52 @@ def prepare_filtered_analytics_data(token_state_value, date_filter_option, custo
     
     logging.info(f"Date range for filtering: Start: {start_dt_filter}, End: {end_dt_filter}")
 
-    # Filter DataFrames
-    filtered_posts_data = pd.DataFrame()
-    if not posts_df.empty:
-        filtered_posts_data = filter_dataframe_by_date(posts_df, date_column_posts, start_dt_filter, end_dt_filter)
+    # Merge posts_df and post_stats_df
+    merged_posts_df = pd.DataFrame()
+    if not posts_df.empty and not post_stats_df.empty:
+        # Assuming posts_df has 'id' and post_stats_df has 'post_id' for merging
+        if 'id' in posts_df.columns and 'post_id' in post_stats_df.columns:
+            merged_posts_df = pd.merge(posts_df, post_stats_df, left_on='id', right_on='post_id', how='left')
+            logging.info(f"Merged posts_df ({len(posts_df)} rows) and post_stats_df ({len(post_stats_df)} rows) into merged_posts_df ({len(merged_posts_df)} rows).")
+        else:
+            logging.warning("Cannot merge posts_df and post_stats_df due to missing 'id' or 'post_id' columns.")
+            # Fallback to using posts_df if merge fails but provide an empty df for stats-dependent plots
+            merged_posts_df = posts_df # Or handle as an error / empty DF for those plots
+    elif not posts_df.empty:
+        logging.warning("post_stats_df is empty. Proceeding with posts_df only for plots that don't require stats.")
+        merged_posts_df = posts_df # Create necessary columns with NaN if they are expected by plots
+        # For columns expected from post_stats_df, add them with NaNs if not present
+        expected_stat_cols = ['engagement', 'impressionCount', 'clickCount', 'likeCount', 'commentCount', 'shareCount']
+        for col in expected_stat_cols:
+            if col not in merged_posts_df.columns:
+                merged_posts_df[col] = pd.NA
+
+
+    # Filter DataFrames by date
+    filtered_merged_posts_data = pd.DataFrame()
+    if not merged_posts_df.empty and date_column_posts in merged_posts_df.columns:
+        filtered_merged_posts_data = filter_dataframe_by_date(merged_posts_df, date_column_posts, start_dt_filter, end_dt_filter)
+    elif not merged_posts_df.empty:
+        logging.warning(f"Date column '{date_column_posts}' not found in merged_posts_df. Returning unfiltered merged posts data.")
+        filtered_merged_posts_data = merged_posts_df # Or apply other logic
     
     filtered_mentions_data = pd.DataFrame()
-    if not mentions_df.empty:
+    if not mentions_df.empty and date_column_mentions in mentions_df.columns:
         filtered_mentions_data = filter_dataframe_by_date(mentions_df, date_column_mentions, start_dt_filter, end_dt_filter)
+    elif not mentions_df.empty:
+        logging.warning(f"Date column '{date_column_mentions}' not found in mentions_df. Returning unfiltered mentions data.")
+        filtered_mentions_data = mentions_df
+
+    date_filtered_follower_stats_df = pd.DataFrame()
+    raw_follower_stats_df = follower_stats_df.copy() # For demographic plots, use raw (or latest snapshot logic)
+
+    if not follower_stats_df.empty and date_column_followers in follower_stats_df.columns:
+        date_filtered_follower_stats_df = filter_dataframe_by_date(follower_stats_df, date_column_followers, start_dt_filter, end_dt_filter)
+    elif not follower_stats_df.empty:
+        logging.warning(f"Date column '{date_column_followers}' not found in follower_stats_df. Time-series follower plots might be empty or use unfiltered data.")
+        # Decide if date_filtered_follower_stats_df should be raw_follower_stats_df or empty
+        date_filtered_follower_stats_df = follower_stats_df # Or pd.DataFrame() if strict filtering is required
 
-    logging.info(f"Processed - Filtered posts: {len(filtered_posts_data)} rows, Filtered Mentions: {len(filtered_mentions_data)} rows.")
+    logging.info(f"Processed - Filtered Merged Posts: {len(filtered_merged_posts_data)} rows, Filtered Mentions: {len(filtered_mentions_data)} rows, Date-Filtered Follower Stats: {len(date_filtered_follower_stats_df)} rows.")
     
-    return filtered_posts_data, filtered_mentions_data, follower_stats_df, start_dt_filter, end_dt_filter
+    return filtered_merged_posts_data, filtered_mentions_data, date_filtered_follower_stats_df, raw_follower_stats_df, start_dt_filter, end_dt_filter