LinkedinMonitor / ui_generators.py
GuglielmoTor's picture
Update ui_generators.py
9c20604 verified
raw
history blame
23.4 kB
# ui_generators.py
"""
Generates HTML content and Matplotlib plots for the Gradio UI tabs,
and UI components for the Analytics tab.
"""
import pandas as pd
import logging
import matplotlib.pyplot as plt
import matplotlib # To ensure backend is switched before any plt import from other modules if app structure changes
import gradio as gr # Added for UI components
# Switch backend for Matplotlib to Agg for Gradio compatibility
matplotlib.use('Agg')
# Assuming config.py contains all necessary constants
from config import (
BUBBLE_POST_DATE_COLUMN_NAME, BUBBLE_MENTIONS_DATE_COLUMN_NAME, BUBBLE_MENTIONS_ID_COLUMN_NAME,
FOLLOWER_STATS_TYPE_COLUMN, FOLLOWER_STATS_CATEGORY_COLUMN, FOLLOWER_STATS_ORGANIC_COLUMN,
FOLLOWER_STATS_PAID_COLUMN, FOLLOWER_STATS_CATEGORY_COLUMN_DT, UI_DATE_FORMAT, UI_MONTH_FORMAT
)
# Configure logging for this module if not already configured at app level
# logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(module)s - %(message)s')
# --- Constants for Button Icons/Text ---
BOMB_ICON = "πŸ’£ Insights"
EXPLORE_ICON = "πŸ” Explore"
FORMULA_ICON = "Ζ’ Formula"
ACTIVE_ICON = "❌ Close"
def display_main_dashboard(token_state):
"""Generates HTML for the main dashboard display using data from token_state."""
if not token_state or not token_state.get("token"):
logging.warning("Dashboard display: Access denied. No token available.")
return "❌ Access denied. No token available for dashboard."
html_parts = ["<div style='padding:10px;'><h3>Dashboard Overview</h3>"]
# Display Recent Posts
posts_df = token_state.get("bubble_posts_df", pd.DataFrame())
html_parts.append(f"<h4>Recent Posts ({len(posts_df)} in Bubble):</h4>")
if not posts_df.empty:
cols_to_show_posts = [col for col in [BUBBLE_POST_DATE_COLUMN_NAME, 'text', 'sentiment', 'summary_text', 'li_eb_label'] if col in posts_df.columns]
if not cols_to_show_posts:
html_parts.append("<p>No relevant post columns found to display.</p>")
else:
display_df_posts = posts_df.copy()
if BUBBLE_POST_DATE_COLUMN_NAME in display_df_posts.columns:
try:
# Ensure the date column is datetime before formatting
display_df_posts[BUBBLE_POST_DATE_COLUMN_NAME] = pd.to_datetime(display_df_posts[BUBBLE_POST_DATE_COLUMN_NAME], errors='coerce')
display_df_posts = display_df_posts.sort_values(by=BUBBLE_POST_DATE_COLUMN_NAME, ascending=False)
# Format for display after sorting
display_df_posts[BUBBLE_POST_DATE_COLUMN_NAME] = display_df_posts[BUBBLE_POST_DATE_COLUMN_NAME].dt.strftime(UI_DATE_FORMAT)
except Exception as e:
logging.error(f"Error formatting post dates for display: {e}")
html_parts.append("<p>Error formatting post dates.</p>")
html_parts.append(display_df_posts[cols_to_show_posts].head().to_html(escape=False, index=False, classes="table table-striped table-sm"))
else:
html_parts.append("<p>No posts loaded from Bubble.</p>")
html_parts.append("<hr/>")
# Display Recent Mentions
mentions_df = token_state.get("bubble_mentions_df", pd.DataFrame())
html_parts.append(f"<h4>Recent Mentions ({len(mentions_df)} in Bubble):</h4>")
if not mentions_df.empty:
cols_to_show_mentions = [col for col in [BUBBLE_MENTIONS_DATE_COLUMN_NAME, "mention_text", "sentiment_label"] if col in mentions_df.columns]
if not cols_to_show_mentions:
html_parts.append("<p>No relevant mention columns found to display.</p>")
else:
display_df_mentions = mentions_df.copy()
if BUBBLE_MENTIONS_DATE_COLUMN_NAME in display_df_mentions.columns:
try:
display_df_mentions[BUBBLE_MENTIONS_DATE_COLUMN_NAME] = pd.to_datetime(display_df_mentions[BUBBLE_MENTIONS_DATE_COLUMN_NAME], errors='coerce')
display_df_mentions = display_df_mentions.sort_values(by=BUBBLE_MENTIONS_DATE_COLUMN_NAME, ascending=False)
display_df_mentions[BUBBLE_MENTIONS_DATE_COLUMN_NAME] = display_df_mentions[BUBBLE_MENTIONS_DATE_COLUMN_NAME].dt.strftime(UI_DATE_FORMAT)
except Exception as e:
logging.error(f"Error formatting mention dates for display: {e}")
html_parts.append("<p>Error formatting mention dates.</p>")
html_parts.append(display_df_mentions[cols_to_show_mentions].head().to_html(escape=False, index=False, classes="table table-striped table-sm"))
else:
html_parts.append("<p>No mentions loaded from Bubble.</p>")
html_parts.append("<hr/>")
# Display Follower Statistics Summary
follower_stats_df = token_state.get("bubble_follower_stats_df", pd.DataFrame())
html_parts.append(f"<h4>Follower Statistics ({len(follower_stats_df)} entries in Bubble):</h4>")
if not follower_stats_df.empty:
monthly_gains = follower_stats_df[follower_stats_df[FOLLOWER_STATS_TYPE_COLUMN] == 'follower_gains_monthly'].copy()
if not monthly_gains.empty and FOLLOWER_STATS_CATEGORY_COLUMN in monthly_gains.columns and \
FOLLOWER_STATS_ORGANIC_COLUMN in monthly_gains.columns and FOLLOWER_STATS_PAID_COLUMN in monthly_gains.columns:
try:
monthly_gains.loc[:, FOLLOWER_STATS_CATEGORY_COLUMN_DT] = pd.to_datetime(monthly_gains[FOLLOWER_STATS_CATEGORY_COLUMN], errors='coerce')
monthly_gains_display = monthly_gains.sort_values(by=FOLLOWER_STATS_CATEGORY_COLUMN_DT, ascending=False)
latest_gain = monthly_gains_display.head(1).copy()
if not latest_gain.empty:
latest_gain.loc[:, FOLLOWER_STATS_CATEGORY_COLUMN] = latest_gain[FOLLOWER_STATS_CATEGORY_COLUMN_DT].dt.strftime(UI_DATE_FORMAT)
html_parts.append("<h5>Latest Monthly Follower Gain:</h5>")
html_parts.append(latest_gain[[FOLLOWER_STATS_CATEGORY_COLUMN, FOLLOWER_STATS_ORGANIC_COLUMN, FOLLOWER_STATS_PAID_COLUMN]].to_html(escape=True, index=False, classes="table table-sm"))
else:
html_parts.append("<p>No valid monthly follower gain data to display after processing.</p>")
except Exception as e:
logging.error(f"Error formatting follower gain dates for display: {e}", exc_info=True)
html_parts.append("<p>Error displaying monthly follower gain data.</p>")
else:
html_parts.append("<p>No monthly follower gain data or required columns are missing.</p>")
demographics_count = len(follower_stats_df[follower_stats_df[FOLLOWER_STATS_TYPE_COLUMN] != 'follower_gains_monthly'])
html_parts.append(f"<p>Total demographic entries (seniority, industry, etc.): {demographics_count}</p>")
else:
html_parts.append("<p>No follower statistics loaded from Bubble.</p>")
html_parts.append("</div>")
return "".join(html_parts)
def run_mentions_tab_display(token_state):
"""Generates HTML and a plot for the Mentions tab."""
logging.info("Updating Mentions Tab display.")
if not token_state or not token_state.get("token"):
logging.warning("Mentions tab: Access denied. No token.")
return "❌ Access denied. No token available for mentions.", None
mentions_df = token_state.get("bubble_mentions_df", pd.DataFrame())
if mentions_df.empty:
logging.info("Mentions tab: No mentions data in Bubble.")
return "<p style='text-align:center;'>No mentions data in Bubble. Try syncing.</p>", None
html_parts = ["<h3 style='text-align:center;'>Recent Mentions</h3>"]
display_columns = [col for col in [BUBBLE_MENTIONS_DATE_COLUMN_NAME, "mention_text", "sentiment_label", BUBBLE_MENTIONS_ID_COLUMN_NAME] if col in mentions_df.columns]
mentions_df_display = mentions_df.copy()
if BUBBLE_MENTIONS_DATE_COLUMN_NAME in mentions_df_display.columns:
try:
mentions_df_display[BUBBLE_MENTIONS_DATE_COLUMN_NAME] = pd.to_datetime(mentions_df_display[BUBBLE_MENTIONS_DATE_COLUMN_NAME], errors='coerce')
mentions_df_display = mentions_df_display.sort_values(by=BUBBLE_MENTIONS_DATE_COLUMN_NAME, ascending=False)
mentions_df_display[BUBBLE_MENTIONS_DATE_COLUMN_NAME] = mentions_df_display[BUBBLE_MENTIONS_DATE_COLUMN_NAME].dt.strftime(UI_DATE_FORMAT)
except Exception as e:
logging.error(f"Error formatting mention dates for tab display: {e}")
html_parts.append("<p>Error formatting mention dates.</p>")
if not display_columns or mentions_df_display[display_columns].empty:
html_parts.append("<p>Required columns for mentions display are missing or no data after processing.</p>")
else:
html_parts.append(mentions_df_display[display_columns].head(20).to_html(escape=False, index=False, classes="table table-sm"))
mentions_html_output = "\n".join(html_parts)
fig = None
if not mentions_df.empty and "sentiment_label" in mentions_df.columns:
try:
fig_plot, ax = plt.subplots(figsize=(6,4))
sentiment_counts = mentions_df["sentiment_label"].value_counts()
sentiment_counts.plot(kind='bar', ax=ax, color=['#4CAF50', '#FFC107', '#F44336', '#9E9E9E', '#2196F3'])
ax.set_title("Mention Sentiment Distribution", y=1.03) # MODIFIED: Added y parameter
ax.set_ylabel("Count")
plt.xticks(rotation=45, ha='right')
plt.tight_layout() # Ensure tight_layout is called
fig = fig_plot
logging.info("Mentions tab: Sentiment distribution plot generated.")
except Exception as e:
logging.error(f"Error generating mentions plot: {e}", exc_info=True)
fig = None # Ensure fig is None on error
finally:
# Only close if fig_plot was successfully created and is not the global plt context
if fig_plot and fig_plot is not plt: # fig_plot is the figure object from subplots
plt.close(fig_plot)
# Fallback if somehow a global figure was left open without being assigned to fig_plot
# elif fig is None and plt.get_fignums():
# plt.close('all') # This might be too aggressive if other plots are intended to be open
return mentions_html_output, fig
def run_follower_stats_tab_display(token_state):
"""Generates HTML and plots for the Follower Stats tab."""
logging.info("Updating Follower Stats Tab display.")
if not token_state or not token_state.get("token"):
logging.warning("Follower stats tab: Access denied. No token.")
return "❌ Access denied. No token available for follower stats.", None, None, None
follower_stats_df_orig = token_state.get("bubble_follower_stats_df", pd.DataFrame())
if follower_stats_df_orig.empty:
logging.info("Follower stats tab: No follower stats data in Bubble.")
return "<p style='text-align:center;'>No follower stats data in Bubble. Try syncing.</p>", None, None, None
follower_stats_df = follower_stats_df_orig.copy()
html_parts = ["<div style='padding:10px;'><h3 style='text-align:center;'>Follower Statistics Overview</h3>"]
plot_monthly_gains = None
plot_seniority_dist = None
plot_industry_dist = None
# Monthly Gains Plot
fig_gains_local = None # Use local var for figure to manage closing
try:
monthly_gains_df = follower_stats_df[
(follower_stats_df[FOLLOWER_STATS_TYPE_COLUMN] == 'follower_gains_monthly') &
(follower_stats_df[FOLLOWER_STATS_CATEGORY_COLUMN].notna()) &
(follower_stats_df[FOLLOWER_STATS_ORGANIC_COLUMN].notna()) &
(follower_stats_df[FOLLOWER_STATS_PAID_COLUMN].notna())
].copy()
if not monthly_gains_df.empty:
monthly_gains_df.loc[:, FOLLOWER_STATS_CATEGORY_COLUMN_DT] = pd.to_datetime(monthly_gains_df[FOLLOWER_STATS_CATEGORY_COLUMN], errors='coerce')
monthly_gains_df_sorted_table = monthly_gains_df.sort_values(by=FOLLOWER_STATS_CATEGORY_COLUMN_DT, ascending=False)
html_parts.append("<h4>Monthly Follower Gains (Last 13 Months):</h4>")
table_display_df = monthly_gains_df_sorted_table.copy()
table_display_df.loc[:,FOLLOWER_STATS_CATEGORY_COLUMN] = table_display_df[FOLLOWER_STATS_CATEGORY_COLUMN_DT].dt.strftime(UI_MONTH_FORMAT)
html_parts.append(table_display_df[[FOLLOWER_STATS_CATEGORY_COLUMN, FOLLOWER_STATS_ORGANIC_COLUMN, FOLLOWER_STATS_PAID_COLUMN]].head(13).to_html(escape=True, index=False, classes="table table-sm"))
monthly_gains_df_sorted_plot = monthly_gains_df.sort_values(by=FOLLOWER_STATS_CATEGORY_COLUMN_DT, ascending=True).copy()
monthly_gains_df_sorted_plot.loc[:, '_plot_month'] = monthly_gains_df_sorted_plot[FOLLOWER_STATS_CATEGORY_COLUMN_DT].dt.strftime(UI_MONTH_FORMAT)
plot_data = monthly_gains_df_sorted_plot.groupby('_plot_month').agg(
organic=(FOLLOWER_STATS_ORGANIC_COLUMN, 'sum'),
paid=(FOLLOWER_STATS_PAID_COLUMN, 'sum')
).reset_index()
# Ensure months are sorted correctly for plotting after groupby
plot_data['_plot_month_dt'] = pd.to_datetime(plot_data['_plot_month'], format=UI_MONTH_FORMAT)
plot_data = plot_data.sort_values(by='_plot_month_dt')
fig_gains_local, ax_gains = plt.subplots(figsize=(10,5))
ax_gains.plot(plot_data['_plot_month'], plot_data['organic'], marker='o', linestyle='-', label='Organic Gain')
ax_gains.plot(plot_data['_plot_month'], plot_data['paid'], marker='x', linestyle='--', label='Paid Gain')
ax_gains.set_title("Monthly Follower Gains Over Time", y=1.03) # MODIFIED
ax_gains.set_ylabel("Follower Count")
ax_gains.set_xlabel("Month (YYYY-MM)")
plt.xticks(rotation=45, ha='right')
ax_gains.legend()
plt.grid(True, linestyle='--', alpha=0.7)
plt.tight_layout()
plot_monthly_gains = fig_gains_local # Assign to the return variable
logging.info("Follower stats tab: Monthly gains plot generated.")
else:
html_parts.append("<p>No monthly follower gain data available or required columns missing.</p>")
except Exception as e:
logging.error(f"Error processing or plotting monthly gains: {e}", exc_info=True)
html_parts.append("<p>Error displaying monthly follower gain data.</p>")
plot_monthly_gains = None # Ensure it's None on error
finally:
if fig_gains_local and fig_gains_local is not plt: # Check local var
plt.close(fig_gains_local)
html_parts.append("<hr/>")
# Seniority Plot
fig_seniority_local = None
try:
seniority_df = follower_stats_df[
(follower_stats_df[FOLLOWER_STATS_TYPE_COLUMN] == 'follower_seniority') &
(follower_stats_df[FOLLOWER_STATS_CATEGORY_COLUMN].notna()) &
(follower_stats_df[FOLLOWER_STATS_ORGANIC_COLUMN].notna())
].copy()
if not seniority_df.empty:
seniority_df_sorted = seniority_df.sort_values(by=FOLLOWER_STATS_ORGANIC_COLUMN, ascending=False)
html_parts.append("<h4>Followers by Seniority (Top 10 Organic):</h4>")
html_parts.append(seniority_df_sorted[[FOLLOWER_STATS_CATEGORY_COLUMN, FOLLOWER_STATS_ORGANIC_COLUMN, FOLLOWER_STATS_PAID_COLUMN]].head(10).to_html(escape=True, index=False, classes="table table-sm"))
fig_seniority_local, ax_seniority = plt.subplots(figsize=(8,5))
top_n_seniority = seniority_df_sorted.nlargest(10, FOLLOWER_STATS_ORGANIC_COLUMN)
ax_seniority.bar(top_n_seniority[FOLLOWER_STATS_CATEGORY_COLUMN], top_n_seniority[FOLLOWER_STATS_ORGANIC_COLUMN], color='skyblue')
ax_seniority.set_title("Follower Distribution by Seniority (Top 10 Organic)", y=1.03) # MODIFIED
ax_seniority.set_ylabel("Organic Follower Count")
plt.xticks(rotation=45, ha='right')
plt.grid(axis='y', linestyle='--', alpha=0.7)
plt.tight_layout()
plot_seniority_dist = fig_seniority_local
logging.info("Follower stats tab: Seniority distribution plot generated.")
else:
html_parts.append("<p>No follower seniority data available or required columns missing.</p>")
except Exception as e:
logging.error(f"Error processing or plotting seniority data: {e}", exc_info=True)
html_parts.append("<p>Error displaying follower seniority data.</p>")
plot_seniority_dist = None
finally:
if fig_seniority_local and fig_seniority_local is not plt:
plt.close(fig_seniority_local)
html_parts.append("<hr/>")
# Industry Plot
fig_industry_local = None
try:
industry_df = follower_stats_df[
(follower_stats_df[FOLLOWER_STATS_TYPE_COLUMN] == 'follower_industry') &
(follower_stats_df[FOLLOWER_STATS_CATEGORY_COLUMN].notna()) &
(follower_stats_df[FOLLOWER_STATS_ORGANIC_COLUMN].notna())
].copy()
if not industry_df.empty:
industry_df_sorted = industry_df.sort_values(by=FOLLOWER_STATS_ORGANIC_COLUMN, ascending=False)
html_parts.append("<h4>Followers by Industry (Top 10 Organic):</h4>")
html_parts.append(industry_df_sorted[[FOLLOWER_STATS_CATEGORY_COLUMN, FOLLOWER_STATS_ORGANIC_COLUMN, FOLLOWER_STATS_PAID_COLUMN]].head(10).to_html(escape=True, index=False, classes="table table-sm"))
fig_industry_local, ax_industry = plt.subplots(figsize=(8,5))
top_n_industry = industry_df_sorted.nlargest(10, FOLLOWER_STATS_ORGANIC_COLUMN)
ax_industry.bar(top_n_industry[FOLLOWER_STATS_CATEGORY_COLUMN], top_n_industry[FOLLOWER_STATS_ORGANIC_COLUMN], color='lightcoral')
ax_industry.set_title("Follower Distribution by Industry (Top 10 Organic)", y=1.03) # MODIFIED
ax_industry.set_ylabel("Organic Follower Count")
plt.xticks(rotation=45, ha='right')
plt.grid(axis='y', linestyle='--', alpha=0.7)
plt.tight_layout()
plot_industry_dist = fig_industry_local
logging.info("Follower stats tab: Industry distribution plot generated.")
else:
html_parts.append("<p>No follower industry data available or required columns missing.</p>")
except Exception as e:
logging.error(f"Error processing or plotting industry data: {e}", exc_info=True)
html_parts.append("<p>Error displaying follower industry data.</p>")
plot_industry_dist = None
finally:
if fig_industry_local and fig_industry_local is not plt:
plt.close(fig_industry_local)
html_parts.append("</div>")
follower_html_output = "\n".join(html_parts)
return follower_html_output, plot_monthly_gains, plot_seniority_dist, plot_industry_dist
# --- UI GENERATION LOGIC FOR ANALYTICS TAB ---
def create_analytics_plot_panel(label, plot_id_str):
"""
Creates a Gradio Column representing a single plot panel.
This panel contains the plot and its associated action buttons.
Returns:
panel_col (gr.Column): The main column for this plot panel.
plot_component (gr.Plot): The plot display area.
bomb_button (gr.Button): Insights button.
explore_button (gr.Button): Explore/Zoom button.
formula_button (gr.Button): Formula button.
"""
with gr.Column() as panel_col: # This will be the component that gets hidden/shown by explore
with gr.Row(equal_height=False, variant="panel"): # Removed variant="compact" as it might affect spacing
plot_component = gr.Plot(label=label, scale=8) # The label here is for Gradio, not the Matplotlib title
with gr.Column(scale=2, min_width=100): # min_width for button column
bomb_button = gr.Button(BOMB_ICON, variant="secondary", size="sm", elem_id=f"bomb_{plot_id_str}")
explore_button = gr.Button(EXPLORE_ICON, variant="secondary", size="sm", elem_id=f"explore_{plot_id_str}")
formula_button = gr.Button(FORMULA_ICON, variant="secondary", size="sm", elem_id=f"formula_{plot_id_str}")
return panel_col, plot_component, bomb_button, explore_button, formula_button
def build_analytics_tab_plot_area(plot_configs):
"""
Builds the main plot area for the Analytics tab, arranging plot panels into rows of two.
Returns a dictionary of plot UI objects.
"""
logging.info(f"Building plot area for {len(plot_configs)} analytics plots.")
plot_ui_objects = {}
current_section_title = ""
i = 0
while i < len(plot_configs):
config1 = plot_configs[i]
# Start a new section if necessary
if config1["section"] != current_section_title:
current_section_title = config1["section"]
gr.Markdown(f"### {current_section_title}") # Section title
# Create a new row for each pair of plots or a single plot
with gr.Row(equal_height=False): # A new row for one or two plots
# Process the first plot of the potential pair
panel_col1, plot_comp1, bomb_btn1, explore_btn1, formula_btn1 = \
create_analytics_plot_panel(config1["label"], config1["id"])
plot_ui_objects[config1["id"]] = {
"plot_component": plot_comp1, "bomb_button": bomb_btn1,
"explore_button": explore_btn1, "formula_button": formula_btn1,
"label": config1["label"], "panel_component": panel_col1 # panel_col1 is the gr.Column from create_analytics_plot_panel
}
logging.debug(f"Created UI panel for plot_id: {config1['id']}")
i += 1
# Check if there's a second plot for this row and if it's in the same section
if i < len(plot_configs):
config2 = plot_configs[i]
if config2["section"] == current_section_title:
panel_col2, plot_comp2, bomb_btn2, explore_btn2, formula_btn2 = \
create_analytics_plot_panel(config2["label"], config2["id"])
plot_ui_objects[config2["id"]] = {
"plot_component": plot_comp2, "bomb_button": bomb_btn2,
"explore_button": explore_btn2, "formula_button": formula_btn2,
"label": config2["label"], "panel_component": panel_col2
}
logging.debug(f"Created UI panel for plot_id: {config2['id']} in same row")
i += 1
# If config2 is in a new section, the row ends with config1.
# config2 will be processed as the first plot in a new row in the next iteration.
# Row ends here. Next iteration will start a new gr.Row if needed.
logging.info(f"Finished building plot area. Total plot objects: {len(plot_ui_objects)}")
if len(plot_ui_objects) != len(plot_configs):
logging.error(f"MISMATCH: Expected {len(plot_configs)} plot objects, but created {len(plot_ui_objects)}.")
return plot_ui_objects