LinkedinMonitor / analytics_plot_generator.py
GuglielmoTor's picture
Update analytics_plot_generator.py
adab1ec verified
raw
history blame
47.3 kB
import pandas as pd
import matplotlib.pyplot as plt
import logging
from io import BytesIO
import base64
import numpy as np
import matplotlib.ticker as mticker
# Configure logging for this module
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(module)s - %(message)s')
def create_placeholder_plot(title="No Data or Plot Error", message="Data might be empty or an error occurred."):
"""Creates a placeholder Matplotlib plot indicating no data or an error."""
try:
fig, ax = plt.subplots(figsize=(8, 4))
ax.text(0.5, 0.5, f"{title}\n{message}", ha='center', va='center', fontsize=10, wrap=True)
ax.axis('off')
plt.tight_layout()
return fig
except Exception as e:
logging.error(f"Error creating placeholder plot: {e}")
# Fallback placeholder if the above fails
fig_err, ax_err = plt.subplots()
ax_err.text(0.5, 0.5, "Fatal: Plot generation error", ha='center', va='center')
ax_err.axis('off')
return fig_err
# No plt.close(fig) here as Gradio handles the figure object.
def generate_posts_activity_plot(df, date_column='published_at'):
"""Generates a plot for posts activity over time."""
logging.info(f"Generating posts activity plot. Date column: '{date_column}'. Input df rows: {len(df) if df is not None else 'None'}")
if df is None or df.empty:
logging.warning(f"Posts activity: DataFrame is empty.")
return create_placeholder_plot(title="Posts Activity Over Time", message="No data available for the selected period.")
if date_column not in df.columns:
logging.warning(f"Posts activity: Date column '{date_column}' is missing. Cols: {df.columns.tolist()}.")
return create_placeholder_plot(title="Posts Activity Over Time", message=f"Date column '{date_column}' not found.")
try:
df_copy = df.copy()
if not pd.api.types.is_datetime64_any_dtype(df_copy[date_column]):
df_copy[date_column] = pd.to_datetime(df_copy[date_column], errors='coerce')
df_copy = df_copy.dropna(subset=[date_column])
if df_copy.empty:
logging.info("Posts activity: DataFrame empty after NaNs dropped from date column.")
return create_placeholder_plot(title="Posts Activity Over Time", message="No valid date entries found.")
posts_over_time = df_copy.set_index(date_column).resample('D').size()
if posts_over_time.empty:
logging.info("Posts activity: No posts after resampling by day.")
return create_placeholder_plot(title="Posts Activity Over Time", message="No posts in the selected period.")
fig, ax = plt.subplots(figsize=(10, 5))
posts_over_time.plot(kind='line', ax=ax, marker='o', linestyle='-')
ax.set_title('Posts Activity Over Time')
ax.set_xlabel('Date')
ax.set_ylabel('Number of Posts')
ax.grid(True, linestyle='--', alpha=0.7)
plt.xticks(rotation=45)
plt.tight_layout()
logging.info("Successfully generated posts activity plot.")
return fig
except Exception as e:
logging.error(f"Error generating posts activity plot: {e}", exc_info=True)
return create_placeholder_plot(title="Posts Activity Error", message=str(e))
finally:
plt.close('all')
def generate_engagement_type_plot(df, likes_col='likeCount', comments_col='commentCount', shares_col='shareCount'): # Updated col names
"""Generates a bar plot for total engagement types (likes, comments, shares)."""
logging.info(f"Generating engagement type plot. Input df rows: {len(df) if df is not None else 'None'}")
required_cols = [likes_col, comments_col, shares_col]
if df is None or df.empty:
logging.warning("Engagement type: DataFrame is empty.")
return create_placeholder_plot(title="Post Engagement Types", message="No data available for the selected period.")
missing_cols = [col for col in required_cols if col not in df.columns]
if missing_cols:
msg = f"Engagement type: Columns missing: {missing_cols}. Available: {df.columns.tolist()}"
logging.warning(msg)
return create_placeholder_plot(title="Post Engagement Types", message=msg)
try:
df_copy = df.copy()
for col in required_cols:
df_copy[col] = pd.to_numeric(df_copy[col], errors='coerce').fillna(0)
total_likes = df_copy[likes_col].sum()
total_comments = df_copy[comments_col].sum()
total_shares = df_copy[shares_col].sum()
if total_likes == 0 and total_comments == 0 and total_shares == 0:
logging.info("Engagement type: All engagement counts are zero.")
return create_placeholder_plot(title="Post Engagement Types", message="No engagement data (likes, comments, shares) in the selected period.")
engagement_data = {
'Likes': total_likes,
'Comments': total_comments,
'Shares': total_shares
}
fig, ax = plt.subplots(figsize=(8, 5))
bars = ax.bar(engagement_data.keys(), engagement_data.values(), color=['skyblue', 'lightgreen', 'salmon'])
ax.set_title('Total Post Engagement Types')
ax.set_xlabel('Engagement Type')
ax.set_ylabel('Total Count')
ax.grid(axis='y', linestyle='--', alpha=0.7)
for bar in bars:
yval = bar.get_height()
ax.text(bar.get_x() + bar.get_width()/2.0, yval + (0.01 * max(engagement_data.values(), default=10)), str(int(yval)), ha='center', va='bottom')
plt.tight_layout()
logging.info("Successfully generated engagement type plot.")
return fig
except Exception as e:
logging.error(f"Error generating engagement type plot: {e}", exc_info=True)
return create_placeholder_plot(title="Engagement Type Error", message=str(e))
finally:
plt.close('all')
def generate_mentions_activity_plot(df, date_column='date'):
"""Generates a plot for mentions activity over time."""
logging.info(f"Generating mentions activity plot. Date column: '{date_column}'. Input df rows: {len(df) if df is not None else 'None'}")
if df is None or df.empty:
logging.warning(f"Mentions activity: DataFrame is empty.")
return create_placeholder_plot(title="Mentions Activity Over Time", message="No data available for the selected period.")
if date_column not in df.columns:
logging.warning(f"Mentions activity: Date column '{date_column}' is missing. Cols: {df.columns.tolist()}.")
return create_placeholder_plot(title="Mentions Activity Over Time", message=f"Date column '{date_column}' not found.")
try:
df_copy = df.copy()
if not pd.api.types.is_datetime64_any_dtype(df_copy[date_column]):
df_copy[date_column] = pd.to_datetime(df_copy[date_column], errors='coerce')
df_copy = df_copy.dropna(subset=[date_column])
if df_copy.empty:
logging.info("Mentions activity: DataFrame empty after NaNs dropped from date column.")
return create_placeholder_plot(title="Mentions Activity Over Time", message="No valid date entries found.")
mentions_over_time = df_copy.set_index(date_column).resample('D').size()
if mentions_over_time.empty:
logging.info("Mentions activity: No mentions after resampling by day.")
return create_placeholder_plot(title="Mentions Activity Over Time", message="No mentions in the selected period.")
fig, ax = plt.subplots(figsize=(10, 5))
mentions_over_time.plot(kind='line', ax=ax, marker='o', linestyle='-', color='purple')
ax.set_title('Mentions Activity Over Time')
ax.set_xlabel('Date')
ax.set_ylabel('Number of Mentions')
ax.grid(True, linestyle='--', alpha=0.7)
plt.xticks(rotation=45)
plt.tight_layout()
logging.info("Successfully generated mentions activity plot.")
return fig
except Exception as e:
logging.error(f"Error generating mentions activity plot: {e}", exc_info=True)
return create_placeholder_plot(title="Mentions Activity Error", message=str(e))
finally:
plt.close('all')
def generate_mention_sentiment_plot(df, sentiment_column='sentiment_label'):
"""Generates a pie chart for mention sentiment distribution."""
logging.info(f"Generating mention sentiment plot. Sentiment column: '{sentiment_column}'. Input df rows: {len(df) if df is not None else 'None'}")
if df is None or df.empty:
logging.warning("Mention sentiment: DataFrame is empty.")
return create_placeholder_plot(title="Mention Sentiment Distribution", message="No data available for the selected period.")
if sentiment_column not in df.columns:
msg = f"Mention sentiment: Column '{sentiment_column}' is missing. Available: {df.columns.tolist()}"
logging.warning(msg)
return create_placeholder_plot(title="Mention Sentiment Distribution", message=msg)
try:
df_copy = df.copy()
sentiment_counts = df_copy[sentiment_column].value_counts()
if sentiment_counts.empty:
logging.info("Mention sentiment: No sentiment data after value_counts.")
return create_placeholder_plot(title="Mention Sentiment Distribution", message="No sentiment data available.")
fig, ax = plt.subplots(figsize=(8, 5))
# Using a more distinct color map if available, or fallback
try:
colors_map = plt.cm.get_cmap('Pastel1', len(sentiment_counts))
except ValueError: # Fallback if Pastel1 doesn't have enough colors or isn't available
colors_map = plt.cm.get_cmap('viridis', len(sentiment_counts))
pie_colors = [colors_map(i) for i in range(len(sentiment_counts))]
ax.pie(sentiment_counts, labels=sentiment_counts.index, autopct='%1.1f%%', startangle=90, colors=pie_colors)
ax.set_title('Mention Sentiment Distribution')
ax.axis('equal')
plt.tight_layout()
logging.info("Successfully generated mention sentiment plot.")
return fig
except Exception as e:
logging.error(f"Error generating mention sentiment plot: {e}", exc_info=True)
return create_placeholder_plot(title="Mention Sentiment Error", message=str(e))
finally:
plt.close('all')
# --- Updated Follower Plot Functions ---
def generate_followers_count_over_time_plot(df, date_info_column='category_name',
organic_count_col='follower_count_organic',
paid_count_col='follower_count_paid',
type_filter_column='follower_count_type',
type_value='follower_gains_monthly'):
"""
Generates a plot for specific follower counts (organic and paid) over time.
Date information is expected in 'date_info_column' as strings (e.g., "2024-08-01").
"""
title = f"Followers Count Over Time ({type_value})"
logging.info(f"Generating {title}. Date Info: '{date_info_column}', Organic: '{organic_count_col}', Paid: '{paid_count_col}', Type Filter: '{type_filter_column}=={type_value}'. DF rows: {len(df) if df is not None else 'None'}")
if df is None or df.empty:
return create_placeholder_plot(title=title, message="No follower data available.")
required_cols = [date_info_column, organic_count_col, paid_count_col, type_filter_column]
missing_cols = [col for col in required_cols if col not in df.columns]
if missing_cols:
return create_placeholder_plot(title=title, message=f"Missing columns: {missing_cols}. Available: {df.columns.tolist()}")
try:
df_copy = df.copy()
df_filtered = df_copy[df_copy[type_filter_column] == type_value].copy() # Use .copy() to avoid SettingWithCopyWarning
if df_filtered.empty:
return create_placeholder_plot(title=title, message=f"No data for type '{type_value}'.")
# Convert date_info_column to datetime
df_filtered['datetime_obj'] = pd.to_datetime(df_filtered[date_info_column], errors='coerce')
df_filtered[organic_count_col] = pd.to_numeric(df_filtered[organic_count_col], errors='coerce').fillna(0)
df_filtered[paid_count_col] = pd.to_numeric(df_filtered[paid_count_col], errors='coerce').fillna(0)
df_filtered = df_filtered.dropna(subset=['datetime_obj', organic_count_col, paid_count_col]).sort_values(by='datetime_obj')
if df_filtered.empty:
return create_placeholder_plot(title=title, message="No valid data after cleaning and filtering.")
fig, ax = plt.subplots(figsize=(10, 5))
ax.plot(df_filtered['datetime_obj'], df_filtered[organic_count_col], marker='o', linestyle='-', color='dodgerblue', label='Organic Followers')
ax.plot(df_filtered['datetime_obj'], df_filtered[paid_count_col], marker='x', linestyle='--', color='seagreen', label='Paid Followers')
ax.set_title(title)
ax.set_xlabel('Date')
ax.set_ylabel('Follower Count')
ax.legend()
ax.grid(True, linestyle='--', alpha=0.7)
plt.xticks(rotation=45)
plt.tight_layout()
return fig
except Exception as e:
logging.error(f"Error generating {title}: {e}", exc_info=True)
return create_placeholder_plot(title=f"{title} Error", message=str(e))
finally:
plt.close('all')
def generate_followers_growth_rate_plot(df, date_info_column='category_name',
organic_count_col='follower_count_organic',
paid_count_col='follower_count_paid',
type_filter_column='follower_count_type',
type_value='follower_gains_monthly'):
"""
Calculates and plots follower growth rate (organic and paid) over time.
Date information is expected in 'date_info_column' as strings (e.g., "2024-08-01").
"""
title = f"Follower Growth Rate ({type_value})"
logging.info(f"Generating {title}. Date Info: '{date_info_column}', Organic: '{organic_count_col}', Paid: '{paid_count_col}', Type Filter: '{type_filter_column}=={type_value}'. DF rows: {len(df) if df is not None else 'None'}")
if df is None or df.empty:
return create_placeholder_plot(title=title, message="No follower data available.")
required_cols = [date_info_column, organic_count_col, paid_count_col, type_filter_column]
missing_cols = [col for col in required_cols if col not in df.columns]
if missing_cols:
return create_placeholder_plot(title=title, message=f"Missing columns: {missing_cols}. Available: {df.columns.tolist()}")
try:
df_copy = df.copy()
df_filtered = df_copy[df_copy[type_filter_column] == type_value].copy()
if df_filtered.empty:
return create_placeholder_plot(title=title, message=f"No data for type '{type_value}'.")
df_filtered['datetime_obj'] = pd.to_datetime(df_filtered[date_info_column], errors='coerce')
df_filtered[organic_count_col] = pd.to_numeric(df_filtered[organic_count_col], errors='coerce')
df_filtered[paid_count_col] = pd.to_numeric(df_filtered[paid_count_col], errors='coerce')
df_filtered = df_filtered.dropna(subset=['datetime_obj']).sort_values(by='datetime_obj').set_index('datetime_obj')
if df_filtered.empty or len(df_filtered) < 2: # Need at least 2 points for pct_change
return create_placeholder_plot(title=title, message="Not enough data points to calculate growth rate.")
df_filtered['organic_growth_rate'] = df_filtered[organic_count_col].pct_change() * 100
df_filtered['paid_growth_rate'] = df_filtered[paid_count_col].pct_change() * 100
# Replace inf with NaN then drop NaNs for growth rates
df_filtered.replace([np.inf, -np.inf], np.nan, inplace=True)
# df_filtered.dropna(subset=['organic_growth_rate', 'paid_growth_rate'], how='all', inplace=True) # Keep row if at least one rate is valid
fig, ax = plt.subplots(figsize=(10, 5))
plotted_organic = False
if 'organic_growth_rate' in df_filtered.columns and not df_filtered['organic_growth_rate'].dropna().empty:
ax.plot(df_filtered.index, df_filtered['organic_growth_rate'], marker='o', linestyle='-', color='lightcoral', label='Organic Growth Rate')
plotted_organic = True
plotted_paid = False
if 'paid_growth_rate' in df_filtered.columns and not df_filtered['paid_growth_rate'].dropna().empty:
ax.plot(df_filtered.index, df_filtered['paid_growth_rate'], marker='x', linestyle='--', color='mediumpurple', label='Paid Growth Rate')
plotted_paid = True
if not plotted_organic and not plotted_paid:
return create_placeholder_plot(title=title, message="No valid growth rate data to display after calculation.")
ax.set_title(title)
ax.set_xlabel('Date')
ax.set_ylabel('Growth Rate (%)')
ax.yaxis.set_major_formatter(mticker.PercentFormatter())
ax.legend()
ax.grid(True, linestyle='--', alpha=0.7)
plt.xticks(rotation=45)
plt.tight_layout()
return fig
except Exception as e:
logging.error(f"Error generating {title}: {e}", exc_info=True)
return create_placeholder_plot(title=f"{title} Error", message=str(e))
finally:
plt.close('all')
def generate_followers_by_demographics_plot(df, category_col='category_name',
organic_count_col='follower_count_organic',
paid_count_col='follower_count_paid',
type_filter_column='follower_count_type',
type_value=None, plot_title="Followers by Demographics"):
"""
Generates a grouped bar chart for follower demographics (organic and paid).
'category_col' here is the demographic attribute (e.g., Location, Industry).
"""
logging.info(f"Generating {plot_title}. Category: '{category_col}', Organic: '{organic_count_col}', Paid: '{paid_count_col}', Type Filter: '{type_filter_column}=={type_value}'. DF rows: {len(df) if df is not None else 'None'}")
if df is None or df.empty:
return create_placeholder_plot(title=plot_title, message="No follower data available.")
required_cols = [category_col, organic_count_col, paid_count_col, type_filter_column]
missing_cols = [col for col in required_cols if col not in df.columns]
if missing_cols:
return create_placeholder_plot(title=plot_title, message=f"Missing columns: {missing_cols}. Available: {df.columns.tolist()}")
if type_value is None:
return create_placeholder_plot(title=plot_title, message="Demographic type (type_value) not specified.")
try:
df_copy = df.copy()
df_filtered = df_copy[df_copy[type_filter_column] == type_value].copy()
if df_filtered.empty:
return create_placeholder_plot(title=plot_title, message=f"No data for demographic type '{type_value}'.")
df_filtered[organic_count_col] = pd.to_numeric(df_filtered[organic_count_col], errors='coerce').fillna(0)
df_filtered[paid_count_col] = pd.to_numeric(df_filtered[paid_count_col], errors='coerce').fillna(0)
demographics_data = df_filtered.groupby(category_col)[[organic_count_col, paid_count_col]].sum()
# Sort by total followers (organic + paid) for better visualization
demographics_data['total_for_sort'] = demographics_data[organic_count_col] + demographics_data[paid_count_col]
demographics_data = demographics_data.sort_values(by='total_for_sort', ascending=False).drop(columns=['total_for_sort'])
if demographics_data.empty:
return create_placeholder_plot(title=plot_title, message="No demographic data to display after filtering and aggregation.")
top_n = 10
if len(demographics_data) > top_n:
demographics_data = demographics_data.head(top_n)
plot_title_updated = f"{plot_title} (Top {top_n})"
else:
plot_title_updated = plot_title
fig, ax = plt.subplots(figsize=(12, 7) if len(demographics_data) > 5 else (10,6) )
bar_width = 0.35
index = np.arange(len(demographics_data.index))
bars1 = ax.bar(index - bar_width/2, demographics_data[organic_count_col], bar_width, label='Organic', color='skyblue')
bars2 = ax.bar(index + bar_width/2, demographics_data[paid_count_col], bar_width, label='Paid', color='lightcoral')
ax.set_title(plot_title_updated)
ax.set_xlabel(category_col.replace('_', ' ').title())
ax.set_ylabel('Number of Followers')
ax.set_xticks(index)
ax.set_xticklabels(demographics_data.index, rotation=45, ha="right")
ax.legend()
ax.grid(axis='y', linestyle='--', alpha=0.7)
# Add labels on top of bars
for bar_group in [bars1, bars2]:
for bar in bar_group:
yval = bar.get_height()
if yval > 0: # Only add label if value is not zero
ax.text(bar.get_x() + bar.get_width()/2.0, yval + (0.01 * ax.get_ylim()[1]),
str(int(yval)), ha='center', va='bottom', fontsize=8)
plt.tight_layout()
return fig
except Exception as e:
logging.error(f"Error generating {plot_title}: {e}", exc_info=True)
return create_placeholder_plot(title=f"{plot_title} Error", message=str(e))
finally:
plt.close('all')
def generate_engagement_rate_over_time_plot(df, date_column='published_at', engagement_rate_col='engagement'):
"""Generates a plot for engagement rate over time."""
title = "Engagement Rate Over Time"
logging.info(f"Generating {title}. Date: '{date_column}', Rate Col: '{engagement_rate_col}'. DF rows: {len(df) if df is not None else 'None'}")
if df is None or df.empty:
return create_placeholder_plot(title=title, message="No post data for engagement rate.")
required_cols = [date_column, engagement_rate_col]
missing_cols = [col for col in required_cols if col not in df.columns]
if missing_cols:
return create_placeholder_plot(title=title, message=f"Missing columns: {missing_cols}. Available: {df.columns.tolist()}")
try:
df_copy = df.copy()
df_copy[date_column] = pd.to_datetime(df_copy[date_column], errors='coerce')
df_copy[engagement_rate_col] = pd.to_numeric(df_copy[engagement_rate_col], errors='coerce')
df_copy = df_copy.dropna(subset=[date_column, engagement_rate_col]).set_index(date_column)
if df_copy.empty:
return create_placeholder_plot(title=title, message="No valid data after cleaning.")
engagement_over_time = df_copy.resample('D')[engagement_rate_col].mean()
engagement_over_time = engagement_over_time.dropna()
if engagement_over_time.empty:
return create_placeholder_plot(title=title, message="No engagement rate data to display after resampling.")
fig, ax = plt.subplots(figsize=(10, 5))
ax.plot(engagement_over_time.index, engagement_over_time.values, marker='.', linestyle='-', color='darkorange')
ax.set_title(title)
ax.set_xlabel('Date')
ax.set_ylabel('Engagement Rate')
# Adjust xmax for PercentFormatter based on whether rate is 0-1 or 0-100
max_rate_val = engagement_over_time.max() if not engagement_over_time.empty else 0
formatter_xmax = 1.0 if max_rate_val <= 1.5 and max_rate_val !=0 else 100.0 # Heuristic: if max is small, assume 0-1 scale
if max_rate_val > 100 and formatter_xmax == 1.0: # If data is clearly > 100 but we assumed 0-1
formatter_xmax = max_rate_val # Or some other sensible upper bound for formatting
ax.yaxis.set_major_formatter(mticker.PercentFormatter(xmax=formatter_xmax))
ax.grid(True, linestyle='--', alpha=0.7)
plt.xticks(rotation=45)
plt.tight_layout()
return fig
except Exception as e:
logging.error(f"Error generating {title}: {e}", exc_info=True)
return create_placeholder_plot(title=f"{title} Error", message=str(e))
finally:
plt.close('all')
def generate_reach_over_time_plot(df, date_column='published_at', reach_col='clickCount'):
"""Generates a plot for reach (clicks) over time."""
title = "Reach Over Time (Clicks)"
logging.info(f"Generating {title}. Date: '{date_column}', Reach Col: '{reach_col}'. DF rows: {len(df) if df is not None else 'None'}")
if df is None or df.empty:
return create_placeholder_plot(title=title, message="No post data for reach.")
required_cols = [date_column, reach_col]
missing_cols = [col for col in required_cols if col not in df.columns]
if missing_cols:
return create_placeholder_plot(title=title, message=f"Missing columns: {missing_cols}. Available: {df.columns.tolist()}")
try:
df_copy = df.copy()
df_copy[date_column] = pd.to_datetime(df_copy[date_column], errors='coerce')
df_copy[reach_col] = pd.to_numeric(df_copy[reach_col], errors='coerce')
df_copy = df_copy.dropna(subset=[date_column, reach_col]).set_index(date_column)
if df_copy.empty: # After dropping NaNs for essential columns
return create_placeholder_plot(title=title, message="No valid data after cleaning for reach plot.")
reach_over_time = df_copy.resample('D')[reach_col].sum()
# No need to check if reach_over_time is empty if df_copy wasn't, sum of NaNs is 0.
# Plot will show 0 if all sums are 0.
fig, ax = plt.subplots(figsize=(10, 5))
ax.plot(reach_over_time.index, reach_over_time.values, marker='.', linestyle='-', color='mediumseagreen')
ax.set_title(title)
ax.set_xlabel('Date')
ax.set_ylabel('Total Clicks')
ax.grid(True, linestyle='--', alpha=0.7)
plt.xticks(rotation=45)
plt.tight_layout()
return fig
except Exception as e:
logging.error(f"Error generating {title}: {e}", exc_info=True)
return create_placeholder_plot(title=f"{title} Error", message=str(e))
finally:
plt.close('all')
def generate_impressions_over_time_plot(df, date_column='published_at', impressions_col='impressionCount'):
"""Generates a plot for impressions over time."""
title = "Impressions Over Time"
logging.info(f"Generating {title}. Date: '{date_column}', Impressions Col: '{impressions_col}'. DF rows: {len(df) if df is not None else 'None'}")
if df is None or df.empty:
return create_placeholder_plot(title=title, message="No post data for impressions.")
required_cols = [date_column, impressions_col]
missing_cols = [col for col in required_cols if col not in df.columns]
if missing_cols:
return create_placeholder_plot(title=title, message=f"Missing columns: {missing_cols}. Available: {df.columns.tolist()}")
try:
df_copy = df.copy()
df_copy[date_column] = pd.to_datetime(df_copy[date_column], errors='coerce')
df_copy[impressions_col] = pd.to_numeric(df_copy[impressions_col], errors='coerce')
df_copy = df_copy.dropna(subset=[date_column, impressions_col]).set_index(date_column)
if df_copy.empty: # After dropping NaNs for essential columns
return create_placeholder_plot(title=title, message="No valid data after cleaning for impressions plot.")
impressions_over_time = df_copy.resample('D')[impressions_col].sum()
fig, ax = plt.subplots(figsize=(10, 5))
ax.plot(impressions_over_time.index, impressions_over_time.values, marker='.', linestyle='-', color='slateblue')
ax.set_title(title)
ax.set_xlabel('Date')
ax.set_ylabel('Total Impressions')
ax.grid(True, linestyle='--', alpha=0.7)
plt.xticks(rotation=45)
plt.tight_layout()
return fig
except Exception as e:
logging.error(f"Error generating {title}: {e}", exc_info=True)
return create_placeholder_plot(title=f"{title} Error", message=str(e))
finally:
plt.close('all')
# --- NEW PLOT FUNCTIONS ---
def generate_likes_over_time_plot(df, date_column='published_at', likes_col='likeCount'):
"""Generates a plot for likes (reactions) over time."""
title = "Reactions (Likes) Over Time"
logging.info(f"Generating {title}. Date: '{date_column}', Likes Col: '{likes_col}'. DF rows: {len(df) if df is not None else 'None'}")
if df is None or df.empty:
return create_placeholder_plot(title=title, message="No post data for likes.")
required_cols = [date_column, likes_col]
missing_cols = [col for col in required_cols if col not in df.columns]
if missing_cols:
return create_placeholder_plot(title=title, message=f"Missing columns: {missing_cols}. Available: {df.columns.tolist()}")
try:
df_copy = df.copy()
df_copy[date_column] = pd.to_datetime(df_copy[date_column], errors='coerce')
df_copy[likes_col] = pd.to_numeric(df_copy[likes_col], errors='coerce')
df_copy = df_copy.dropna(subset=[date_column, likes_col]).set_index(date_column)
if df_copy.empty:
return create_placeholder_plot(title=title, message="No valid data after cleaning for likes plot.")
likes_over_time = df_copy.resample('D')[likes_col].sum()
fig, ax = plt.subplots(figsize=(10, 5))
ax.plot(likes_over_time.index, likes_over_time.values, marker='.', linestyle='-', color='crimson')
ax.set_title(title)
ax.set_xlabel('Date')
ax.set_ylabel('Total Likes')
ax.grid(True, linestyle='--', alpha=0.7)
plt.xticks(rotation=45)
plt.tight_layout()
return fig
except Exception as e:
logging.error(f"Error generating {title}: {e}", exc_info=True)
return create_placeholder_plot(title=f"{title} Error", message=str(e))
finally:
plt.close('all')
def generate_clicks_over_time_plot(df, date_column='published_at', clicks_col='clickCount'):
"""Generates a plot for clicks over time (distinct from general reach if needed)."""
title = "Clicks Over Time"
logging.info(f"Generating {title}. Date: '{date_column}', Clicks Col: '{clicks_col}'. DF rows: {len(df) if df is not None else 'None'}")
if df is None or df.empty:
return create_placeholder_plot(title=title, message="No post data for clicks.")
required_cols = [date_column, clicks_col]
missing_cols = [col for col in required_cols if col not in df.columns]
if missing_cols:
return create_placeholder_plot(title=title, message=f"Missing columns: {missing_cols}. Available: {df.columns.tolist()}")
try:
df_copy = df.copy()
df_copy[date_column] = pd.to_datetime(df_copy[date_column], errors='coerce')
df_copy[clicks_col] = pd.to_numeric(df_copy[clicks_col], errors='coerce')
df_copy = df_copy.dropna(subset=[date_column, clicks_col]).set_index(date_column)
if df_copy.empty:
return create_placeholder_plot(title=title, message="No valid data after cleaning for clicks plot.")
clicks_over_time = df_copy.resample('D')[clicks_col].sum()
fig, ax = plt.subplots(figsize=(10, 5))
ax.plot(clicks_over_time.index, clicks_over_time.values, marker='.', linestyle='-', color='teal')
ax.set_title(title)
ax.set_xlabel('Date')
ax.set_ylabel('Total Clicks')
ax.grid(True, linestyle='--', alpha=0.7)
plt.xticks(rotation=45)
plt.tight_layout()
return fig
except Exception as e:
logging.error(f"Error generating {title}: {e}", exc_info=True)
return create_placeholder_plot(title=f"{title} Error", message=str(e))
finally:
plt.close('all')
def generate_shares_over_time_plot(df, date_column='published_at', shares_col='shareCount'):
"""Generates a plot for shares over time."""
title = "Shares Over Time"
logging.info(f"Generating {title}. Date: '{date_column}', Shares Col: '{shares_col}'. DF rows: {len(df) if df is not None else 'None'}")
if df is None or df.empty:
return create_placeholder_plot(title=title, message="No post data for shares.")
required_cols = [date_column, shares_col]
missing_cols = [col for col in required_cols if col not in df.columns]
if missing_cols:
return create_placeholder_plot(title=title, message=f"Missing columns: {missing_cols}. Available: {df.columns.tolist()}")
try:
df_copy = df.copy()
df_copy[date_column] = pd.to_datetime(df_copy[date_column], errors='coerce')
df_copy[shares_col] = pd.to_numeric(df_copy[shares_col], errors='coerce')
df_copy = df_copy.dropna(subset=[date_column, shares_col]).set_index(date_column)
if df_copy.empty:
return create_placeholder_plot(title=title, message="No valid data after cleaning for shares plot.")
shares_over_time = df_copy.resample('D')[shares_col].sum()
fig, ax = plt.subplots(figsize=(10, 5))
ax.plot(shares_over_time.index, shares_over_time.values, marker='.', linestyle='-', color='goldenrod')
ax.set_title(title)
ax.set_xlabel('Date')
ax.set_ylabel('Total Shares')
ax.grid(True, linestyle='--', alpha=0.7)
plt.xticks(rotation=45)
plt.tight_layout()
return fig
except Exception as e:
logging.error(f"Error generating {title}: {e}", exc_info=True)
return create_placeholder_plot(title=f"{title} Error", message=str(e))
finally:
plt.close('all')
def generate_comments_over_time_plot(df, date_column='published_at', comments_col='commentCount'):
"""Generates a plot for comments over time."""
title = "Comments Over Time"
logging.info(f"Generating {title}. Date: '{date_column}', Comments Col: '{comments_col}'. DF rows: {len(df) if df is not None else 'None'}")
if df is None or df.empty:
return create_placeholder_plot(title=title, message="No post data for comments.")
required_cols = [date_column, comments_col]
missing_cols = [col for col in required_cols if col not in df.columns]
if missing_cols:
return create_placeholder_plot(title=title, message=f"Missing columns: {missing_cols}. Available: {df.columns.tolist()}")
try:
df_copy = df.copy()
df_copy[date_column] = pd.to_datetime(df_copy[date_column], errors='coerce')
df_copy[comments_col] = pd.to_numeric(df_copy[comments_col], errors='coerce')
df_copy = df_copy.dropna(subset=[date_column, comments_col]).set_index(date_column)
if df_copy.empty:
return create_placeholder_plot(title=title, message="No valid data after cleaning for comments plot.")
comments_over_time = df_copy.resample('D')[comments_col].sum()
fig, ax = plt.subplots(figsize=(10, 5))
ax.plot(comments_over_time.index, comments_over_time.values, marker='.', linestyle='-', color='forestgreen')
ax.set_title(title)
ax.set_xlabel('Date')
ax.set_ylabel('Total Comments')
ax.grid(True, linestyle='--', alpha=0.7)
plt.xticks(rotation=45)
plt.tight_layout()
return fig
except Exception as e:
logging.error(f"Error generating {title}: {e}", exc_info=True)
return create_placeholder_plot(title=f"{title} Error", message=str(e))
finally:
plt.close('all')
def generate_comments_sentiment_breakdown_plot(df, sentiment_column='comment_sentiment', text_column_for_check='comment_text'):
"""
Generates a pie chart for comment sentiment distribution.
Expects a DataFrame with a 'sentiment_column' for sentiment labels of individual comments.
'text_column_for_check' is used to infer if the DataFrame might contain comment-level data.
"""
title = "Breakdown of Comments by Sentiment"
logging.info(f"Generating {title}. Sentiment Col: '{sentiment_column}'. DF rows: {len(df) if df is not None else 'None'}")
if df is None or df.empty:
return create_placeholder_plot(title=title, message="No data available for comment sentiment analysis.")
# Check if the expected sentiment column exists.
# Also check for a text column as a heuristic, as sentiment often comes with text.
if sentiment_column not in df.columns:
msg = (f"Column '{sentiment_column}' for comment sentiment not found. "
f"This plot requires a DataFrame with pre-analyzed comment sentiments. "
f"Available columns: {df.columns.tolist()}")
logging.warning(msg)
return create_placeholder_plot(title=title, message=msg)
# Optional: A light check if it seems like post-level data instead of comment-level
# if text_column_for_check not in df.columns and 'commentCount' in df.columns:
# logging.warning(f"'{sentiment_column}' found, but '{text_column_for_check}' is missing. Ensure '{sentiment_column}' refers to individual comment sentiments, not post sentiment.")
try:
df_copy = df.copy()
# Ensure sentiment column is treated as categorical/string
df_copy[sentiment_column] = df_copy[sentiment_column].astype(str)
sentiment_counts = df_copy[sentiment_column].value_counts()
if sentiment_counts.empty or sentiment_counts.sum() == 0:
return create_placeholder_plot(title=title, message="No comment sentiment data to display after processing.")
fig, ax = plt.subplots(figsize=(8, 5))
try:
colors_map = plt.cm.get_cmap('Pastel2', len(sentiment_counts))
except ValueError:
colors_map = plt.cm.get_cmap('Accent', len(sentiment_counts))
pie_colors = [colors_map(i) for i in range(len(sentiment_counts))]
ax.pie(sentiment_counts, labels=sentiment_counts.index, autopct='%1.1f%%', startangle=140, colors=pie_colors)
ax.set_title(title)
ax.axis('equal')
plt.tight_layout()
logging.info(f"Successfully generated {title} plot.")
return fig
except Exception as e:
logging.error(f"Error generating {title}: {e}", exc_info=True)
return create_placeholder_plot(title=f"{title} Error", message=str(e))
finally:
plt.close('all')
if __name__ == '__main__':
# Create dummy data for testing
posts_data = {
'id': [f'post{i}' for i in range(1, 8)], # Extended to 7 for more data points
'published_at': pd.to_datetime(['2023-01-01', '2023-01-01', '2023-01-02', '2023-01-03', '2023-01-03', '2023-01-03', '2023-01-04']),
'likeCount': [10, 5, 12, 8, 15, 3, 20],
'commentCount': [2, 1, 3, 1, 4, 0, 5],
'shareCount': [1, 0, 1, 1, 2, 0, 1],
'clickCount': [20, 15, 30, 22, 40, 10, 50],
'impressionCount': [200, 150, 300, 220, 400, 100, 500],
'engagement': [0.05, 0.04, 0.06, 0.055, 0.07, 0.03, 0.08]
}
sample_merged_posts_df = pd.DataFrame(posts_data)
# Dummy data for comments with sentiment (hypothetical)
comments_sentiment_data = {
'comment_id': range(10),
'post_id': ['post1']*3 + ['post2']*2 + ['post3']*5,
'comment_text': ['Great post!', 'I disagree.', 'Nice work.', 'Interesting.', 'Could be better.', 'Loved it!', 'Not sure.', 'Thanks!', 'Helpful.', 'Okay.'],
'comment_sentiment': ['Positive', 'Negative', 'Positive', 'Neutral', 'Negative', 'Positive', 'Neutral', 'Positive', 'Positive', 'Neutral']
}
sample_comments_with_sentiment_df = pd.DataFrame(comments_sentiment_data)
# Updated Follower Stats Data
follower_data = {
'follower_count_type': [
'follower_gains_monthly', 'follower_gains_monthly', 'follower_gains_monthly',
'follower_geo', 'follower_geo', 'follower_geo',
'follower_function', 'follower_function',
'follower_industry', 'follower_industry',
'follower_seniority', 'follower_seniority'
],
# 'category_name' now holds dates for time-series, and actual categories for demographics
'category_name': [
'2024-01-01', '2024-02-01', '2024-03-01', # Dates for monthly gains
'USA', 'Canada', 'UK', # Geo
'Engineering', 'Sales', # Function/Role
'Tech', 'Finance', # Industry
'Senior', 'Junior' # Seniority
],
'follower_count_organic': [
100, 110, 125, # Organic monthly gains
500, 300, 150, # Organic Geo counts
400, 200, # Organic Role counts
250, 180, # Organic Industry counts
300, 220 # Organic Seniority counts
],
'follower_count_paid': [
20, 30, 25, # Paid monthly gains
50, 40, 60, # Paid Geo counts
30, 20, # Paid Role counts
45, 35, # Paid Industry counts
60, 40 # Paid Seniority counts
]
}
sample_follower_stats_df = pd.DataFrame(follower_data)
logging.info("--- Testing Updated Follower Plot Generations ---")
fig_followers_count = generate_followers_count_over_time_plot(
sample_follower_stats_df.copy(),
type_value='follower_gains_monthly' # date_info_column defaults to 'category_name'
)
if fig_followers_count and not isinstance(fig_followers_count, str): logging.info("Followers Count Over Time (monthly, organic/paid) plot generated.")
fig_followers_rate = generate_followers_growth_rate_plot(
sample_follower_stats_df.copy(),
type_value='follower_gains_monthly' # date_info_column defaults to 'category_name'
)
if fig_followers_rate and not isinstance(fig_followers_rate, str): logging.info("Followers Growth Rate (monthly, organic/paid) plot generated.")
fig_geo = generate_followers_by_demographics_plot(
sample_follower_stats_df.copy(),
type_value='follower_geo', # category_col defaults to 'category_name'
plot_title="Followers by Location (Organic/Paid)"
)
if fig_geo and not isinstance(fig_geo, str): logging.info("Followers by Location (grouped organic/paid) plot generated.")
fig_role = generate_followers_by_demographics_plot(
sample_follower_stats_df.copy(),
type_value='follower_function',
plot_title="Followers by Role (Organic/Paid)"
)
if fig_role and not isinstance(fig_role, str): logging.info("Followers by Role (grouped organic/paid) plot generated.")
fig_industry = generate_followers_by_demographics_plot(
sample_follower_stats_df.copy(),
type_value='follower_industry',
plot_title="Followers by Industry (Organic/Paid)"
)
if fig_industry and not isinstance(fig_industry, str): logging.info("Followers by Industry (grouped organic/paid) plot generated.")
fig_seniority = generate_followers_by_demographics_plot(
sample_follower_stats_df.copy(),
type_value='follower_seniority',
plot_title="Followers by Seniority (Organic/Paid)"
)
if fig_seniority and not isinstance(fig_seniority, str): logging.info("Followers by Seniority (grouped organic/paid) plot generated.")
logging.info("--- Testing Other Plot Generations (No Changes to these) ---")
fig_posts_activity = generate_posts_activity_plot(sample_merged_posts_df.copy())
if fig_posts_activity and not isinstance(fig_posts_activity, str): logging.info("Posts activity plot generated.")
fig_engagement_type = generate_engagement_type_plot(sample_merged_posts_df.copy())
if fig_engagement_type and not isinstance(fig_engagement_type, str): logging.info("Engagement type plot generated.")
# Dummy mentions for testing
mentions_data = {
'date': pd.to_datetime(['2023-01-01', '2023-01-02', '2023-01-02', '2023-01-03']),
'sentiment_label': ['Positive', 'Negative', 'Positive', 'Neutral']
}
sample_mentions_df = pd.DataFrame(mentions_data)
fig_mentions_activity = generate_mentions_activity_plot(sample_mentions_df.copy())
if fig_mentions_activity and not isinstance(fig_mentions_activity, str): logging.info("Mentions activity plot generated.")
fig_mention_sentiment = generate_mention_sentiment_plot(sample_mentions_df.copy())
if fig_mention_sentiment and not isinstance(fig_mention_sentiment, str): logging.info("Mention sentiment plot generated.")
fig_eng_rate = generate_engagement_rate_over_time_plot(sample_merged_posts_df.copy())
if fig_eng_rate and not isinstance(fig_eng_rate, str): logging.info("Engagement Rate Over Time plot generated.")
fig_reach = generate_reach_over_time_plot(sample_merged_posts_df.copy())
if fig_reach and not isinstance(fig_reach, str): logging.info("Reach Over Time (Clicks) plot generated.")
fig_impressions = generate_impressions_over_time_plot(sample_merged_posts_df.copy())
if fig_impressions and not isinstance(fig_impressions, str): logging.info("Impressions Over Time plot generated.")
logging.info("--- Testing NEW Plot Generations ---")
fig_likes_time = generate_likes_over_time_plot(sample_merged_posts_df.copy())
if fig_likes_time and not isinstance(fig_likes_time, str): logging.info("Likes Over Time plot generated.")
fig_clicks_time = generate_clicks_over_time_plot(sample_merged_posts_df.copy())
if fig_clicks_time and not isinstance(fig_clicks_time, str): logging.info("Clicks Over Time plot generated.")
fig_shares_time = generate_shares_over_time_plot(sample_merged_posts_df.copy())
if fig_shares_time and not isinstance(fig_shares_time, str): logging.info("Shares Over Time plot generated.")
fig_comments_time = generate_comments_over_time_plot(sample_merged_posts_df.copy())
if fig_comments_time and not isinstance(fig_comments_time, str): logging.info("Comments Over Time plot generated.")
fig_comments_sentiment = generate_comments_sentiment_breakdown_plot(sample_comments_with_sentiment_df.copy())
if fig_comments_sentiment and not isinstance(fig_comments_sentiment, str): logging.info("Comments Sentiment Breakdown plot generated (with dummy comment sentiment data).")
fig_comments_sentiment_no_data = generate_comments_sentiment_breakdown_plot(sample_merged_posts_df.copy()) # Test with df lacking the sentiment col
if fig_comments_sentiment_no_data and not isinstance(fig_comments_sentiment_no_data, str) : logging.info("Comments Sentiment Breakdown plot generated (placeholder, as expected).")
logging.info("Test script finished. Review plots if displayed locally or saved.")