import logging from typing import Dict, Any, List, Optional import pandas as pd # Configure logger for this module. Assumes logging is configured in app.py or main entry point. logger = logging.getLogger(__name__) def format_report_for_display(report_data: Optional[pd.Series]) -> Dict[str, str]: """ Enhanced report formatting. Returns a dictionary with separate HTML for the header and Markdown for the report body, allowing flexible rendering. Args: report_data: A pandas Series representing a single row from the agentic analysis DataFrame. It should contain 'report_text', 'report_type', and 'Created Date'. Returns: A dictionary with 'header_html' and 'body_markdown' keys. 'header_html' contains the HTML for the report title and subtitle. 'body_markdown' contains the raw Markdown text for the report body. If data is invalid, it returns empty state HTML for both. """ if report_data is None or report_data.empty: empty_state_html = """
📄
Report Not Available
The selected report could not be loaded. Please try selecting a different report or refresh the page.
""" return {'header_html': '', 'body_markdown': empty_state_html} # Extract report data # Ensure 'report_text' is treated as raw text that will be interpreted as Markdown report_text = report_data.get('report_text', '*Report content not found.*') report_type = report_data.get('report_type') created_date_str = report_data.get('Created Date') # Generate dynamic title and subtitle title = "Comprehensive Analysis Report" subtitle = "" try: if report_type == 'Quarter': title = "📈 Quarterly Insights Report" subtitle = "Strategic analysis of your quarterly performance" elif report_type == 'Week' and pd.notna(created_date_str): # Ensure pandas is used for datetime conversion created_date = pd.to_datetime(created_date_str) day_name = created_date.strftime('%A') title = f"📊 {day_name}'s Weekly Update" subtitle = f"Weekly performance analysis for {created_date.strftime('%B %d, %Y')}" except Exception as e: logger.error(f"Error generating dynamic report title: {e}") # In case of an error, default title and no subtitle will be used. # Format the report header HTML header_html = f"""

{title}

