# 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 from datetime import date # 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}") today = date.today() current_year = today.year current_quarter = (today.month - 1) // 3 + 1 if not org_urn: logger.warning("fetch_latest_agentic_analysis: org_urn is missing.") return None, "org_urn is missing." additional_constraint = [ {"key": 'quarter', "constraint_type": "equals", "value": current_quarter}, {"key": 'year', "constraint_type": "equals", "value": current_year} ] try: report_data_df, error = fetch_linkedin_posts_data_from_bubble( data_type=BUBBLE_REPORT_TABLE_NAME, constraint_value=org_urn, constraint_key='organization_urn', constraint_type = 'equals' ) 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}") def fetch_and_reconstruct_data_from_bubble(report_df: pd.DataFrame) -> Optional[Dict[str, Any]]: """ Fetches the latest report, OKRs, Key Results, and Tasks from Bubble for a given organization and reconstructs them into the nested structure expected by the application. Args: org_urn: The URN of the organization. Returns: A dictionary containing the reconstructed data ('report_str', 'actionable_okrs', etc.) or None if the report is not found or an error occurs. """ # logger.info(f"Starting data fetch and reconstruction for org_urn: {org_urn}") # try: # # 1. Fetch the latest report for the organization # # We add a sort field to get the most recent one. # report_df, error = fetch_linkedin_posts_data_from_bubble( # data_type=BUBBLE_REPORT_TABLE_NAME, # org_urn=org_urn, # constraint_key="organization_urn" # ) # if error or report_df is None or report_df.empty: # logger.error(f"Could not fetch latest report for org_urn {org_urn}. Error: {error}") # return None logger.info(f"Starting data fetch and reconstruction") try: # Get the most recent report (assuming the first one is the latest) latest_report = report_df.iloc[0] report_id = latest_report.get('_id') if not report_id: logger.error("Fetched report is missing a Bubble '_id'.") return None logger.info(f"Fetched latest report with ID: {report_id}") # 2. Fetch all related OKRs using the report_id okrs_df, error = fetch_linkedin_posts_data_from_bubble( data_type=BUBBLE_OKR_TABLE_NAME, constraint_value=str(report_id), constraint_key='report', constraint_type = 'equals' ) if error: logger.error(f"Error fetching OKRs for report_id {report_id}: {error}") okrs_df = pd.DataFrame() logger.info(f" okr_df {okrs_df}") # 3. Fetch all related Key Results using the OKR IDs okr_ids = okrs_df['_id'].tolist() if not okrs_df.empty else [] logger.info(f" retrieved {len(okr_ids)} okr ID: {okr_ids}") krs_df = pd.DataFrame() if okr_ids: krs_df, error = fetch_linkedin_posts_data_from_bubble( data_type=BUBBLE_KEY_RESULTS_TABLE_NAME, constraint_value=okr_ids, constraint_key='okr', constraint_type='in' ) if error: logger.error(f"Error fetching Key Results for OKR IDs {okr_ids}: {error}") krs_df = pd.DataFrame() # 4. Fetch all related Tasks using the Key Result IDs kr_ids = krs_df['_id'].tolist() if not krs_df.empty else [] tasks_df = pd.DataFrame() if kr_ids: tasks_df, error = fetch_linkedin_posts_data_from_bubble( data_type=BUBBLE_TASKS_TABLE_NAME, constraint_value=kr_ids, constraint_key='key_result', constraint_type='in' ) if error: logger.error(f"Error fetching Tasks for KR IDs {kr_ids}: {error}") tasks_df = pd.DataFrame() # 5. Reconstruct the nested 'actionable_okrs' dictionary tasks_by_kr_id = tasks_df.groupby('key_result').apply(lambda x: x.to_dict('records')).to_dict() krs_by_okr_id = krs_df.groupby('okr').apply(lambda x: x.to_dict('records')).to_dict() reconstructed_okrs = [] for okr_data in okrs_df.to_dict('records'): okr_id = okr_data['_id'] key_results_list = krs_by_okr_id.get(okr_id, []) for kr_data in key_results_list: kr_id = kr_data['_id'] # Attach tasks to each key result kr_data['tasks'] = tasks_by_kr_id.get(kr_id, []) # Attach key results to the objective okr_data['key_results'] = key_results_list reconstructed_okrs.append(okr_data) actionable_okrs = {"okrs": reconstructed_okrs} return { "report_str": latest_report.get("report_text", "Nessun report trovato."), "quarter": latest_report.get("quarter"), "year": latest_report.get("year"), "actionable_okrs": actionable_okrs, "report_id": report_id } except Exception as e: logger.exception(f"An unexpected error occurred during data reconstruction for org_urn {org_urn}: {e}") return None