Spaces:
Running
Running
| # services/report_data_handler.py | |
| import pandas as pd | |
| import logging | |
| from apis.Bubble_API_Calls import fetch_linkedin_posts_data_from_bubble, bulk_upload_to_bubble | |
| from config import ( | |
| BUBBLE_REPORT_TABLE_NAME, | |
| BUBBLE_OKR_TABLE_NAME, | |
| BUBBLE_KEY_RESULTS_TABLE_NAME, | |
| BUBBLE_TASKS_TABLE_NAME, | |
| BUBBLE_KR_UPDATE_TABLE_NAME, | |
| ) | |
| import json # For handling JSON data | |
| from typing import List, Dict, Any, Optional, Tuple | |
| # It's good practice to configure the logger at the application entry point, | |
| # but setting a default handler here prevents "No handler found" warnings. | |
| logging.basicConfig(level=logging.INFO) | |
| logger = logging.getLogger(__name__) | |
| def fetch_latest_agentic_analysis(org_urn: str) -> Tuple[Optional[pd.DataFrame], Optional[str]]: | |
| """ | |
| Fetches all agentic analysis data for a given org_urn from Bubble. | |
| Returns the full dataframe and any error message, or None, None. | |
| """ | |
| logger.info(f"Starting fetch_latest_agentic_analysis for org_urn: {org_urn}") | |
| if not org_urn: | |
| logger.warning("fetch_latest_agentic_analysis: org_urn is missing.") | |
| return None, "org_urn is missing." | |
| try: | |
| report_data_df, error = fetch_linkedin_posts_data_from_bubble( | |
| data_type=BUBBLE_REPORT_TABLE_NAME, | |
| org_urn=org_urn | |
| ) | |
| if error: | |
| logger.error(f"Error fetching data from Bubble for org_urn {org_urn}: {error}") | |
| return None, str(error) | |
| if report_data_df is None or report_data_df.empty: | |
| logger.info(f"No existing agentic analysis found in Bubble for org_urn {org_urn}.") | |
| return None, None | |
| logger.info(f"Successfully fetched {len(report_data_df)} records for org_urn {org_urn}") | |
| return report_data_df, None # Return full dataframe and no error | |
| except Exception as e: | |
| logger.exception(f"An unexpected error occurred in fetch_latest_agentic_analysis for org_urn {org_urn}: {e}") | |
| return None, str(e) | |
| def save_report_results( | |
| org_urn: str, | |
| report_markdown: str, | |
| quarter: int, | |
| year: int, | |
| report_type: str, | |
| ) -> Optional[str]: | |
| """Saves the agentic pipeline results to Bubble. Returns the new record ID or None.""" | |
| logger.info(f"Starting save_report_results for org_urn: {org_urn}") | |
| if not org_urn: | |
| logger.error("Cannot save agentic results: org_urn is missing.") | |
| return None | |
| try: | |
| payload = { | |
| "organization_urn": org_urn, | |
| "report_text": report_markdown if report_markdown else "N/A", | |
| "quarter": quarter, | |
| "year": year, | |
| "report_type": report_type, | |
| } | |
| logger.info(f"Attempting to save agentic analysis to Bubble for org_urn: {org_urn}") | |
| response = bulk_upload_to_bubble([payload], BUBBLE_REPORT_TABLE_NAME) | |
| # Handle API response which could be a list of dicts (for bulk) or a single dict. | |
| if response and isinstance(response, list) and len(response) > 0 and isinstance(response[0], dict) and 'id' in response[0]: | |
| record_id = response[0]['id'] # Get the ID from the first dictionary in the list | |
| logger.info(f"Successfully saved agentic analysis to Bubble. Record ID: {record_id}") | |
| return record_id | |
| elif response and isinstance(response, dict) and "id" in response: # Handle non-bulk response | |
| record_id = response["id"] | |
| logger.info(f"Successfully saved agentic analysis to Bubble. Record ID: {record_id}") | |
| return record_id | |
| else: | |
| # Catches None, False, empty lists, or other unexpected formats. | |
| logger.error(f"Failed to save agentic analysis to Bubble. Unexpected API Response: {response}") | |
| return None | |
| except Exception as e: | |
| logger.exception(f"An unexpected error occurred in save_report_results for org_urn {org_urn}: {e}") | |
| return None | |
| # --- Data Saving Functions --- | |
| def save_objectives( | |
| org_urn: str, | |
| report_id: str, | |
| objectives_data: List[Dict[str, Any]] | |
| ) -> Optional[List[str]]: | |
| """ | |
| Saves Objective records to Bubble. | |
| Returns a list of the newly created Bubble record IDs for the objectives, or None on failure. | |
| """ | |
| logger.info(f"Starting save_objectives for report_id: {report_id}") | |
| if not objectives_data: | |
| logger.info("No objectives to save.") | |
| return [] | |
| try: | |
| payloads = [] | |
| for obj in objectives_data: | |
| timeline = obj.get("objective_timeline") | |
| payloads.append({ | |
| "description": obj.get("objective_description"), | |
| # FIX: Convert Enum to its value before sending. | |
| "timeline": timeline.value if hasattr(timeline, 'value') else timeline, | |
| "owner": obj.get("objective_owner"), | |
| "report": report_id, | |
| }) | |
| logger.info(f"Attempting to save {payloads} objectives for report_id: {report_id}") | |
| response_data = bulk_upload_to_bubble(payloads, BUBBLE_OKR_TABLE_NAME) | |
| # Validate response and extract IDs from the list of dictionaries. | |
| if not response_data or not isinstance(response_data, list): | |
| logger.error(f"Failed to save objectives. API response was not a list: {response_data}") | |
| return None | |
| try: | |
| # Extract the ID from each dictionary in the response list. | |
| extracted_ids = [item['id'] for item in response_data] | |
| except (TypeError, KeyError): | |
| logger.error(f"Failed to parse IDs from API response. Response format invalid: {response_data}", exc_info=True) | |
| return None | |
| # Check if we extracted the expected number of IDs | |
| if len(extracted_ids) != len(payloads): | |
| logger.error(f"Failed to save all objectives for report_id: {report_id}. " | |
| f"Expected {len(payloads)} IDs, but got {len(extracted_ids)} from response: {response_data}") | |
| return None | |
| logger.info(f"Successfully saved {len(extracted_ids)} objectives.") | |
| return extracted_ids | |
| except Exception as e: | |
| logger.exception(f"An unexpected error occurred in save_objectives for report_id {report_id}: {e}") | |
| return None | |
| def save_key_results( | |
| org_urn: str, | |
| objectives_with_ids: List[Tuple[Dict[str, Any], str]] | |
| ) -> Optional[List[Tuple[Dict[str, Any], str]]]: | |
| """ | |
| Saves Key Result records to Bubble, linking them to their parent objectives. | |
| Returns a list of tuples containing the original key result data and its new Bubble ID, or None on failure. | |
| """ | |
| logger.info(f"Starting save_key_results for {len(objectives_with_ids)} objectives.") | |
| key_result_payloads = [] | |
| # This list preserves the original KR data in the correct order to match the returned IDs | |
| key_results_to_process = [] | |
| if not objectives_with_ids: | |
| logger.info("No objectives provided to save_key_results.") | |
| return [] | |
| try: | |
| for objective_data, parent_objective_id in objectives_with_ids: | |
| # Defensive check to ensure the parent_objective_id is a valid-looking string. | |
| if not isinstance(parent_objective_id, str) or not parent_objective_id: | |
| logger.error(f"Invalid parent_objective_id found: '{parent_objective_id}'. Skipping KRs for this objective.") | |
| continue # Skip this loop iteration | |
| for kr in objective_data.get("key_results", []): | |
| kr_type = kr.get("key_result_type") | |
| key_results_to_process.append(kr) | |
| key_result_payloads.append({ | |
| "okr": parent_objective_id, | |
| "description": kr.get("key_result_description"), | |
| "target_metric": kr.get("target_metric"), | |
| "target_value": kr.get("target_value"), | |
| # FIX: Convert Enum to its value before sending. | |
| "kr_type": kr_type.value if hasattr(kr_type, 'value') else kr_type, | |
| "data_subject": kr.get("data_subject"), | |
| }) | |
| if not key_result_payloads: | |
| logger.info("No key results to save.") | |
| return [] | |
| logger.info(f"Attempting to save {key_result_payloads} key results for org_urn: {org_urn}") | |
| response_data = bulk_upload_to_bubble(key_result_payloads, BUBBLE_KEY_RESULTS_TABLE_NAME) | |
| # Validate response and extract IDs. | |
| if not response_data or not isinstance(response_data, list): | |
| logger.error(f"Failed to save key results. API response was not a list: {response_data}") | |
| return None | |
| try: | |
| extracted_ids = [item['id'] for item in response_data] | |
| except (TypeError, KeyError): | |
| logger.error(f"Failed to parse IDs from key result API response: {response_data}", exc_info=True) | |
| return None | |
| if len(extracted_ids) != len(key_result_payloads): | |
| logger.error(f"Failed to save all key results for org_urn: {org_urn}. " | |
| f"Expected {len(key_result_payloads)} IDs, but got {len(extracted_ids)} from response: {response_data}") | |
| return None | |
| logger.info(f"Successfully saved {len(extracted_ids)} key results.") | |
| return list(zip(key_results_to_process, extracted_ids)) | |
| except Exception as e: | |
| logger.exception(f"An unexpected error occurred in save_key_results for org_urn {org_urn}: {e}") | |
| return None | |
| def save_tasks( | |
| org_urn: str, | |
| key_results_with_ids: List[Tuple[Dict[str, Any], str]] | |
| ) -> Optional[List[str]]: | |
| """ | |
| Saves Task records to Bubble, linking them to their parent key results. | |
| Returns a list of the newly created Bubble record IDs for the tasks, or None on failure. | |
| """ | |
| logger.info(f"Starting save_tasks for {len(key_results_with_ids)} key results.") | |
| if not key_results_with_ids: | |
| logger.info("No key results provided to save_tasks.") | |
| return [] | |
| try: | |
| task_payloads = [] | |
| for key_result_data, parent_key_result_id in key_results_with_ids: | |
| for task in key_result_data.get("tasks", []): | |
| priority = task.get("priority") | |
| effort = task.get("effort") | |
| timeline = task.get("timeline") | |
| task_payloads.append({ | |
| "key_result": parent_key_result_id, | |
| "description": task.get("task_description"), | |
| "deliverable": task.get("objective_deliverable"), | |
| "category": task.get("task_category"), | |
| # FIX: Convert Enum to its value before sending. | |
| "priority": priority.value if hasattr(priority, 'value') else priority, | |
| "priority_justification": task.get("priority_justification"), | |
| "effort": effort.value if hasattr(effort, 'value') else effort, | |
| "timeline": timeline.value if hasattr(timeline, 'value') else timeline, | |
| "responsible_party": task.get("responsible_party"), | |
| "success_criteria_metrics": task.get("success_criteria_metrics"), | |
| "dependencies": task.get("dependencies_prerequisites"), | |
| "why": task.get("why_proposed"), | |
| }) | |
| if not task_payloads: | |
| logger.info("No tasks to save.") | |
| return [] | |
| logger.info(f"Attempting to save {task_payloads} tasks for org_urn: {org_urn}") | |
| response_data = bulk_upload_to_bubble(task_payloads, BUBBLE_TASKS_TABLE_NAME) | |
| # Validate response and extract IDs. | |
| if not response_data or not isinstance(response_data, list): | |
| logger.error(f"Failed to save tasks. API response was not a list: {response_data}") | |
| return None | |
| try: | |
| extracted_ids = [item['id'] for item in response_data] | |
| except (TypeError, KeyError): | |
| logger.error(f"Failed to parse IDs from task API response: {response_data}", exc_info=True) | |
| return None | |
| if len(extracted_ids) != len(task_payloads): | |
| logger.error(f"Failed to save all tasks for org_urn: {org_urn}. " | |
| f"Expected {len(task_payloads)} IDs, but got {len(extracted_ids)} from response: {response_data}") | |
| return None | |
| logger.info(f"Successfully saved {len(extracted_ids)} tasks.") | |
| return extracted_ids | |
| except Exception as e: | |
| logger.exception(f"An unexpected error occurred in save_tasks for org_urn {org_urn}: {e}") | |
| return None | |
| # --- Orchestrator Function --- | |
| def save_actionable_okrs(org_urn: str, actionable_okrs: Dict[str, Any], report_id: str): | |
| """ | |
| Orchestrates the sequential saving of objectives, key results, and tasks. | |
| """ | |
| logger.info(f"--- Starting OKR save process for org_urn: {org_urn}, report_id: {report_id} ---") | |
| try: | |
| objectives_data = actionable_okrs.get("okrs", []) | |
| # Defensive check: If data is a string, try to parse it as JSON. | |
| if isinstance(objectives_data, str): | |
| logger.warning("The 'okrs' data is a string. Attempting to parse as JSON.") | |
| try: | |
| objectives_data = json.loads(objectives_data) | |
| logger.info("Successfully parsed 'okrs' data from JSON string.") | |
| except json.JSONDecodeError: | |
| logger.error("Failed to parse 'okrs' data. The string is not valid JSON.", exc_info=True) | |
| return # Abort if data is malformed | |
| if not objectives_data: | |
| logger.warning(f"No OKRs found in the input for org_urn: {org_urn}. Aborting save process.") | |
| return | |
| # Step 1: Save the top-level objectives | |
| objective_ids = save_objectives(org_urn, report_id, objectives_data) | |
| if objective_ids is None: | |
| logger.error("OKR save process aborted due to failure in saving objectives.") | |
| return | |
| # Combine the original objective data with their new IDs for the next step | |
| objectives_with_ids = list(zip(objectives_data, objective_ids)) | |
| # Step 2: Save the key results, linking them to the objectives | |
| key_results_with_ids = save_key_results(org_urn, objectives_with_ids) | |
| if key_results_with_ids is None: | |
| logger.error("OKR save process aborted due to failure in saving key results.") | |
| return | |
| # Step 3: Save the tasks, linking them to the key results | |
| task_ids = save_tasks(org_urn, key_results_with_ids) | |
| if task_ids is None: | |
| logger.error("Task saving failed, but objectives and key results were saved.") | |
| # For now, we just log the error and complete. | |
| return | |
| logger.info(f"--- OKR save process completed successfully for org_urn: {org_urn} ---") | |
| except Exception as e: | |
| logger.exception(f"An unhandled exception occurred during the save_actionable_okrs orchestration for org_urn {org_urn}: {e}") | |