LinkedinMonitor / services /analytics_tab_module.py
GuglielmoTor's picture
Rename services/analytics_handlers.py to services/analytics_tab_module.py
1b37007 verified
# analytics_tab_module.py
import gradio as gr
import pandas as pd
import logging
import time
from datetime import datetime, timedelta
import numpy as np
from collections import OrderedDict, defaultdict
import asyncio
import matplotlib # Keep this if create_placeholder_plot or other plot fns need it
matplotlib.use('Agg') # Keep if necessary for plotting functions
import matplotlib.pyplot as plt # Keep if necessary
# It's assumed that PLOT_CONFIGS is specific to this analytics tab.
# If it's used elsewhere, it might need to be passed in or imported from a central config.
PLOT_CONFIGS = [
{"label": "Numero di Follower nel Tempo", "id": "followers_count", "section": "Dinamiche dei Follower"},
{"label": "Tasso di Crescita Follower", "id": "followers_growth_rate", "section": "Dinamiche dei Follower"},
{"label": "Follower per Località", "id": "followers_by_location", "section": "Demografia Follower"},
{"label": "Follower per Ruolo (Funzione)", "id": "followers_by_role", "section": "Demografia Follower"},
{"label": "Follower per Settore", "id": "followers_by_industry", "section": "Demografia Follower"},
{"label": "Follower per Anzianità", "id": "followers_by_seniority", "section": "Demografia Follower"},
{"label": "Tasso di Engagement nel Tempo", "id": "engagement_rate", "section": "Approfondimenti Performance Post"},
{"label": "Copertura nel Tempo", "id": "reach_over_time", "section": "Approfondimenti Performance Post"},
{"label": "Visualizzazioni nel Tempo", "id": "impressions_over_time", "section": "Approfondimenti Performance Post"},
{"label": "Reazioni (Like) nel Tempo", "id": "likes_over_time", "section": "Approfondimenti Performance Post"},
{"label": "Click nel Tempo", "id": "clicks_over_time", "section": "Engagement Dettagliato Post nel Tempo"},
{"label": "Condivisioni nel Tempo", "id": "shares_over_time", "section": "Engagement Dettagliato Post nel Tempo"},
{"label": "Commenti nel Tempo", "id": "comments_over_time", "section": "Engagement Dettagliato Post nel Tempo"},
{"label": "Ripartizione Commenti per Sentiment", "id": "comments_sentiment", "section": "Engagement Dettagliato Post nel Tempo"},
{"label": "Frequenza Post", "id": "post_frequency_cs", "section": "Analisi Strategia Contenuti"},
{"label": "Ripartizione Contenuti per Formato", "id": "content_format_breakdown_cs", "section": "Analisi Strategia Contenuti"},
{"label": "Ripartizione Contenuti per Argomenti", "id": "content_topic_breakdown_cs", "section": "Analisi Strategia Contenuti"},
{"label": "Volume Menzioni nel Tempo (Dettaglio)", "id": "mention_analysis_volume", "section": "Analisi Menzioni (Dettaglio)"},
{"label": "Ripartizione Menzioni per Sentiment (Dettaglio)", "id": "mention_analysis_sentiment", "section": "Analisi Menzioni (Dettaglio)"}
]
# IMPORTANT: Review if 'mention_analysis_volume' and 'mention_analysis_sentiment' plots
# can still be generated without the dedicated mentions data processing.
# If not, they should also be removed from plot_configs.
# For now, I am assuming they might draw from a general data pool in token_state.
assert len(PLOT_CONFIGS) == 19, "Mancata corrispondenza in PLOT_CONFIGS e grafici attesi. (If mentions plots were removed, adjust this number)"
UNIQUE_ORDERED_SECTIONS = list(OrderedDict.fromkeys(pc["section"] for pc in PLOT_CONFIGS))
NUM_UNIQUE_SECTIONS = len(UNIQUE_ORDERED_SECTIONS)
class AnalyticsTab:
def __init__(self, token_state, chat_histories_st, current_chat_plot_id_st, plot_data_for_chatbot_st,
# External dependencies (functions, data, icons)
plot_id_to_formula_map, plot_formulas_data, icons,
fn_build_plot_area, fn_update_plot_figures, fn_create_placeholder_plot,
fn_get_initial_insight, fn_generate_llm_response):
# Shared Gradio states passed from the main app
self.token_state = token_state
self.chat_histories_st = chat_histories_st
self.current_chat_plot_id_st = current_chat_plot_id_st
self.plot_data_for_chatbot_st = plot_data_for_chatbot_st
# Store external dependencies
self.PLOT_ID_TO_FORMULA_KEY_MAP = plot_id_to_formula_map
self.PLOT_FORMULAS = plot_formulas_data
self.BOMB_ICON = icons['bomb']
self.EXPLORE_ICON = icons['explore']
self.FORMULA_ICON = icons['formula']
self.ACTIVE_ICON = icons['active']
self.build_analytics_tab_plot_area = fn_build_plot_area
self.update_analytics_plots_figures = fn_update_plot_figures
self.create_placeholder_plot = fn_create_placeholder_plot
self.get_initial_insight_prompt_and_suggestions = fn_get_initial_insight
self.generate_llm_response = fn_generate_llm_response
# Internal Gradio states for this tab
self.active_panel_action_state = gr.State(None)
self.explored_plot_id_state = gr.State(None)
# To store UI objects created by build_analytics_tab_plot_area
self.plot_ui_objects = {}
self.section_titles_map = {}
# UI components that will be created in create_tab_ui
self.analytics_status_md = None
self.date_filter_selector = None
self.custom_start_date_picker = None
self.custom_end_date_picker = None
self.apply_filter_btn = None
self.plots_area_col = None
self.global_actions_column_ui = None
self.insights_chatbot_ui = None
self.insights_chat_input_ui = None
self.insights_suggestions_row_ui = None
self.insights_suggestion_1_btn = None
self.insights_suggestion_2_btn = None
self.insights_suggestion_3_btn = None
self.formula_display_markdown_ui = None
self.formula_close_hint_md = None
# Lists for Gradio callback outputs, will be populated after UI creation
self.graph_refresh_outputs_list = []
self.action_panel_outputs_list = []
self.explore_outputs_list = []
def _toggle_custom_date_pickers(self, selection):
is_custom = selection == "Intervallo Personalizzato"
return gr.update(visible=is_custom), gr.update(visible=is_custom)
async def _handle_panel_action(
self, plot_id_clicked: str, action_type: str, current_active_action_from_state: dict,
current_chat_histories: dict, current_chat_plot_id: str,
current_plot_data_for_chatbot: dict, current_explored_plot_id: str
):
logging.info(f"Panel Action: '{action_type}' for plot '{plot_id_clicked}'. Active: {current_active_action_from_state}, Explored: {current_explored_plot_id}")
clicked_plot_config = next((p for p in PLOT_CONFIGS if p["id"] == plot_id_clicked), None)
if not clicked_plot_config:
logging.error(f"Config not found for plot_id {plot_id_clicked}")
num_plots = len(PLOT_CONFIGS)
error_list_len = 15 + (4 * num_plots) + NUM_UNIQUE_SECTIONS
error_list = [gr.update()] * error_list_len
# Fill specific indices if needed, matching the expected output structure
error_list[11] = current_active_action_from_state # active_panel_action_state
error_list[12] = current_chat_plot_id # current_chat_plot_id_st
error_list[13] = current_chat_histories # chat_histories_st
error_list[14] = current_explored_plot_id # explored_plot_id_state
return error_list
clicked_plot_label = clicked_plot_config["label"]
clicked_plot_section = clicked_plot_config["section"]
hypothetical_new_active_state = {"plot_id": plot_id_clicked, "type": action_type}
is_toggling_off = current_active_action_from_state == hypothetical_new_active_state
action_col_visible_update = gr.update(visible=False)
insights_chatbot_visible_update, insights_chat_input_visible_update, insights_suggestions_row_visible_update = gr.update(visible=False), gr.update(visible=False), gr.update(visible=False)
formula_display_visible_update = gr.update(visible=False)
formula_close_hint_visible_update = gr.update(visible=False)
chatbot_content_update, s1_upd, s2_upd, s3_upd, formula_content_update = gr.update(), gr.update(), gr.update(), gr.update(), gr.update()
new_active_action_state_to_set = None
new_current_chat_plot_id = current_chat_plot_id # Preserve by default
updated_chat_histories = current_chat_histories # Preserve by default
new_explored_plot_id_to_set = current_explored_plot_id # Preserve by default
generated_panel_vis_updates = []
generated_bomb_btn_updates = []
generated_formula_btn_updates = []
generated_explore_btn_updates = []
section_title_vis_updates = [gr.update()] * NUM_UNIQUE_SECTIONS
if is_toggling_off:
new_active_action_state_to_set = None
action_col_visible_update = gr.update(visible=False)
logging.info(f"Toggling OFF panel {action_type} for {plot_id_clicked}.")
for _ in PLOT_CONFIGS:
generated_bomb_btn_updates.append(gr.update(value=self.BOMB_ICON))
generated_formula_btn_updates.append(gr.update(value=self.FORMULA_ICON))
if current_explored_plot_id: # If an explore view was active, restore it
explored_cfg = next((p for p in PLOT_CONFIGS if p["id"] == current_explored_plot_id), None)
explored_sec = explored_cfg["section"] if explored_cfg else None
for i, sec_name in enumerate(UNIQUE_ORDERED_SECTIONS):
section_title_vis_updates[i] = gr.update(visible=(sec_name == explored_sec))
for cfg in PLOT_CONFIGS:
is_exp = (cfg["id"] == current_explored_plot_id)
generated_panel_vis_updates.append(gr.update(visible=is_exp))
generated_explore_btn_updates.append(gr.update(value=self.ACTIVE_ICON if is_exp else self.EXPLORE_ICON))
else: # No explore view, all plots visible
for i in range(NUM_UNIQUE_SECTIONS):
section_title_vis_updates[i] = gr.update(visible=True)
for _ in PLOT_CONFIGS:
generated_panel_vis_updates.append(gr.update(visible=True))
generated_explore_btn_updates.append(gr.update(value=self.EXPLORE_ICON))
if action_type == "insights": # Specifically when closing insights chat
new_current_chat_plot_id = None # Clear the chat context
else: # Toggling ON a panel action
new_active_action_state_to_set = hypothetical_new_active_state
action_col_visible_update = gr.update(visible=True)
new_explored_plot_id_to_set = None # Cancel any active explore view
logging.info(f"Toggling ON panel {action_type} for {plot_id_clicked}. Cancelling explore view if any.")
# Show only the clicked plot and its section title
for i, sec_name in enumerate(UNIQUE_ORDERED_SECTIONS):
section_title_vis_updates[i] = gr.update(visible=(sec_name == clicked_plot_section))
for cfg in PLOT_CONFIGS:
generated_panel_vis_updates.append(gr.update(visible=(cfg["id"] == plot_id_clicked)))
generated_explore_btn_updates.append(gr.update(value=self.EXPLORE_ICON)) # Reset all explore buttons
# Update bomb and formula button icons based on the new active action
for cfg_btn in PLOT_CONFIGS:
is_active_insights = new_active_action_state_to_set == {"plot_id": cfg_btn["id"], "type": "insights"}
is_active_formula = new_active_action_state_to_set == {"plot_id": cfg_btn["id"], "type": "formula"}
generated_bomb_btn_updates.append(gr.update(value=self.ACTIVE_ICON if is_active_insights else self.BOMB_ICON))
generated_formula_btn_updates.append(gr.update(value=self.ACTIVE_ICON if is_active_formula else self.FORMULA_ICON))
if action_type == "insights":
insights_chatbot_visible_update = gr.update(visible=True)
insights_chat_input_visible_update = gr.update(visible=True)
insights_suggestions_row_visible_update = gr.update(visible=True)
new_current_chat_plot_id = plot_id_clicked # Set chat context to this plot
history = current_chat_histories.get(plot_id_clicked, [])
summary = current_plot_data_for_chatbot.get(plot_id_clicked, f"No summary available for '{clicked_plot_label}'.")
if not history: # First time opening chat for this plot
prompt, sugg = self.get_initial_insight_prompt_and_suggestions(plot_id_clicked, clicked_plot_label, summary)
llm_history_for_api = [{"role": "user", "content": prompt}] # API expects list of dicts
resp = await self.generate_llm_response(prompt, plot_id_clicked, clicked_plot_label, llm_history_for_api, summary)
history = [{"role": "assistant", "content": resp}] # Gradio expects list of tuples or specific dicts for Chatbot
updated_chat_histories = {**current_chat_histories, plot_id_clicked: history}
else: # Re-opening chat, just get new suggestions if any
_, sugg = self.get_initial_insight_prompt_and_suggestions(plot_id_clicked, clicked_plot_label, summary)
chatbot_content_update = gr.update(value=history)
s1_upd = gr.update(value=sugg[0] if sugg and len(sugg) > 0 else "N/A")
s2_upd = gr.update(value=sugg[1] if sugg and len(sugg) > 1 else "N/A")
s3_upd = gr.update(value=sugg[2] if sugg and len(sugg) > 2 else "N/A")
elif action_type == "formula":
formula_display_visible_update = gr.update(visible=True)
formula_close_hint_visible_update = gr.update(visible=True)
formula_key = self.PLOT_ID_TO_FORMULA_KEY_MAP.get(plot_id_clicked)
formula_text = f"**Formula/Methodology for: {clicked_plot_label}** (ID: `{plot_id_clicked}`)\n\n"
if formula_key and formula_key in self.PLOT_FORMULAS:
formula_data = self.PLOT_FORMULAS[formula_key]
formula_text += f"### {formula_data['title']}\n\n{formula_data['description']}\n\n**Calculation:**\n"
formula_text += "\n".join([f"- {step}" for step in formula_data['calculation_steps']])
else:
formula_text += "(No detailed formula information found.)"
formula_content_update = gr.update(value=formula_text)
new_current_chat_plot_id = None # Clear chat context if formula is opened
# Order of updates must match self.action_panel_outputs_list
final_updates = [
action_col_visible_update, # global_actions_column_ui
insights_chatbot_visible_update, # insights_chatbot_ui (visibility)
chatbot_content_update, # insights_chatbot_ui (content)
insights_chat_input_visible_update, # insights_chat_input_ui
insights_suggestions_row_visible_update, # insights_suggestions_row_ui
s1_upd, s2_upd, s3_upd, # suggestion buttons
formula_display_visible_update, # formula_display_markdown_ui (visibility)
formula_content_update, # formula_display_markdown_ui (content)
formula_close_hint_visible_update, # formula_close_hint_md
new_active_action_state_to_set, # active_panel_action_state
new_current_chat_plot_id, # current_chat_plot_id_st
updated_chat_histories, # chat_histories_st
new_explored_plot_id_to_set # explored_plot_id_state
]
final_updates.extend(generated_panel_vis_updates) # Plot panel visibilities
final_updates.extend(generated_bomb_btn_updates) # Bomb button icons
final_updates.extend(generated_formula_btn_updates) # Formula button icons
final_updates.extend(generated_explore_btn_updates) # Explore button icons
final_updates.extend(section_title_vis_updates) # Section title visibilities
logging.debug(f"handle_panel_action returning {len(final_updates)} updates. Expected {15 + 4*len(PLOT_CONFIGS) + NUM_UNIQUE_SECTIONS}.")
return final_updates
async def _handle_chat_message_submission(self, user_message: str, current_plot_id: str, chat_histories: dict, current_plot_data_for_chatbot: dict):
if not current_plot_id or not user_message.strip():
current_history_for_plot = chat_histories.get(current_plot_id, [])
# Ensure history is in the format Gradio Chatbot expects (list of lists/tuples or specific dicts)
# Assuming it's already [{"role": "user/assistant", "content": "..."}]
yield current_history_for_plot, gr.update(value=""), chat_histories
return
clicked_plot_config = next((p for p in PLOT_CONFIGS if p["id"] == current_plot_id), None)
plot_label = clicked_plot_config["label"] if clicked_plot_config else "Selected Plot"
summary_data = current_plot_data_for_chatbot.get(current_plot_id, f"No summary available for '{plot_label}'.")
# Ensure history is a list of dicts {"role": ..., "content": ...} for the API
history_for_api = chat_histories.get(current_plot_id, []).copy() # Get a mutable copy
if not isinstance(history_for_api, list): history_for_api = [] # Should not happen if initialized correctly
history_for_api.append({"role": "user", "content": user_message})
# Update Gradio chatbot UI immediately with user message
# Gradio Chatbot expects list of (user_msg, assistant_msg) tuples or specific dicts.
# If current_chat_histories stores [{"role": "user", "content": "..."}], convert for display if needed,
# or ensure generate_llm_response and initial prompt also use this dict format.
# For simplicity, let's assume chat_histories_st stores a list of such dicts.
current_display_history = history_for_api.copy() # This is what will be displayed
yield current_display_history, gr.update(value=""), chat_histories # Update UI, clear input, pass original histories
assistant_response = await self.generate_llm_response(user_message, current_plot_id, plot_label, history_for_api, summary_data)
history_for_api.append({"role": "assistant", "content": assistant_response})
updated_chat_histories = {**chat_histories, current_plot_id: history_for_api}
current_display_history.append({"role": "assistant", "content": assistant_response})
yield current_display_history, "", updated_chat_histories
async def _handle_suggested_question_click(self, suggestion_text: str, current_plot_id: str, chat_histories: dict, current_plot_data_for_chatbot: dict):
if not current_plot_id or not suggestion_text.strip() or suggestion_text == "N/A":
current_history_for_plot = chat_histories.get(current_plot_id, [])
yield current_history_for_plot, gr.update(value=""), chat_histories
return
# Use async for to stream updates from _handle_chat_message_submission
async for update_chunk in self._handle_chat_message_submission(suggestion_text, current_plot_id, chat_histories, current_plot_data_for_chatbot):
yield update_chunk
def _handle_explore_click(self, plot_id_clicked, current_explored_plot_id_from_state, current_active_panel_action_state):
logging.info(f"Explore Click: Plot '{plot_id_clicked}'. Current Explored: {current_explored_plot_id_from_state}. Active Panel: {current_active_panel_action_state}")
num_plots = len(PLOT_CONFIGS)
if not self.plot_ui_objects: # Check if plot UI objects are populated
logging.error("plot_ui_objects not populated for handle_explore_click.")
# Prepare a list of gr.update() of the correct length
error_list_len = 4 + (4 * num_plots) + NUM_UNIQUE_SECTIONS
error_list = [gr.update()] * error_list_len
error_list[0] = current_explored_plot_id_from_state # explored_plot_id_state
error_list[2] = current_active_panel_action_state # active_panel_action_state
return error_list
new_explored_id_to_set = None
is_toggling_off_explore = (plot_id_clicked == current_explored_plot_id_from_state)
action_col_upd = gr.update() # Default to no change
new_active_panel_state_upd = current_active_panel_action_state # Default to no change
formula_hint_upd = gr.update(visible=False) # Hide by default, only shown by formula panel
panel_vis_updates = []
explore_btns_updates = []
bomb_btns_updates = [gr.update()] * num_plots # Default to no change for bomb/formula unless panel closes
formula_btns_updates = [gr.update()] * num_plots
section_title_vis_updates = [gr.update()] * NUM_UNIQUE_SECTIONS
clicked_cfg = next((p for p in PLOT_CONFIGS if p["id"] == plot_id_clicked), None)
section_of_clicked_plot = clicked_cfg["section"] if clicked_cfg else None
if is_toggling_off_explore:
new_explored_id_to_set = None # Clear explore state
logging.info(f"Stopping explore for {plot_id_clicked}. All plots/sections to be visible.")
for i in range(NUM_UNIQUE_SECTIONS):
section_title_vis_updates[i] = gr.update(visible=True)
for _ in PLOT_CONFIGS:
panel_vis_updates.append(gr.update(visible=True))
explore_btns_updates.append(gr.update(value=self.EXPLORE_ICON))
# If an action panel was open, it remains open. Its visibility is tied to active_panel_action_state.
# Bomb and formula buttons don't change unless an action panel is closed.
else: # Exploring a new plot or switching explore
new_explored_id_to_set = plot_id_clicked
logging.info(f"Exploring {plot_id_clicked}. Hiding other plots/sections.")
for i, sec_name in enumerate(UNIQUE_ORDERED_SECTIONS):
section_title_vis_updates[i] = gr.update(visible=(sec_name == section_of_clicked_plot))
for cfg in PLOT_CONFIGS:
is_target_plot = (cfg["id"] == new_explored_id_to_set)
panel_vis_updates.append(gr.update(visible=is_target_plot))
explore_btns_updates.append(gr.update(value=self.ACTIVE_ICON if is_target_plot else self.EXPLORE_ICON))
if current_active_panel_action_state: # If an insights/formula panel was open, close it
logging.info("Closing active insight/formula panel due to explore click.")
action_col_upd = gr.update(visible=False)
new_active_panel_state_upd = None # This will be set to the state
# Reset bomb and formula buttons to their default icons
bomb_btns_updates = [gr.update(value=self.BOMB_ICON) for _ in PLOT_CONFIGS]
formula_btns_updates = [gr.update(value=self.FORMULA_ICON) for _ in PLOT_CONFIGS]
formula_hint_upd = gr.update(visible=False) # Ensure hint is hidden
# Order of updates must match self.explore_outputs_list
final_explore_updates = [
new_explored_id_to_set, # explored_plot_id_state
action_col_upd, # global_actions_column_ui
new_active_panel_state_upd, # active_panel_action_state
formula_hint_upd # formula_close_hint_md
]
final_explore_updates.extend(panel_vis_updates) # Plot panel visibilities
final_explore_updates.extend(explore_btns_updates) # Explore button icons
final_explore_updates.extend(bomb_btns_updates) # Bomb button icons
final_explore_updates.extend(formula_btns_updates) # Formula button icons
final_explore_updates.extend(section_title_vis_updates) # Section title visibilities
logging.debug(f"handle_explore_click returning {len(final_explore_updates)} updates. Expected {4 + 4*len(PLOT_CONFIGS) + NUM_UNIQUE_SECTIONS}.")
return final_explore_updates
def _create_panel_action_handler(self, p_id, action_type_str):
# This wrapper is needed because Gradio's .click() fn doesn't easily pass extra args fixed at definition time
# without using lambdas that might have late binding issues in loops, or functools.partial.
# An inner async def is a clean way for this specific pattern with async handlers.
async def _handler(curr_active_val, curr_chats_val, curr_chat_pid, curr_plot_data, curr_explored_id):
return await self._handle_panel_action(p_id, action_type_str, curr_active_val, curr_chats_val, curr_chat_pid, curr_plot_data, curr_explored_id)
return _handler
async def _refresh_analytics_graphs_ui(self, current_token_state_val, date_filter_val, custom_start_val, custom_end_val, current_chat_histories_val):
logging.info("Refreshing analytics graph UI elements and resetting actions/chat (within module).")
start_time = time.time()
plot_gen_results = self.update_analytics_plots_figures(current_token_state_val, date_filter_val, custom_start_val, custom_end_val, PLOT_CONFIGS)
status_msg, gen_figs, new_summaries_for_chatbot = plot_gen_results[0], plot_gen_results[1:-1], plot_gen_results[-1]
all_updates = []
# 1. Status Markdown
all_updates.append(status_msg) # For self.analytics_status_md
# 2. Plot components
if len(gen_figs) == len(PLOT_CONFIGS):
all_updates.extend(gen_figs)
else:
logging.error(f"Mismatch in generated figures ({len(gen_figs)}) and plot_configs ({len(PLOT_CONFIGS)})")
all_updates.extend([self.create_placeholder_plot("Error", f"Figura mancante {i}") for i in range(len(PLOT_CONFIGS))])
# 3. UI Resets for Action Panel (9 components)
all_updates.extend([
gr.update(visible=False), # global_actions_column_ui
gr.update(value=[], visible=False), # insights_chatbot_ui (content & visibility)
gr.update(value="", visible=False), # insights_chat_input_ui
gr.update(visible=False), # insights_suggestions_row_ui
gr.update(value="Suggerimento 1"), gr.update(value="Suggerimento 2"), gr.update(value="Suggerimento 3"), # suggestion_btns
gr.update(value="I dettagli sulla formula/metodologia appariranno qui.", visible=False), # formula_display_markdown_ui
gr.update(visible=False) # formula_close_hint_md
])
# 4. State Resets (4 states)
all_updates.extend([
None, # active_panel_action_state (reset)
None, # current_chat_plot_id_st (reset)
current_chat_histories_val, # chat_histories_st (pass through, or {} to reset all chats) - original code implies reset with {}
new_summaries_for_chatbot # plot_data_for_chatbot_st (update with new summaries)
])
# If chat_histories_st should be fully reset on graph refresh:
# all_updates[-2] = {} # This would reset all chat histories
# 5. Plot-specific UI Resets (4 components per plot)
for _ in PLOT_CONFIGS:
all_updates.extend([
gr.update(value=self.BOMB_ICON), # bomb_button
gr.update(value=self.FORMULA_ICON), # formula_button
gr.update(value=self.EXPLORE_ICON), # explore_button
gr.update(visible=True) # panel_component (plot visibility itself)
])
# 6. Explored Plot ID State Reset (1 state)
all_updates.append(None) # explored_plot_id_state (reset)
# 7. Section Title Visibilities
all_updates.extend([gr.update(visible=True)] * NUM_UNIQUE_SECTIONS)
end_time = time.time()
logging.info(f"Analytics graph refresh (module) took {end_time - start_time:.2f} seconds.")
expected_len = 1 + len(PLOT_CONFIGS) + 9 + 4 + (4 * len(PLOT_CONFIGS)) + 1 + NUM_UNIQUE_SECTIONS
logging.info(f"Prepared {len(all_updates)} updates for graph refresh. Expected {expected_len}.")
if len(all_updates) != expected_len:
logging.error(f"Output length mismatch in _refresh_analytics_graphs_ui: got {len(all_updates)}, expected {expected_len}")
return tuple(all_updates)
def _define_callback_outputs(self):
# This method populates the output lists for various callbacks.
# It MUST be called after all UI components of this tab are created.
# Outputs for _refresh_analytics_graphs_ui
self.graph_refresh_outputs_list.append(self.analytics_status_md)
self.graph_refresh_outputs_list.extend([self.plot_ui_objects.get(pc["id"], {}).get("plot_component", gr.update()) for pc in PLOT_CONFIGS])
self.graph_refresh_outputs_list.extend([
self.global_actions_column_ui, self.insights_chatbot_ui, self.insights_chat_input_ui,
self.insights_suggestions_row_ui, self.insights_suggestion_1_btn, self.insights_suggestion_2_btn,
self.insights_suggestion_3_btn, self.formula_display_markdown_ui, self.formula_close_hint_md
])
self.graph_refresh_outputs_list.extend([
self.active_panel_action_state, self.current_chat_plot_id_st,
self.chat_histories_st, self.plot_data_for_chatbot_st
])
for pc in PLOT_CONFIGS:
pid = pc["id"]
self.graph_refresh_outputs_list.extend([
self.plot_ui_objects.get(pid, {}).get("bomb_button", gr.update()),
self.plot_ui_objects.get(pid, {}).get("formula_button", gr.update()),
self.plot_ui_objects.get(pid, {}).get("explore_button", gr.update()),
self.plot_ui_objects.get(pid, {}).get("panel_component", gr.update())
])
self.graph_refresh_outputs_list.append(self.explored_plot_id_state)
self.graph_refresh_outputs_list.extend([self.section_titles_map.get(s_name, gr.update()) for s_name in UNIQUE_ORDERED_SECTIONS])
# Outputs for _handle_panel_action
self.action_panel_outputs_list.extend([
self.global_actions_column_ui, self.insights_chatbot_ui, self.insights_chatbot_ui, # Chatbot visibility and content
self.insights_chat_input_ui, self.insights_suggestions_row_ui,
self.insights_suggestion_1_btn, self.insights_suggestion_2_btn, self.insights_suggestion_3_btn,
self.formula_display_markdown_ui, self.formula_display_markdown_ui, # Formula visibility and content
self.formula_close_hint_md
])
self.action_panel_outputs_list.extend([
self.active_panel_action_state, self.current_chat_plot_id_st,
self.chat_histories_st, self.explored_plot_id_state
])
self.action_panel_outputs_list.extend([self.plot_ui_objects.get(pc["id"], {}).get("panel_component", gr.update()) for pc in PLOT_CONFIGS])
self.action_panel_outputs_list.extend([self.plot_ui_objects.get(pc["id"], {}).get("bomb_button", gr.update()) for pc in PLOT_CONFIGS])
self.action_panel_outputs_list.extend([self.plot_ui_objects.get(pc["id"], {}).get("formula_button", gr.update()) for pc in PLOT_CONFIGS])
self.action_panel_outputs_list.extend([self.plot_ui_objects.get(pc["id"], {}).get("explore_button", gr.update()) for pc in PLOT_CONFIGS])
self.action_panel_outputs_list.extend([self.section_titles_map.get(s_name, gr.update()) for s_name in UNIQUE_ORDERED_SECTIONS])
# Outputs for _handle_explore_click
self.explore_outputs_list.extend([
self.explored_plot_id_state, self.global_actions_column_ui,
self.active_panel_action_state, self.formula_close_hint_md
])
self.explore_outputs_list.extend([self.plot_ui_objects.get(pc["id"], {}).get("panel_component", gr.update()) for pc in PLOT_CONFIGS])
self.explore_outputs_list.extend([self.plot_ui_objects.get(pc["id"], {}).get("explore_button", gr.update()) for pc in PLOT_CONFIGS])
self.explore_outputs_list.extend([self.plot_ui_objects.get(pc["id"], {}).get("bomb_button", gr.update()) for pc in PLOT_CONFIGS]) # For resetting if panel closes
self.explore_outputs_list.extend([self.plot_ui_objects.get(pc["id"], {}).get("formula_button", gr.update()) for pc in PLOT_CONFIGS])# For resetting if panel closes
self.explore_outputs_list.extend([self.section_titles_map.get(s_name, gr.update()) for s_name in UNIQUE_ORDERED_SECTIONS])
def _setup_callbacks(self):
# Apply filter button
graph_refresh_inputs = [
self.token_state, self.date_filter_selector,
self.custom_start_date_picker, self.custom_end_date_picker,
self.chat_histories_st # Pass the state object itself
]
self.apply_filter_btn.click(
fn=self._refresh_analytics_graphs_ui,
inputs=graph_refresh_inputs,
outputs=self.graph_refresh_outputs_list,
show_progress="full",
api_name="refresh_analytics_graphs_module"
)
# Panel action buttons (bomb, formula, explore)
action_click_inputs = [
self.active_panel_action_state, self.chat_histories_st,
self.current_chat_plot_id_st, self.plot_data_for_chatbot_st,
self.explored_plot_id_state
]
explore_click_inputs = [self.explored_plot_id_state, self.active_panel_action_state]
for config_item in PLOT_CONFIGS:
plot_id = config_item["id"]
if plot_id in self.plot_ui_objects:
ui_obj = self.plot_ui_objects[plot_id]
if ui_obj.get("bomb_button"):
ui_obj["bomb_button"].click(
fn=self._create_panel_action_handler(plot_id, "insights"),
inputs=action_click_inputs,
outputs=self.action_panel_outputs_list,
api_name=f"action_insights_{plot_id}_module"
)
if ui_obj.get("formula_button"):
ui_obj["formula_button"].click(
fn=self._create_panel_action_handler(plot_id, "formula"),
inputs=action_click_inputs,
outputs=self.action_panel_outputs_list,
api_name=f"action_formula_{plot_id}_module"
)
if ui_obj.get("explore_button"):
# Lambda is okay here as p_id is captured correctly due to loop variable usage
ui_obj["explore_button"].click(
fn=lambda current_explored_val, current_active_panel_val, p_id=plot_id: self._handle_explore_click(p_id, current_explored_val, current_active_panel_val),
inputs=explore_click_inputs,
outputs=self.explore_outputs_list,
api_name=f"action_explore_{plot_id}_module"
)
else:
logging.warning(f"UI object for plot_id '{plot_id}' not found for click handlers in module.")
# Chat submission
chat_submission_outputs = [self.insights_chatbot_ui, self.insights_chat_input_ui, self.chat_histories_st]
chat_submission_inputs = [
self.insights_chat_input_ui, self.current_chat_plot_id_st,
self.chat_histories_st, self.plot_data_for_chatbot_st
]
self.insights_chat_input_ui.submit(
fn=self._handle_chat_message_submission,
inputs=chat_submission_inputs,
outputs=chat_submission_outputs,
api_name="submit_chat_message_module"
)
# Suggested questions
suggestion_click_inputs_base = [
self.current_chat_plot_id_st, self.chat_histories_st, self.plot_data_for_chatbot_st
]
self.insights_suggestion_1_btn.click(
fn=self._handle_suggested_question_click,
inputs=[self.insights_suggestion_1_btn] + suggestion_click_inputs_base, # Pass the button itself for its value
outputs=chat_submission_outputs,
api_name="click_suggestion_1_module"
)
self.insights_suggestion_2_btn.click(
fn=self._handle_suggested_question_click,
inputs=[self.insights_suggestion_2_btn] + suggestion_click_inputs_base,
outputs=chat_submission_outputs,
api_name="click_suggestion_2_module"
)
self.insights_suggestion_3_btn.click(
fn=self._handle_suggested_question_click,
inputs=[self.insights_suggestion_3_btn] + suggestion_click_inputs_base,
outputs=chat_submission_outputs,
api_name="click_suggestion_3_module"
)
def create_tab_ui(self):
# This method is called by the main app to build the UI for this tab
with gr.TabItem("2️⃣ Analisi Grafici", id="tab_analytics_module"): # Changed id to avoid conflict if old tab exists temporarily
gr.Markdown("## 📈 Analisi Performance LinkedIn")
gr.Markdown("Seleziona un intervallo di date per i grafici. Clicca i pulsanti (💣 Insights, ƒ Formula, 🧭 Esplora) su un grafico per azioni.")
self.analytics_status_md = gr.Markdown("Stato analisi grafici...")
with gr.Row():
self.date_filter_selector = gr.Radio(
["Sempre", "Ultimi 7 Giorni", "Ultimi 30 Giorni", "Intervallo Personalizzato"],
label="Seleziona Intervallo Date per Grafici", value="Sempre", scale=3
)
with gr.Column(scale=2):
self.custom_start_date_picker = gr.DateTime(label="Data Inizio", visible=False, include_time=False, type="datetime")
self.custom_end_date_picker = gr.DateTime(label="Data Fine", visible=False, include_time=False, type="datetime")
self.apply_filter_btn = gr.Button("🔍 Applica Filtro & Aggiorna Grafici", variant="primary")
self.date_filter_selector.change(
fn=self._toggle_custom_date_pickers,
inputs=[self.date_filter_selector],
outputs=[self.custom_start_date_picker, self.custom_end_date_picker]
)
# Plot area and actions column
with gr.Row(equal_height=False):
with gr.Column(scale=8) as self.plots_area_col:
# Dynamically build plot UI objects and section titles
ui_elements_tuple = self.build_analytics_tab_plot_area(PLOT_CONFIGS) # Call injected function
if isinstance(ui_elements_tuple, tuple) and len(ui_elements_tuple) == 2:
self.plot_ui_objects, self.section_titles_map = ui_elements_tuple
# Verify section_titles_map completeness (optional, good for debugging)
if not all(sec_name in self.section_titles_map for sec_name in UNIQUE_ORDERED_SECTIONS):
logging.error("section_titles_map from build_analytics_tab_plot_area is incomplete in module.")
# Create placeholders if missing, to prevent errors with output lists
for sec_name in UNIQUE_ORDERED_SECTIONS:
if sec_name not in self.section_titles_map:
logging.warning(f"Creating fallback Markdown for missing section title: {sec_name}")
self.section_titles_map[sec_name] = gr.Markdown(f"### {sec_name} (Error Placeholder)")
else:
logging.error("build_analytics_tab_plot_area did not return a tuple of (plot_ui_objects, section_titles_map). Using fallback.")
# Fallback: try to use the result if it's a dict, otherwise empty
self.plot_ui_objects = ui_elements_tuple if isinstance(ui_elements_tuple, dict) else {}
for sec_name in UNIQUE_ORDERED_SECTIONS: # Ensure section_titles_map has entries
self.section_titles_map[sec_name] = gr.Markdown(f"### {sec_name} (Error Placeholder)")
with gr.Column(scale=4, visible=False) as self.global_actions_column_ui:
gr.Markdown("### 💡 Azioni Contestuali Grafico")
self.insights_chatbot_ui = gr.Chatbot(
label="Chat Insights", type="messages", height=450,
bubble_full_width=False, visible=False, show_label=False,
placeholder="L'analisi AI del grafico apparirà qui. Fai domande di approfondimento!"
)
self.insights_chat_input_ui = gr.Textbox(
label="La tua domanda:", placeholder="Chiedi all'AI riguardo a questo grafico...",
lines=2, visible=False, show_label=False
)
with gr.Row(visible=False) as self.insights_suggestions_row_ui:
self.insights_suggestion_1_btn = gr.Button(value="Suggerimento 1", size="sm", min_width=50)
self.insights_suggestion_2_btn = gr.Button(value="Suggerimento 2", size="sm", min_width=50)
self.insights_suggestion_3_btn = gr.Button(value="Suggerimento 3", size="sm", min_width=50)
self.formula_display_markdown_ui = gr.Markdown(
"I dettagli sulla formula/metodologia appariranno qui.", visible=False
)
self.formula_close_hint_md = gr.Markdown(
"<p style='font-size:0.9em; text-align:center; margin-top:10px;'><em>Click the active ƒ button on the plot again to close this panel.</em></p>",
visible=False
)
# After all UI components are defined, populate the output lists
self._define_callback_outputs()
# Then, set up the callbacks that use these components and output lists
self._setup_callbacks()
# The method doesn't need to return anything as it modifies the Gradio app context directly.
# The main app will access necessary components via the instance attributes if needed (e.g., for .then() inputs).