{f'

{subtitle}

' if subtitle else ''}
""" # The report_text itself should be the raw Markdown for the body body_markdown = report_text.strip() return {'header_html': header_html, 'body_markdown': body_markdown} def format_report_to_markdown(report_string: Optional[str]) -> str: """ This function was previously for general Markdown formatting. Given the new structure where 'format_report_for_display' handles the split, this function might become redundant or repurposed. Keeping it for now but noting its potential redundancy depending on upstream calls. """ if not report_string or not report_string.strip(): return "## Comprehensive Analysis Report\n\n*No analysis report was generated, or an error occurred during its generation.*" # Simple formatting for now. Could be enhanced (e.g., looking for patterns like "Section X:" to make them H3) formatted_report = f"## Comprehensive Analysis Report\n\n{report_string.strip()}" return formatted_report def extract_key_results_for_selection( actionable_okrs_and_tasks_dict: Optional[Dict[str, Any]] ) -> List[Dict[str, Any]]: """ Extracts Key Results from the OKR structure for UI selection in Gradio. Each Key Result is given a unique ID for state management in the Gradio app. Args: actionable_okrs_and_tasks_dict: The dictionary representation of TaskExtractionOutput, typically `orchestration_results["actionable_okrs_and_tasks"]`. Expected structure: {'okrs': List[OKR_dict], ...} Returns: A list of dictionaries, where each dictionary represents a Key Result: {'okr_index': int, 'kr_index': int, 'okr_objective': str, 'kr_description': str, 'unique_kr_id': str} """ key_results_for_ui: List[Dict[str, Any]] = [] if not actionable_okrs_and_tasks_dict or not isinstance(actionable_okrs_and_tasks_dict.get('okrs'), list): logger.warning("No 'okrs' list found or it's not a list in the provided task extraction output.") return key_results_for_ui okrs_list = actionable_okrs_and_tasks_dict['okrs'] for okr_idx, okr_data in enumerate(okrs_list): if not isinstance(okr_data, dict): logger.warning(f"OKR item at index {okr_idx} is not a dictionary, skipping.") continue okr_objective = okr_data.get('objective_description', f"Objective {okr_idx + 1} (Unnamed)") key_results_list = okr_data.get('key_results', []) if not isinstance(key_results_list, list): logger.warning(f"Expected 'key_results' in OKR '{okr_objective}' (index {okr_idx}) to be a list, got {type(key_results_list)}.") continue for kr_idx, kr_data in enumerate(key_results_list): if not isinstance(kr_data, dict): logger.warning(f"Key Result item for OKR '{okr_objective}' at KR index {kr_idx} is not a dictionary, skipping.") continue kr_description = kr_data.get('key_result_description') or kr_data.get('description') or f"Key Result {kr_idx+1} (No description)" key_results_for_ui.append({ 'okr_index': okr_idx, # Index of the parent OKR in the original list 'kr_index': kr_idx, # Index of this KR within its parent OKR 'okr_objective': okr_objective, 'kr_description': kr_description, 'unique_kr_id': f"okr{okr_idx}_kr{kr_idx}" # Unique ID for Gradio component linking }) if not key_results_for_ui: logger.info("No Key Results were extracted for selection from the OKR data.") return key_results_for_ui def format_single_okr_for_display( okr_data: Dict[str, Any], accepted_kr_indices: Optional[List[int]] = None, okr_main_index: Optional[int] = None # For titling if needed ) -> str: """ Formats a single complete OKR object (with its Key Results and Tasks) into a detailed Markdown string for display. Optionally filters to show only accepted Key Results. Args: okr_data: A dictionary representing a single OKR from the TaskExtractionOutput. accepted_kr_indices: Optional list of indices of Key Results within this OKR that were accepted by the user. If None, all KRs are displayed. okr_main_index: Optional index of this OKR in the main list, for titling. Returns: A Markdown formatted string representing the OKR. """ if not okr_data or not isinstance(okr_data, dict): return "*Invalid OKR data provided for display.*\n" md_parts = [] objective_title_num = f" {okr_main_index + 1}" if okr_main_index is not None else "" objective = okr_data.get('objective_description') or okr_data.get('description') or f"Unnamed Objective{objective_title_num}" logger.info(f"OKR data desccr {objective}") objective_timeline = okr_data.get('objective_timeline', '') objective_owner = okr_data.get('objective_owner', 'N/A') md_parts.append(f"### Objective{objective_title_num}: {objective}") if objective_timeline: md_parts.append(f"**Overall Timeline:** {objective_timeline}") if objective_owner and objective_owner != 'N/A': md_parts.append(f"**Overall Owner:** {objective_owner}") md_parts.append("\n---") key_results_list = okr_data.get('key_results', []) displayed_kr_count = 0 if not isinstance(key_results_list, list) or not key_results_list: md_parts.append("\n*No Key Results defined for this objective.*") else: for kr_idx, kr_data in enumerate(key_results_list): if accepted_kr_indices is not None and kr_idx not in accepted_kr_indices: continue # Skip this KR if a filter is applied and it's not in the accepted list displayed_kr_count +=1 if not isinstance(kr_data, dict): md_parts.append(f"\n**Key Result {kr_idx+1}:** *Invalid data format for this Key Result.*") continue kr_desc = kr_data.get('key_result_description') or kr_data.get('description') or f"Key Result {kr_idx+1} (No description)" logger.info(f"KR data desccr {kr_desc}") target_metric = kr_data.get('target_metric') target_value = kr_data.get('target_value') kr_data_subj = kr_data.get('data_subject') kr_type = kr_data.get('key_result_type') md_parts.append(f"\n#### Key Result {displayed_kr_count} (Original Index: {kr_idx+1}): {kr_desc}") if target_metric and target_value: md_parts.append(f" - **Target:** Measure `{target_metric}` to achieve/reach `{target_value}`") if kr_type and kr_data_subj: md_parts.append(f" **Key result type**: {kr_type}, for **data subject** {kr_data_subj}") tasks_list = kr_data.get('tasks', []) if tasks_list and isinstance(tasks_list, list): md_parts.append(" **Associated Tasks:**") for task_idx, task_data in enumerate(tasks_list): if not isinstance(task_data, dict): md_parts.append(f" - Task {task_idx+1}: *Invalid data format for this task.*") continue task_desc = task_data.get('task_description') or task_data.get('description') or f"Task {task_idx+1} (No description)" logger.info(f"task data desccr {task_desc}") task_cat = task_data.get('task_category') or task_data.get('category') or 'N/A' task_effort = task_data.get('effort', 'N/A') task_timeline = task_data.get('timeline', 'N/A') task_priority = task_data.get('priority', 'N/A') task_responsible = task_data.get('responsible_party', 'N/A') task_type = task_data.get('task_type', 'N/A') data_subject_val = task_data.get('data_subject') data_subject_str = f", Data Subject: `{data_subject_val}`" if data_subject_val and task_type == 'tracking' else "" md_parts.append(f" - **{task_idx+1}. {task_desc}**") md_parts.append(f" - *Category:* {task_cat} | *Type:* {task_type}{data_subject_str}") md_parts.append(f" - *Priority:* **{task_priority}** | *Effort:* {task_effort} | *Timeline:* {task_timeline}") md_parts.append(f" - *Responsible:* {task_responsible}") obj_deliv = task_data.get('objective_deliverable') if obj_deliv: md_parts.append(f" - *Objective/Deliverable:* {obj_deliv}") success_crit = task_data.get('success_criteria_metrics') if success_crit: md_parts.append(f" - *Success Metrics:* {success_crit}") why_prop = task_data.get('why_proposed') if why_prop: md_parts.append(f" - *Rationale:* {why_prop}") priority_just = task_data.get('priority_justification') if priority_just: md_parts.append(f" - *Priority Justification:* {priority_just}") dependencies = task_data.get('dependencies_prerequisites') if dependencies: md_parts.append(f" - *Dependencies:* {dependencies}") md_parts.append("") # Extra newline for spacing between tasks details else: md_parts.append(" *No tasks defined for this Key Result.*") md_parts.append("\n---\n") # Separator between Key Results if displayed_kr_count == 0 and accepted_kr_indices is not None: md_parts.append("\n*No Key Results matching the 'accepted' filter for this objective.*") return "\n".join(md_parts)