Create insights_ui_generator.py

insights_ui_generator.py

@@ -0,0 +1,193 @@
+# 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', f"Key Result {kr_idx + 1} (No description provided)")
+            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', f"Unnamed Objective{objective_title_num}")
+    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', f"Key Result {kr_idx+1} (No description)")
+            target_metric = kr_data.get('target_metric')
+            target_value = kr_data.get('target_value')
+
+            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}`")
+            
+            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', f"Task {task_idx+1} (No description)")
+                    task_cat = task_data.get('task_category', '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)
+