from typing import Any, Dict, Optional import pandas as pd import numpy as np from pandas.api.types import is_numeric_dtype, is_string_dtype from pydantic import BaseModel class CsvInfoRequest(BaseModel): csv_url: str class CsvInfoResponse(BaseModel): success: bool data: Optional[Dict[str, Any]] = None error: Optional[str] = None request_id: str duration: float class CsvDataRequest(BaseModel): csv_url: str class PythonExecutionRequest(BaseModel): code: str context: Optional[Dict[str, Any]] = None class PythonExecutionResponse(BaseModel): success: bool output: str result: Optional[Any] = None isStructured: bool error: Optional[str] = None request_id: str def clean_data(input_data, drop_constants=True): """ The 'Ultimate' generic data cleaner. MODIFIED: Keeps column names raw and original (no stripping, no lowercasing). """ try: # 1. Flexible Input & Delimiter Detection if isinstance(input_data, str): try: # 'sep=None' with engine='python' attempts to auto-detect delimiters df = pd.read_csv(input_data, sep=None, engine='python') except UnicodeDecodeError: # Fallback to latin1 if utf-8 fails df = pd.read_csv(input_data, sep=None, engine='python', encoding='latin1') except Exception: # Final fallback to standard read_csv df = pd.read_csv(input_data) elif isinstance(input_data, pd.DataFrame): df = input_data.copy() else: raise ValueError("Input must be a CSV URL string or a pandas DataFrame.") # 2. Standardize Column Names -> SKIPPED # We keep the raw column names exactly as they are in the source file. # df.columns = df.columns... (Removed) # 3. Remove Duplicate Rows df = df.drop_duplicates() # 4. Intelligent Type Inference for col in df.columns: # Skip if already numeric if is_numeric_dtype(df[col]): continue # A. Number Parsing (remove currency symbols etc) if is_string_dtype(df[col]): clean_col = df[col].astype(str).str.replace(r'[$,%]', '', regex=True) converted = pd.to_numeric(clean_col, errors='coerce') # Only apply if it converts the majority of the data if converted.notna().mean() > 0.8: df[col] = converted continue # B. Date Parsing if is_string_dtype(df[col]): try: sample = str(df[col].dropna().iloc[0]) if not df[col].dropna().empty else "" # Simple heuristic to check if it looks like a date is_date_like = any(x in sample for x in ['-', '/', ':']) if is_date_like: converted = pd.to_datetime(df[col], errors='coerce') if converted.notna().mean() > 0.8: df[col] = converted except Exception: pass # 5. Handle Infinite Values df.replace([np.inf, -np.inf], np.nan, inplace=True) # 6. Robust Missing Value Filling # Fill numeric columns with 0 num_cols = df.select_dtypes(include=[np.number]).columns df[num_cols] = df[num_cols].fillna(0) # Fill categorical/object columns with 'Unknown' cat_cols = df.select_dtypes(include=['object', 'category']).columns for col in cat_cols: if df[col].dtype.name == 'category': if 'Unknown' not in df[col].cat.categories: df[col] = df[col].cat.add_categories(['Unknown']) df[col] = df[col].fillna('Unknown') else: df[col] = df[col].fillna('Unknown') # Fill boolean columns with False bool_cols = df.select_dtypes(include=['bool']).columns df[bool_cols] = df[bool_cols].fillna(False) # 7. Remove Constant Columns (columns with only 1 unique value) if drop_constants: cols_to_drop = [col for col in df.columns if df[col].nunique() <= 1] if cols_to_drop: df = df.drop(columns=cols_to_drop) return df except Exception as e: raise Exception(f"Data Cleaning Failed: {str(e)}") def get_csv_basic_info(csv_path): """ Get basic information about a CSV file. Includes JSON serialization fix for dates and numpy types. """ try: # Read and clean the CSV file df = clean_data(csv_path) # Helper to make data JSON compliant (Fixes Timestamp and NaN issues) def json_serializable(val): if pd.isna(val): return None if isinstance(val, (pd.Timestamp, np.datetime64)): return str(val) # Convert date to string if isinstance(val, (np.integer, np.int64)): return int(val) if isinstance(val, (np.floating, np.float64)): return float(val) return val # Extract first row and sanitize it raw_sample = df.head(1).to_dict('records') clean_sample = [] if raw_sample: clean_sample = [{k: json_serializable(v) for k, v in raw_sample[0].items()}] print(f"CSV file read successfully: {csv_path}") info = { 'num_rows': int(len(df)), # Ensure Python int, not numpy int 'num_cols': int(len(df.columns)), 'example_rows': clean_sample, # Use the sanitized sample 'dtypes': {col: str(df[col].dtype) for col in df.columns}, 'columns': list(df.columns), 'numeric_columns': [col for col in df.columns if pd.api.types.is_numeric_dtype(df[col])], 'categorical_columns': [col for col in df.columns if pd.api.types.is_string_dtype(df[col])] } return info except Exception as e: error_info = { 'error': f"Error reading CSV file: {str(e)}", } return error_info def get_robust_csv_rows(csv_url: str): """ Reads a CSV securely and robustly for frontend table rendering. - Auto-detects delimiters (semicolon vs comma). - Handles encoding issues. - Replaces NaNs with empty strings for JSON safety. """ try: # 1. Robust Reading (Auto-detect separator, handle encoding) try: df = pd.read_csv(csv_url, sep=None, engine='python') except UnicodeDecodeError: df = pd.read_csv(csv_url, sep=None, engine='python', encoding='latin1') except Exception: # Fallback to standard C engine if python engine fails df = pd.read_csv(csv_url) # 2. Clean for JSON Rendering # Replace infinite values with NaN df.replace([np.inf, -np.inf], np.nan, inplace=True) # Replace NaN with empty string (better for UI tables than 'null') df = df.fillna("") # 3. Convert to List of Dictionaries data_list = df.to_dict(orient='records') return data_list except Exception as e: return {"error": f"Failed to read CSV: {str(e)}"} #--------- GENERIC MODAL CODE EXECUTION LOGIC --------- import io from contextlib import redirect_stdout, redirect_stderr from typing import Any, Dict import requests def check_structured_data(data: Any) -> bool: if isinstance(data, list) and data and all(isinstance(item, dict) for item in data): return all( all(isinstance(v, (str, int, float, bool)) or v is None for v in item.values()) for item in data ) elif isinstance(data, dict): return all(isinstance(v, (str, int, float, bool)) or v is None for v in data.values()) return False def clean_output(stdout: str, stderr: str) -> str: output = [] if stdout.strip(): output.append(stdout.strip()) if stderr.strip(): output.append(stderr.strip()) return '\n'.join(output) if output else '' def execute_python_logic(code: str, custom_context: dict = None) -> Dict[str, Any]: stdout = io.StringIO() stderr = io.StringIO() result = None is_structured = False error = None try: with redirect_stdout(stdout), redirect_stderr(stderr): exec_globals = { '__builtins__': __builtins__, 'requests': requests, 'print': print, } try: compiled = compile(code, '', 'eval') result = eval(compiled, exec_globals) except SyntaxError: compiled = compile(code, '', 'exec') exec(compiled, exec_globals) result = exec_globals.get('result') or exec_globals.get('_') if result is not None: is_structured = check_structured_data(result) except Exception as e: error = f"Execution error: {str(e)}" stderr.write(error) output = clean_output(stdout.getvalue(), stderr.getvalue()) return { 'output': output, 'result': result, 'isStructured': is_structured, 'error': error }