joelgilbert/NL2SQL

1"""2General utility helper functions.3"""4 5import logging6import json7import uuid8from typing import List, Dict, Any, Optional9from datetime import datetime10import pandas as pd11 12logger = logging.getLogger(__name__)13 14 15def format_result_table(results: List[Dict], max_rows: int = 100) -> str:16    """17    Convert query results to markdown table.18    19    Args:20        results: Query results as list of dictionaries21        max_rows: Maximum number of rows to display22        23    Returns:24        Markdown formatted table string25    """26    if not results:27        return "_No results_"28    29    # Limit rows30    display_results = results[:max_rows]31    truncated = len(results) > max_rows32    33    try:34        # Use pandas for nice table formatting35        df = pd.DataFrame(display_results)36        37        # Format values38        for col in df.columns:39            df[col] = df[col].apply(lambda x: truncate_long_text(str(x), 100) if x is not None else 'NULL')40        41        # Convert to markdown42        markdown = df.to_markdown(index=False)43        44        if truncated:45            markdown += f"\n\n_... and {len(results) - max_rows} more rows_"46        47        return markdown48        49    except Exception as e:50        logger.error(f"Failed to format result table: {e}")51        # Fallback to simple format52        return str(results[:5])53 54 55def truncate_long_text(text: str, max_length: int = 50) -> str:56    """57    Safely truncate text with ellipsis.58    59    Args:60        text: Text to truncate61        max_length: Maximum length62        63    Returns:64        Truncated text65    """66    if len(text) <= max_length:67        return text68    69    return text[:max_length - 3] + "..."70 71 72def format_execution_time(seconds: float) -> str:73    """74    Format execution time in human-readable format.75    76    Args:77        seconds: Execution time in seconds78        79    Returns:80        Formatted time string (e.g., "450 ms", "2.3 s")81    """82    if seconds < 1:83        return f"{seconds * 1000:.0f} ms"84    elif seconds < 60:85        return f"{seconds:.2f} s"86    else:87        minutes = int(seconds // 60)88        remaining_seconds = seconds % 6089        return f"{minutes}m {remaining_seconds:.1f}s"90 91 92def generate_session_id() -> str:93    """94    Generate unique session identifier.95    96    Returns:97        Unique session ID string98    """99    timestamp = datetime.now().strftime('%Y%m%d_%H%M%S')100    unique_id = uuid.uuid4().hex[:8]101    return f"session_{timestamp}_{unique_id}"102 103 104def safe_json_loads(json_str: str) -> Dict:105    """106    Parse JSON string with error handling.107    108    Args:109        json_str: JSON string to parse110        111    Returns:112        Parsed dictionary or empty dict if parsing fails113    """114    try:115        return json.loads(json_str)116    except json.JSONDecodeError as e:117        logger.error(f"Failed to parse JSON: {e}")118        return {}119    except Exception as e:120        logger.error(f"Unexpected error parsing JSON: {e}")121        return {}122 123 124def format_timestamp(timestamp: Optional[datetime] = None) -> str:125    """126    Format timestamp for display.127    128    Args:129        timestamp: Datetime object (None = current time)130        131    Returns:132        Formatted timestamp string133    """134    if timestamp is None:135        timestamp = datetime.now()136    137    return timestamp.strftime('%Y-%m-%d %H:%M:%S')138 139 140def sanitize_filename(filename: str) -> str:141    """142    Sanitize filename for safe file system operations.143    144    Args:145        filename: Original filename146        147    Returns:148        Sanitized filename149    """150    # Remove or replace unsafe characters151    import re152    # Keep only alphanumeric, dash, underscore, and dot153    sanitized = re.sub(r'[^\w\-.]', '_', filename)154    # Remove leading/trailing dots and underscores155    sanitized = sanitized.strip('._')156    # Limit length157    if len(sanitized) > 100:158        sanitized = sanitized[:100]159    160    return sanitized161 162 163def pluralize(count: int, singular: str, plural: Optional[str] = None) -> str:164    """165    Return singular or plural form based on count.166    167    Args:168        count: Number of items169        singular: Singular form of word170        plural: Plural form (if None, adds 's' to singular)171        172    Returns:173        Appropriate form with count174    """175    if plural is None:176        plural = f"{singular}s"177    178    word = singular if count == 1 else plural179    return f"{count} {word}"180 181 182def format_bytes(bytes_size: int) -> str:183    """184    Format bytes in human-readable format.185    186    Args:187        bytes_size: Size in bytes188        189    Returns:190        Formatted size string (e.g., "1.5 KB", "2.3 MB")191    """192    for unit in ['B', 'KB', 'MB', 'GB', 'TB']:193        if bytes_size < 1024.0:194            return f"{bytes_size:.1f} {unit}"195        bytes_size /= 1024.0196    return f"{bytes_size:.1f} PB"197 198 199def calculate_success_rate(successful: int, total: int) -> float:200    """201    Calculate success rate percentage.202    203    Args:204        successful: Number of successful operations205        total: Total number of operations206        207    Returns:208        Success rate as percentage (0-100)209    """210    if total == 0:211        return 0.0212    return (successful / total) * 100213 214 215def merge_dicts(*dicts: Dict) -> Dict:216    """217    Merge multiple dictionaries (later dicts override earlier ones).218    219    Args:220        *dicts: Variable number of dictionaries to merge221        222    Returns:223        Merged dictionary224    """225    result = {}226    for d in dicts:227        result.update(d)228    return result229 230 231def chunk_list(lst: List, chunk_size: int) -> List[List]:232    """233    Split list into chunks of specified size.234    235    Args:236        lst: List to chunk237        chunk_size: Size of each chunk238        239    Returns:240        List of chunks241    """242    return [lst[i:i + chunk_size] for i in range(0, len(lst), chunk_size)]243 244 245def dict_to_table(data: Dict, title: Optional[str] = None) -> str:246    """247    Convert dictionary to simple markdown table.248    249    Args:250        data: Dictionary to convert251        title: Optional table title252        253    Returns:254        Markdown formatted table255    """256    lines = []257    258    if title:259        lines.append(f"### {title}\n")260    261    lines.append("| Key | Value |")262    lines.append("|-----|-------|")263    264    for key, value in data.items():265        # Format value266        if isinstance(value, (list, dict)):267            value = json.dumps(value)268        value_str = truncate_long_text(str(value), 100)269        270        lines.append(f"| {key} | {value_str} |")271    272    return "\n".join(lines)273