Spaces:
Running
Running
# 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). | |