""" Common utility functions for GAIA implementation. This module contains shared utility functions used across different components of the GAIA system, including error handling, logging, and configuration helpers. """ import re import json import logging import traceback from typing import Dict, Any, List, Optional, Union, Tuple, Set, Callable logger = logging.getLogger("gaia_agent.utils.common") def setup_logging(level: str = "INFO", log_file: Optional[str] = None, format_str: str = "%(asctime)s - %(name)s - %(levelname)s - %(message)s"): """ Set up logging with a consistent format across the application. Args: level: Logging level (DEBUG, INFO, WARNING, ERROR, CRITICAL) log_file: Optional file path for logging to a file format_str: String format for log messages """ log_level = getattr(logging, level.upper(), logging.INFO) handlers = [] if log_file: handlers.append(logging.FileHandler(log_file)) handlers.append(logging.StreamHandler()) logging.basicConfig( level=log_level, format=format_str, handlers=handlers ) def safe_execute(func: Callable, *args, **kwargs) -> Tuple[Any, Optional[Exception]]: """ Safely execute a function and catch any exceptions. Args: func: Function to execute *args: Positional arguments to pass to the function **kwargs: Keyword arguments to pass to the function Returns: Tuple of (result, exception) where exception is None if no exception occurred """ try: result = func(*args, **kwargs) return result, None except Exception as e: logger.error(f"Error executing {func.__name__}: {str(e)}") logger.error(traceback.format_exc()) return None, e def get_nested_value(data: Dict[str, Any], key_path: str, default: Any = None) -> Any: """ Get a value from a nested dictionary using dot notation. Args: data: Dictionary to search key_path: Path to the value using dot notation (e.g., 'api.keys.openai') default: Default value to return if the key is not found Returns: The value if found, otherwise the default """ if not key_path: return default parts = key_path.split('.') current = data try: for part in parts: if not isinstance(current, dict): return default if part not in current: return default current = current.get(part) return current except (KeyError, TypeError, AttributeError): return default def set_nested_value(data: Dict[str, Any], key_path: str, value: Any) -> None: """ Set a value in a nested dictionary using dot notation. Args: data: Dictionary to modify key_path: Path to the value using dot notation (e.g., 'api.keys.openai') value: Value to set """ if not key_path: return parts = key_path.split('.') current = data for part in parts[:-1]: if part not in current or not isinstance(current[part], dict): current[part] = {} current = current[part] current[parts[-1]] = value def load_file_content(file_path: str) -> Optional[str]: """ Load the content of a file. Args: file_path: Path to the file Returns: The content of the file, or None if the file could not be loaded """ try: with open(file_path, 'r', encoding='utf-8') as f: return f.read() except Exception as e: logger.error(f"Error loading file {file_path}: {str(e)}") return None def load_json_file(file_path: str) -> Optional[Dict[str, Any]]: """ Load a JSON file. Args: file_path: Path to the JSON file Returns: The parsed JSON data, or None if the file could not be loaded """ content = load_file_content(file_path) if content is None: return None try: return json.loads(content) except json.JSONDecodeError as e: logger.error(f"Error parsing JSON file {file_path}: {str(e)}") return None def validate_api_key(api_key: Optional[str]) -> bool: """ Validate that an API key is not None or empty. Args: api_key: API key to validate Returns: True if the API key is valid, False otherwise """ return api_key is not None and api_key.strip() != "" def format_error_message(error: Union[Exception, str], include_traceback: bool = True) -> str: """ Format an error message with optional traceback. Args: error: Exception or error message include_traceback: Whether to include the traceback Returns: Formatted error message """ if isinstance(error, Exception): error_message = str(error) stack_trace = traceback.format_exc() if include_traceback else "" else: error_message = error stack_trace = "".join(traceback.format_stack()[:-1]) if include_traceback else "" if stack_trace: return f"{error_message}\n\nTraceback:\n{stack_trace}" else: return error_message