csv_metadata_service.py

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, '<string>', 'eval')
                result = eval(compiled, exec_globals)
            except SyntaxError:
                compiled = compile(code, '<string>', '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
    }