# insights_ui_generator.py import logging from typing import Dict, Any, List, Optional # Configure logger for this module. Assumes logging is configured in app.py or main entry point. logger = logging.getLogger(__name__) def format_report_to_markdown(report_string: Optional[str]) -> str: """ Formats the comprehensive analysis report string into a displayable Markdown format. This can be enhanced to add more structure if the report has implicit sections. Args: report_string: The raw text report from the orchestrator. Returns: A Markdown formatted string. """ 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) # Ensure paragraphs are separated. Replace multiple newlines with double newlines for Markdown paragraphs. # report_string_cleaned = re.sub(r'\n\s*\n', '\n\n', report_string.strip()) formatted_report = f"## Comprehensive Analysis Report\n\n{report_string.strip()}" # You might add more sophisticated parsing here if your LLM output for the report # has a consistent structure that can be converted to richer Markdown. 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)