Initial commit

Dockerfile

@@ -0,0 +1,46 @@
+# Use the official Python 3.11 slim image
+FROM python:3.11-slim
+
+# Install curl for keep-alive script and clean up
+RUN apt-get update && \
+    apt-get install -y curl && \
+    rm -rf /var/lib/apt/lists/*
+
+# Set the working directory inside the container
+WORKDIR /app
+
+# Create all required directories with proper permissions upfront
+RUN mkdir -p /app/generated_outputs && \
+    mkdir -p /app/generated_charts && \
+    mkdir -p /app/cache && \
+    chmod -R 777 /app/generated_outputs && \
+    chmod -R 777 /app/generated_charts && \
+    chmod -R 777 /app/cache
+
+# Create log files with proper permissions
+RUN touch /app/pandasai.log && \
+    touch /app/api_key_rotation.log && \
+    chmod 666 /app/pandasai.log && \
+    chmod 666 /app/api_key_rotation.log
+
+# Set environment variables
+ENV MPLCONFIGDIR=/app/cache
+
+# Copy the requirements file first
+COPY requirements.txt .
+
+# Install dependencies
+RUN pip install --no-cache-dir -r requirements.txt
+
+# Copy the rest of the application code
+COPY . .
+
+# Ensure the user has write permissions to all directories
+RUN chown -R 1000:1000 /app && \
+    chmod -R 777 /app
+
+# Expose port 7860 (required by Hugging Face Spaces)
+EXPOSE 7860
+
+# Keep-alive command (pings every 5 minutes) + start Uvicorn
+CMD bash -c "while true; do curl -s https://soumik555-fastapi.hf.space/ping >/dev/null && sleep 300; done & uvicorn controller:app --host 0.0.0.0 --port 7860"

controller.py

@@ -0,0 +1,474 @@
+import base64
+from contextlib import asynccontextmanager
+import logging
+import time
+import uuid
+import os
+import re
+import asyncio  # Added for parallel batching
+import multiprocessing  # Added for worker calculation
+from typing import List, Optional, Dict, Any, TypeVar, Generic  # Added Generic/TypeVar
+from urllib.parse import urlparse, parse_qs, urlencode, urlunparse
+
+# --- FastAPI & Core ---
+from fastapi import FastAPI, HTTPException, Depends
+from fastapi.encoders import jsonable_encoder
+from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
+from fastapi.middleware.cors import CORSMiddleware
+from starlette.concurrency import run_in_threadpool
+from anyio import to_thread # Added for AnyIO 4.x concurrency tuning
+from dotenv import load_dotenv
+from pydantic import BaseModel
+
+# --- Database Drivers ---
+from bson import ObjectId
+import mysql.connector
+import psycopg2
+from psycopg2.extras import RealDictCursor
+
+# --- Existing Services ---
+from csv_analysis_service import execute_analysis_logic
+from csv_chart_service import execute_python_code
+from csv_metadata_service import CsvDataRequest, CsvInfoRequest, CsvInfoResponse, PythonExecutionRequest, PythonExecutionResponse, execute_python_logic, get_csv_basic_info, get_robust_csv_rows
+from mongo_service import execute_mongo_operation, parse_query_input
+from pydantic_csv_analysis_model import AnalysisRequest, AnalysisResponse
+from pydantic_csv_charts_model import ChartExecutionPayload, ChartExecutionResponse
+from pydantic_mongo_executor_model import ExecutorPayload, ExecutorResponse
+from report_service import FileBoxProps, ReportRequest, execute_report_generation
+from supabase_service import upload_bytes_to_supabase
+
+# --- Configuration & Setup ---
+load_dotenv()
+
+logging.basicConfig(
+    format="%(asctime)s - %(name)s - %(levelname)s - %(message)s",
+    level=logging.INFO
+)
+logger = logging.getLogger("API_Controller")
+
+@asynccontextmanager
+async def lifespan(app: FastAPI):
+    # Startup logic
+    to_thread.current_default_thread_limiter().total_tokens = 2000
+    logger.info("Worker Process Started: Thread pool capacity set to 2000.")
+    yield
+    # Shutdown logic (if any) goes here
+
+app = FastAPI(
+    title="Unified Data Executor API (Mongo, SQL, CSV)", 
+    lifespan=lifespan
+)
+
+# ==============================================================================
+# HIGH-CONCURRENCY BATCH MODELS & STARTUP
+# ==============================================================================
+T = TypeVar("T")
+
+class BatchRequest(BaseModel, Generic[T]):
+    requests: List[T]
+
+class BatchResponse(BaseModel, Generic[T]):
+    responses: List[T]
+
+# --- Directory Setup ---
+CHART_DIR = "generated_charts"
+os.makedirs(CHART_DIR, exist_ok=True)
+
+# --- CORS ---
+origins_env = os.getenv("ALLOWED_ORIGINS", "*")
+ORIGINS = [origin.strip() for origin in origins_env.split(",")]
+
+app.add_middleware(
+    CORSMiddleware,
+    allow_origins=ORIGINS,
+    allow_credentials=True,
+    allow_methods=["*"],
+    allow_headers=["*"],
+)
+
+# --- Security ---
+security = HTTPBearer()
+API_SECRET_TOKEN = os.getenv("API_BEARER_TOKEN")
+
+if not API_SECRET_TOKEN:
+    logger.warning("WARNING: API_BEARER_TOKEN not set in .env file! Security is compromised.")
+
+async def validate_token(credentials: HTTPAuthorizationCredentials = Depends(security)):
+    if credentials.credentials != API_SECRET_TOKEN:
+        raise HTTPException(status_code=403, detail="Invalid Authentication Token")
+    return credentials.credentials
+
+
+# ==============================================================================
+# PYDANTIC MODELS
+# ==============================================================================
+
+# --- MySQL Models ---
+class SqlQueryRequest(BaseModel):
+    database_url: str
+    sql_query: str
+    limit_rows: Optional[int] = 20
+    limited: bool = False
+
+class SqlQueryResponse(BaseModel):
+    success: bool
+    results: Optional[List[Dict[str, Any]]] = None
+    columns: Optional[List[str]] = None
+    rowCount: Optional[int] = 0
+    executionTime: Optional[float] = 0.0
+    error: Optional[str] = None
+    request_id: str
+    is_aggregate: bool = False
+    limited: bool = False
+    message: Optional[str] = None
+
+# --- PostgreSQL Models ---
+class PgQueryRequest(BaseModel):
+    database_url: str
+    sql_query: str
+    limit_rows: Optional[int] = 20
+    limited: bool = False 
+
+class PgQueryResponse(BaseModel):
+    success: bool
+    results: Optional[List[Dict[str, Any]]] = None
+    columns: Optional[List[str]] = None
+    rowCount: Optional[int] = 0
+    executionTime: Optional[float] = 0.0
+    error: Optional[str] = None
+    request_id: str
+    is_aggregate: bool = False
+    limited: bool = False
+    message: Optional[str] = None
+
+
+# ==============================================================================
+# SHARED HELPER FUNCTIONS
+# ==============================================================================
+
+def is_aggregate_query(query: str) -> bool:
+    """Checks for aggregate keywords."""
+    query_lower = query.lower()
+    aggregate_patterns = [
+        r'\bcount\s*\(', r'\bsum\s*\(', r'\bavg\s*\(', 
+        r'\bmin\s*\(', r'\bmax\s*\(', r'\bgroup\s+by\b',
+        r'\bdistinct\b', r'\bhaving\b'
+    ]
+    for pattern in aggregate_patterns:
+        if re.search(pattern, query_lower):
+            return True
+    return False
+
+# ==============================================================================
+# MYSQL LOGIC
+# ==============================================================================
+
+def normalize_mysql_uri(uri: str) -> str:
+    try:
+        parsed_uri = urlparse(uri)
+        query_params = parse_qs(parsed_uri.query)
+        query_params.pop('ssl-mode', None)
+        new_query = urlencode(query_params, doseq=True)
+        parsed_uri = parsed_uri._replace(query=new_query)
+        return urlunparse(parsed_uri)
+    except Exception:
+        return uri
+
+def _run_mysql_synchronously(db_url: str, sql_query: str, max_rows: int = 20, limited: bool = False) -> dict:
+    start_time = time.time()
+    connection = None
+    cursor = None
+    response = {"success": False, "results": None, "columns": None, "rowCount": 0, "executionTime": 0.0, "error": None, "is_aggregate": False, "limited": False, "message": ""}
+
+    try:
+        parsed = urlparse(db_url)
+        db_config = {
+            "user": parsed.username, "password": parsed.password,
+            "host": parsed.hostname, "port": parsed.port or 3306,
+            "database": parsed.path.lstrip("/"), "connect_timeout": 5
+        }
+        connection = mysql.connector.connect(**db_config)
+        cursor = connection.cursor(dictionary=True)
+        
+        clean_query = sql_query.strip()
+        query_lower = clean_query.lower()
+
+        if not query_lower.startswith("select"):
+            cursor.execute(clean_query)
+            connection.commit()
+            response.update({"success": True, "message": "Query executed successfully (Non-SELECT)."})
+            return response
+
+        if not limited:
+            cursor.execute(clean_query)
+            results = cursor.fetchall()
+            response["message"] = f"Raw query executed. Returned {len(results)} row(s)."
+            response["limited"] = False
+            response["is_aggregate"] = is_aggregate_query(clean_query)
+        else:
+            if is_aggregate_query(clean_query):
+                cursor.execute(clean_query)
+                results = cursor.fetchall()
+                response["message"] = f"Aggregate query completed. Returned {len(results)} row(s)."
+                response["is_aggregate"] = True
+            else:
+                final_query = clean_query.rstrip(';').strip()
+                if not re.search(r'\blimit\s+\d+', query_lower):
+                    final_query = f"{final_query} LIMIT {max_rows}"
+                
+                cursor.execute(final_query)
+                results = cursor.fetchall()
+                
+                is_limited_result = (len(results) == max_rows)
+                response["message"] = f"Showing first {max_rows} rows only." if is_limited_result else f"Returned {len(results)} rows."
+                response["limited"] = is_limited_result
+                response["is_aggregate"] = False
+
+        columns = [col[0] for col in cursor.description] if cursor.description else []
+        response.update({"success": True, "results": jsonable_encoder(results), "columns": columns, "rowCount": len(results), "executionTime": time.time() - start_time})
+        return response
+
+    except Exception as e:
+        response["error"] = str(e)
+        return response
+    finally:
+        if cursor: cursor.close()
+        if connection and connection.is_connected(): connection.close()
+
+# ==============================================================================
+# POSTGRES LOGIC
+# ==============================================================================
+
+def normalize_postgres_uri(uri: str) -> str:
+    try:
+        parsed_uri = urlparse(uri)
+        if parsed_uri.scheme == 'postgres':
+            parsed_uri = parsed_uri._replace(scheme='postgresql')
+        return urlunparse(parsed_uri)
+    except Exception:
+        return uri
+
+def _run_postgres_synchronously(db_url: str, sql_query: str, max_rows: int = 20, limited: bool = False) -> dict:
+    start_time = time.time()
+    connection = None
+    cursor = None
+    response = {"success": False, "results": None, "columns": None, "rowCount": 0, "executionTime": 0.0, "error": None, "is_aggregate": False, "limited": False, "message": ""}
+    try:
+        parsed = urlparse(db_url)
+        qs = parse_qs(parsed.query)
+        sslmode = qs.get('sslmode', ['require'])[0] if 'sslmode' in qs else 'prefer'
+        db_config = {"host": parsed.hostname, "port": parsed.port or 5432, "database": parsed.path.lstrip("/"), "user": parsed.username, "password": parsed.password, "sslmode": sslmode, "connect_timeout": 5}
+        connection = psycopg2.connect(**db_config)
+        cursor = connection.cursor(cursor_factory=RealDictCursor)
+        clean_query = sql_query.strip()
+        query_lower = clean_query.lower()
+        if not query_lower.startswith(("select", "show", "explain", "with")):
+            cursor.execute(clean_query)
+            connection.commit()
+            response.update({"success": True, "message": "Query executed successfully (Non-SELECT)."})
+            return response
+        if not limited:
+            cursor.execute(clean_query)
+            results = cursor.fetchall()
+            response["message"] = f"Raw query executed. Returned {len(results)} row(s)."
+            response["limited"] = False
+            response["is_aggregate"] = is_aggregate_query(clean_query)
+        else:
+            if is_aggregate_query(clean_query):
+                cursor.execute(clean_query)
+                results = cursor.fetchall()
+                response["message"] = f"Aggregate query completed. Returned {len(results)} row(s)."
+                response["is_aggregate"] = True
+                response["limited"] = False
+            else:
+                final_query = clean_query.rstrip(';').strip()
+                if not re.search(r'\blimit\s+\d+', query_lower):
+                    final_query = f"{final_query} LIMIT {max_rows}"
+                cursor.execute(final_query)
+                results = cursor.fetchall()
+                is_limited_result = (len(results) == max_rows)
+                response["message"] = f"Showing first {max_rows} rows only." if is_limited_result else f"Returned {len(results)} rows."
+                response["limited"] = is_limited_result
+                response["is_aggregate"] = False
+        columns = [desc[0] for desc in cursor.description] if cursor.description else []
+        clean_results = jsonable_encoder(results, custom_encoder={uuid.UUID: str, ObjectId: str})
+        response.update({"success": True, "results": clean_results, "columns": columns, "rowCount": len(results), "executionTime": time.time() - start_time})
+        return response
+    except Exception as e:
+        if connection: connection.rollback()
+        response["error"] = str(e)
+        return response
+    finally:
+        if cursor: cursor.close()
+        if connection: connection.close()
+
+
+# ==============================================================================
+# API ROUTES
+# ==============================================================================
+
+@app.post("/api/execute_mongo", response_model=ExecutorResponse)
+async def execute_mongo_endpoint(payload: ExecutorPayload, token: str = Depends(validate_token)):
+    request_id = str(uuid.uuid4())[:8]
+    start_time = time.time()
+    try:
+        parsed_query = parse_query_input(payload.generated_query)
+        result_data = await run_in_threadpool(execute_mongo_operation, mongo_uri=payload.mongo_uri, db_name=payload.db_name, collection_name=payload.collection_name, query=parsed_query, limited=payload.limited, limit_rows=payload.limit_rows)
+        return ExecutorResponse(status="success", count=len(result_data), data=jsonable_encoder(result_data, custom_encoder={ObjectId: str}), duration_seconds=round(time.time() - start_time, 4), request_id=request_id)
+    except Exception as e:
+        raise HTTPException(status_code=500, detail=str(e))
+
+@app.post("/api/execute_chart", response_model=ChartExecutionResponse)
+async def execute_chart_endpoint(payload: ChartExecutionPayload, token: str = Depends(validate_token)):
+    request_id = str(uuid.uuid4())[:8]
+    try:
+        # 1. Execute Code
+        image_bytes, error_msg, logs = await run_in_threadpool(execute_python_code, code=payload.code, csv_url=payload.csv_url)
+        
+        if error_msg: 
+            return ChartExecutionResponse(status="error", error=error_msg, output_log=logs, request_id=request_id)
+
+        # 2. Handle Output Format
+        if payload.return_base64:
+            # OPTION A: Return Base64 (No Supabase Upload)
+            base64_str = base64.b64encode(image_bytes).decode('utf-8')
+            return ChartExecutionResponse(
+                status="success", 
+                base64_image=base64_str, 
+                output_log=logs, 
+                request_id=request_id
+            )
+        else:
+            # OPTION B: Upload to Supabase (Standard behavior)
+            unique_name = f"{uuid.uuid4()}.png"
+            public_url = await run_in_threadpool(upload_bytes_to_supabase, image_bytes=image_bytes, file_name=unique_name, chat_id=payload.chat_id)
+            return ChartExecutionResponse(
+                status="success", 
+                image_url=public_url, 
+                output_log=logs, 
+                request_id=request_id
+            )
+
+    except Exception as e:
+        raise HTTPException(status_code=500, detail=str(e))
+
+@app.post("/api/execute_sql_query", response_model=SqlQueryResponse)
+async def execute_mysql_endpoint(query: SqlQueryRequest, token: str = Depends(validate_token)):
+    request_id = str(uuid.uuid4())[:8]
+    try:
+        normalized_url = normalize_mysql_uri(query.database_url)
+        limit_val = query.limit_rows if query.limit_rows is not None else 20
+        result_dict = await run_in_threadpool(_run_mysql_synchronously, db_url=normalized_url, sql_query=query.sql_query, max_rows=limit_val, limited=query.limited)
+        result_dict["request_id"] = request_id
+        return SqlQueryResponse(**result_dict)
+    except Exception as e:
+        raise HTTPException(status_code=500, detail={"success": False, "error": str(e), "request_id": request_id})
+
+@app.post("/api/execute_postgres_query", response_model=PgQueryResponse)
+async def execute_postgres_endpoint(query: PgQueryRequest, token: str = Depends(validate_token)):
+    request_id = str(uuid.uuid4())[:8]
+    try:
+        clean_url = normalize_postgres_uri(query.database_url)
+        limit_val = query.limit_rows if query.limit_rows is not None else 20
+        result_dict = await run_in_threadpool(_run_postgres_synchronously, db_url=clean_url, sql_query=query.sql_query, max_rows=limit_val, limited=query.limited)
+        result_dict["request_id"] = request_id
+        return PgQueryResponse(**result_dict)
+    except Exception as e:
+        raise HTTPException(status_code=500, detail={"success": False, "error": str(e), "request_id": request_id})
+
+@app.post("/api/execute_csv_analysis", response_model=AnalysisResponse)
+async def execute_analysis_endpoint(payload: AnalysisRequest, token: str = Depends(validate_token)):
+    request_id = str(uuid.uuid4())[:8]
+    try:
+        result = await run_in_threadpool(execute_analysis_logic, code=payload.code, csv_url=payload.csv_url)
+        return AnalysisResponse(success=result["success"], output_log=result["output_log"], results=result["results"], error=result["error"], request_id=request_id)
+    except Exception as e:
+        raise HTTPException(status_code=500, detail=str(e))
+    
+@app.post("/api/generate_report", response_model=FileBoxProps)
+async def generate_report_endpoint(payload: ReportRequest, token: str = Depends(validate_token)):
+    try:
+        result = await execute_report_generation(code=payload.code, csv_url=payload.csv_url, chat_id=payload.chat_id)
+        return result
+    except Exception as e:
+        raise HTTPException(status_code=500, detail=str(e))
+    
+@app.post("/api/get_csv_info", response_model=CsvInfoResponse)
+async def get_csv_info_endpoint(payload: CsvInfoRequest, token: str = Depends(validate_token)):
+    request_id = str(uuid.uuid4())[:8]
+    start_time = time.time()
+    try:
+        info_result = await run_in_threadpool(get_csv_basic_info, csv_path=payload.csv_url)
+        if "error" in info_result:
+            return CsvInfoResponse(success=False, error=info_result["error"], request_id=request_id, duration=time.time() - start_time)
+        return CsvInfoResponse(success=True, data=info_result, request_id=request_id, duration=time.time() - start_time)
+    except Exception as e:
+        raise HTTPException(status_code=500, detail={"success": False, "error": str(e), "request_id": request_id})
+        
+@app.post("/api/csv_data")
+async def get_csv_data_endpoint(payload: CsvDataRequest, token: str = Depends(validate_token)):
+    try:
+        result = await run_in_threadpool(get_robust_csv_rows, csv_url=payload.csv_url)
+        if isinstance(result, dict) and "error" in result: raise HTTPException(status_code=400, detail=result["error"])
+        return result
+    except Exception as e:
+        raise HTTPException(status_code=500, detail=f"Internal server error: {str(e)}")
+
+@app.post("/api/execute_python", response_model=PythonExecutionResponse)
+async def execute_python_endpoint(payload: PythonExecutionRequest, token: str = Depends(validate_token)):
+    request_id = str(uuid.uuid4())[:8]
+    try:
+        execution_result = await run_in_threadpool(execute_python_logic, code=payload.code, custom_context=payload.context)
+        return PythonExecutionResponse(success=execution_result['error'] is None, output=execution_result['output'], result=jsonable_encoder(execution_result['result']), isStructured=execution_result['isStructured'], error=execution_result['error'], request_id=request_id)
+    except Exception as e:
+        raise HTTPException(status_code=500, detail={"success": False, "error": str(e), "request_id": request_id})
+
+# ==============================================================================
+# NEW: BATCH HANDLER LOGIC (Scaling for 1000+ Requests)
+# ==============================================================================
+
+async def batch_parallel_handler(func, requests: List[Any], token: str):
+    """
+    Executes multiple requests in parallel within a single worker process 
+    using asyncio.gather, utilizing the high-token thread pool.
+    """
+    tasks = [func(req, token) for req in requests]
+    results = await asyncio.gather(*tasks, return_exceptions=True)
+    return [res if not isinstance(res, Exception) else {"success": False, "error": str(res)} for res in results]
+
+@app.post("/api/batch/execute_sql_query", response_model=BatchResponse[SqlQueryResponse])
+async def batch_execute_sql(payload: BatchRequest[SqlQueryRequest], token: str = Depends(validate_token)):
+    responses = await batch_parallel_handler(execute_mysql_endpoint, payload.requests, token)
+    return BatchResponse(responses=responses)
+
+@app.post("/api/batch/execute_postgres_query", response_model=BatchResponse[PgQueryResponse])
+async def batch_execute_pg(payload: BatchRequest[PgQueryRequest], token: str = Depends(validate_token)):
+    responses = await batch_parallel_handler(execute_postgres_endpoint, payload.requests, token)
+    return BatchResponse(responses=responses)
+
+@app.post("/api/batch/execute_mongo", response_model=BatchResponse[ExecutorResponse])
+async def batch_execute_mongo(payload: BatchRequest[ExecutorPayload], token: str = Depends(validate_token)):
+    responses = await batch_parallel_handler(execute_mongo_endpoint, payload.requests, token)
+    return BatchResponse(responses=responses)
+
+# ==============================================================================
+# HIGH PERFORMANCE SERVER EXECUTION
+# ==============================================================================
+
+if __name__ == "__main__":
+    import uvicorn
+    host = os.getenv("HOST", "0.0.0.0")
+    port = int(os.getenv("PORT", 8000))
+    
+    # Scale processes: 16 Cores - 2 = 14 Workers
+    num_workers = max(1, multiprocessing.cpu_count() - 2)
+    
+    print(f"Starting production server on {host}:{port} with {num_workers} workers...")
+
+    uvicorn.run(
+        "controller:app",  
+        host=host, 
+        port=port, 
+        workers=num_workers,
+        loop="auto",
+    )

csv_analysis_service.py

@@ -0,0 +1,215 @@
+import pandas as pd
+import numpy as np
+import io
+import contextlib
+import traceback
+import json
+import logging
+import math
+import re
+from datetime import datetime, date, timedelta
+from typing import Any, Dict, Optional, Union
+
+# --- Robust Library Loading ---
+# We try to import common analysis libraries. 
+# If they are missing, the service still runs, just without those specifics.
+try:
+    import scipy
+    import scipy.stats as stats
+except ImportError:
+    scipy = None
+    stats = None
+
+try:
+    import sklearn
+    from sklearn.linear_model import LinearRegression, LogisticRegression
+    from sklearn.model_selection import train_test_split
+    from sklearn import metrics
+except ImportError:
+    sklearn = None
+
+try:
+    import statsmodels.api as sm
+    import statsmodels.formula.api as smf
+except ImportError:
+    sm = None
+    smf = None
+
+# Configure Logging
+logging.basicConfig(level=logging.INFO)
+logger = logging.getLogger("CSV_Analysis_Executor")
+
+def robust_json_serializer(obj: Any) -> Any:
+    """
+    Universal Serializer.
+    1. Handles DataFrames/Series (returns FULL data).
+    2. Handles NumPy types (int/float/arrays).
+    3. Handles Dates.
+    4. Handles Unknown Objects (Models, Classes) -> returns String Repr.
+    """
+    # 1. Pandas Types
+    if isinstance(obj, pd.DataFrame):
+        # UNCONSTRAINED: Return the full dataset as list of dicts
+        return obj.to_dict(orient='records')
+    elif isinstance(obj, pd.Series):
+        return obj.to_dict()
+    elif isinstance(obj, pd.Index):
+        return obj.tolist()
+    
+    # 2. NumPy Types
+    elif isinstance(obj, np.integer):
+        return int(obj)
+    elif isinstance(obj, np.floating):
+        if np.isnan(obj) or np.isinf(obj):
+            return None
+        return float(obj)
+    elif isinstance(obj, np.ndarray):
+        return obj.tolist()
+    elif isinstance(obj, np.bool_):
+        return bool(obj)
+    
+    # 3. Standard Python Dates
+    elif isinstance(obj, (datetime, date)):
+        return obj.isoformat()
+    elif isinstance(obj, timedelta):
+        return str(obj)
+    
+    # 4. Fallback for Complex Objects (e.g., sklearn model, statsmodels result)
+    # Instead of crashing, we return the string representation
+    if hasattr(obj, '__dict__'):
+        return str(obj)
+    
+    return str(obj)
+
+class CsvAnalysisExecutor:
+    """
+    A 'Natural' Executor. 
+    It mimics a local Jupyter notebook environment with common libraries pre-loaded.
+    """
+    def __init__(self, df: pd.DataFrame):
+        # NATURAL MODE: We do NOT sanitize column names. 
+        # We rely on the code generator to handle keys like df['User ID'] correctly.
+        self.df = df
+        self.exec_locals = {}
+
+    def execute(self, code: str) -> Dict[str, Any]:
+        """
+        Executes code with a rich data science context.
+        """
+        output_buffer = io.StringIO()
+        error_result = None
+        success = False
+
+        # --- Rich Environment Injection ---
+        exec_globals = {
+            # Core
+            'pd': pd,
+            'np': np,
+            'df': self.df,
+            'math': math,
+            're': re,
+            'json': json,
+            'datetime': datetime,
+            'timedelta': timedelta,
+            
+            # Statistics & ML (if available)
+            'scipy': scipy,
+            'stats': stats,
+            'sklearn': sklearn,
+            'sm': sm,
+            'smf': smf,
+            
+            # Common shortcuts (makes generated code more natural)
+            'LinearRegression': LinearRegression if sklearn else None,
+            'train_test_split': train_test_split if sklearn else None,
+        }
+
+        try:
+            # Capture standard output (print statements)
+            with contextlib.redirect_stdout(output_buffer):
+                exec(code, exec_globals, self.exec_locals)
+            
+            success = True
+
+        except Exception:
+            # Clean traceback for the user
+            error_result = traceback.format_exc()
+            logger.error(f"Analysis Execution Error:\n{error_result}")
+
+        # --- Extract "Natural" Results ---
+        # We capture everything the user defined, excluding imports and modules
+        final_vars = {}
+        
+        # List of system modules to ignore in output
+        ignored_types = (type(pd), type(np), type(math), type(json))
+        
+        for k, v in self.exec_locals.items():
+            # Skip hidden vars
+            if k.startswith('_'): continue
+            
+            # Skip the input dataframe ref (unless they made a copy)
+            if k == 'df': continue
+            
+            # Skip modules (e.g., if user did 'import random', don't return the random module)
+            if isinstance(v, ignored_types) or hasattr(v, '__name__') and 'module' in str(type(v)):
+                continue
+
+            # Capture the variable
+            final_vars[k] = v
+
+        return {
+            "success": success,
+            "output_log": output_buffer.getvalue(),
+            "error": error_result,
+            "data_vars": final_vars
+        }
+
+def execute_analysis_logic(code: str, csv_url: str) -> Dict[str, Any]:
+    """
+    Entry point. Loads CSV (handling errors) and executes logic.
+    """
+    try:
+        # 1. Load Data
+        logger.info(f"Loading CSV from: {csv_url}")
+        try:
+            # Use 'on_bad_lines' to be robust against messy CSVs
+            df = pd.read_csv(csv_url, on_bad_lines='skip')
+        except Exception as e:
+            return {
+                "success": False,
+                "error": f"Failed to load CSV: {str(e)}",
+                "output_log": "",
+                "results": {}
+            }
+        
+        # 2. Execute
+        executor = CsvAnalysisExecutor(df)
+        result = executor.execute(code)
+        
+        # 3. Serialize (Unconstrained)
+        clean_vars = {}
+        if result["data_vars"]:
+            try:
+                # Force strictly valid JSON
+                serialized_str = json.dumps(result["data_vars"], default=robust_json_serializer)
+                clean_vars = json.loads(serialized_str)
+            except Exception as e:
+                # Fallback mechanism
+                logger.error(f"Serialization Warning: {e}")
+                clean_vars = {"serialization_error": str(e), "raw_str_dump": str(result["data_vars"])}
+
+        return {
+            "success": result["success"],
+            "error": result["error"],
+            "output_log": result["output_log"],
+            "results": clean_vars
+        }
+
+    except Exception as e:
+        err_msg = f"System Error: {str(e)}\n{traceback.format_exc()}"
+        return {
+            "success": False,
+            "error": err_msg,
+            "output_log": "",
+            "results": {}
+        }

csv_chart_service.py

@@ -0,0 +1,68 @@
+# --- Data Science Stack ---
+from typing import Optional, Tuple
+import uuid
+import pandas as pd
+import numpy as np
+import matplotlib
+import matplotlib.pyplot as plt
+import seaborn as sns
+import datetime as dt
+import io
+import contextlib
+import os
+import traceback
+from dotenv import load_dotenv
+
+load_dotenv()
+
+# Set Matplotlib to non-interactive mode (server backend)
+matplotlib.use('Agg')
+
+def execute_python_code(code: str, csv_url: Optional[str] = None) -> Tuple[Optional[bytes], Optional[str], str]:
+    """
+    Executes Python code and captures the Matplotlib plot into memory bytes.
+    """
+    # 1. Define available libraries
+    local_scope = {
+        "pd": pd, "np": np, "plt": plt, "sns": sns, "dt": dt,
+        "uuid": uuid, "os": os, "csv_url": csv_url
+    }
+
+    # 2. [FIX] Load the CSV into 'df' so the AI code can find it
+    if csv_url:
+        try:
+            # Read the CSV from the URL
+            df = pd.read_csv(csv_url)
+            # Inject it into the local_scope with the variable name 'df'
+            local_scope["df"] = df
+        except Exception as e:
+            # Return early if we can't even load the data
+            return None, f"System Error: Failed to load CSV data. {str(e)}", ""
+
+    stdout_capture = io.StringIO()
+    
+    try:
+        plt.clf()
+        plt.close('all')
+        
+        with contextlib.redirect_stdout(stdout_capture):
+            # 3. Execute the code (now 'df' is defined in local_scope)
+            exec(code, {}, local_scope)
+            
+        # Capture the current figure into a BytesIO buffer
+        buf = io.BytesIO()
+        fig = plt.gcf()
+        
+        # Check if the figure actually has content (axes)
+        if not fig.get_axes():
+            return None, "No plot was generated by the code.", stdout_capture.getvalue()
+            
+        fig.savefig(buf, format='png', bbox_inches='tight')
+        buf.seek(0)
+        image_bytes = buf.getvalue()
+        buf.close()
+
+        return image_bytes, None, stdout_capture.getvalue()
+            
+    except Exception:
+        return None, traceback.format_exc(), stdout_capture.getvalue()

csv_metadata_service.py

@@ -0,0 +1,265 @@
+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
+    }

mongo_service.py

@@ -0,0 +1,158 @@
+
+import logging
+from typing import Dict, List, Union
+from bson import ObjectId
+import ast
+import json
+import re
+from pymongo.collation import Collation
+from pymongo import MongoClient
+
+logging.basicConfig(
+    format="%(asctime)s - %(name)s - %(levelname)s - %(message)s",
+    level=logging.INFO
+)
+logger = logging.getLogger("mongo_service")
+
+def convert_oid(obj):
+    """Recursively convert $oid format or ObjectId objects to strings."""
+    if isinstance(obj, ObjectId):
+        return str(obj)
+    if isinstance(obj, dict):
+        if set(obj.keys()) == {"$oid"}:
+            return str(obj["$oid"])
+        return {k: convert_oid(v) for k, v in obj.items()}
+    elif isinstance(obj, list):
+        return [convert_oid(item) for item in obj]
+    else:
+        return obj
+
+def parse_query_input(query_input: Union[str, Dict, List]) -> Union[Dict, List]:
+    """Ensures the input is a valid MongoDB query object (Dict or List)."""
+    if isinstance(query_input, (dict, list)):
+        return convert_oid(query_input)
+
+    query_str = str(query_input).strip()
+    
+    # 1. Try JSON
+    try:
+        return json.loads(query_str)
+    except json.JSONDecodeError:
+        pass
+
+    # 2. Extract code block
+    match = re.search(r"```(?:python|json|javascript)?\s*(.*?)\s*```", query_str, re.DOTALL)
+    if match:
+        query_str = match.group(1).strip()
+
+    # 3. Clean up MongoDB shell syntax
+    query_str = re.sub(r'^db\.\w+\.\w+\(', '', query_str)
+    if query_str.endswith(')'):
+        query_str = query_str[:-1]
+
+    # 4. AST Literal Eval
+    try:
+        parsed = ast.literal_eval(query_str)
+        return convert_oid(parsed)
+    except (ValueError, SyntaxError) as e:
+        logger.error(f"Failed to parse query string: {e}")
+        raise ValueError(f"Could not parse query string: {str(e)}")
+
+def get_value_ignore_case(d: Dict, keys: List[str], default=None):
+    """Helper to get value checking multiple key variations (camelCase/snake_case)."""
+    for k in keys:
+        if k in d:
+            return d[k]
+    return default
+
+def execute_mongo_operation(
+    mongo_uri: str, 
+    db_name: str, 
+    collection_name: str, 
+    query: Union[Dict, List],
+    limited: bool = False,   # <--- Added
+    limit_rows: int = 20     # <--- Added
+):
+    """
+    Generic MongoDB Executor.
+    Supports: Aggregation Pipelines (List) and Find Operations (Dict).
+    Now supports explicit row limiting.
+    """
+    client = None
+    try:
+        client = MongoClient(mongo_uri, serverSelectionTimeoutMS=5000, connectTimeoutMS=5000)
+        db = client[db_name]
+        collection = db[collection_name]
+        
+        results = []
+        
+        # --- Aggregation ---
+        if isinstance(query, list):
+            logger.info("Executing Aggregation Pipeline")
+            
+            # Apply Limit if requested and not already present at the end
+            if limited:
+                # Check if the last stage is already a $limit
+                if not (query and "$limit" in query[-1]):
+                    logger.info(f"Injecting $limit: {limit_rows} to pipeline")
+                    query.append({"$limit": limit_rows})
+            
+            cursor = collection.aggregate(query, allowDiskUse=True)
+            results = list(cursor)
+            
+        # --- Find / Command ---
+        elif isinstance(query, dict):
+            logger.info("Executing Structured Find Query")
+            
+            command_keys = {'filter', 'query', '$query', 'projection', 'sort', 'limit', 'skip'}
+            has_command_keys = bool(set(query.keys()) & command_keys)
+            
+            cursor = None
+
+            if has_command_keys and ('filter' in query or 'query' in query or '$query' in query):
+                # It is a Structured Command
+                query_filter = get_value_ignore_case(query, ['filter', 'query', '$query'], {})
+                projection = get_value_ignore_case(query, ['projection', 'fields'], None)
+                if projection == {}: projection = None 
+                
+                # Check internal limit
+                internal_limit = int(get_value_ignore_case(query, ['limit'], 0))
+                skip = int(get_value_ignore_case(query, ['skip'], 0))
+                sort_val = get_value_ignore_case(query, ['sort', '$orderby'], None)
+                
+                cursor = collection.find(query_filter, projection)
+                
+                if sort_val:
+                    if isinstance(sort_val, dict):
+                        sort_val = list(sort_val.items())
+                    cursor = cursor.sort(sort_val)
+                
+                if skip > 0: cursor = cursor.skip(skip)
+                
+                # Logic: If 'limited' is forced (Chat Mode), use limit_rows. 
+                # Otherwise use the query's internal limit if it exists.
+                if limited:
+                    cursor = cursor.limit(limit_rows)
+                elif internal_limit > 0: 
+                    cursor = cursor.limit(internal_limit)
+
+            else:
+                # It is a Raw Filter
+                logger.info("Treating input dictionary as Raw Filter")
+                cursor = collection.find(query)
+                
+                if limited:
+                    cursor = cursor.limit(limit_rows)
+            
+            results = list(cursor)
+        else:
+            raise ValueError("Query must be a Dictionary (find) or List (aggregate)")
+
+        return results
+
+    except Exception as e:
+        logger.error(f"DB Execution Error: {e}")
+        raise e
+    finally:
+        if client:
+            client.close()

pydantic_csv_analysis_model.py

@@ -0,0 +1,14 @@
+from typing import Any, Dict, Optional
+from pydantic import BaseModel
+
+
+class AnalysisRequest(BaseModel):
+    csv_url: str
+    code: str
+
+class AnalysisResponse(BaseModel):
+    success: bool
+    output_log: str
+    results: Dict[str, Any]  # Unconstrained Dictionary
+    error: Optional[str] = None
+    request_id: str

pydantic_csv_charts_model.py

@@ -0,0 +1,19 @@
+
+# --- Chart Models ---
+from typing import Optional
+from pydantic import BaseModel, Field
+
+# --- Chart Models ---
+class ChartExecutionPayload(BaseModel):
+    code: str = Field(..., description="The Python code to execute.")
+    csv_url: Optional[str] = Field(None, description="Optional CSV URL to be injected into scope")
+    chat_id: str = Field(..., description="Chat ID required for Supabase mapping")
+    return_base64: bool = Field(False, description="If True, returns base64 string instead of uploading to Supabase") 
+
+class ChartExecutionResponse(BaseModel):
+    status: str
+    image_url: Optional[str] = None
+    base64_image: Optional[str] = None 
+    output_log: Optional[str] = None
+    error: Optional[str] = None
+    request_id: str

pydantic_mongo_executor_model.py

@@ -0,0 +1,21 @@
+
+# --- Mongo Models ---
+from typing import Any, Dict, List, Optional, Union
+from pydantic import BaseModel, Field
+
+
+class ExecutorPayload(BaseModel):
+    mongo_uri: str = Field(..., description="MongoDB connection string")
+    db_name: str = Field(..., description="Target Database Name")
+    collection_name: str = Field(..., description="Target Collection Name")
+    generated_query: Union[Dict, List, str] = Field(..., description="The query to execute (Find dict or Aggregation list)")
+    user_query: Optional[str] = Field(None, description="Original user question for logging context")
+    limited: bool = False
+    limit_rows: int = 20
+
+class ExecutorResponse(BaseModel):
+    status: str
+    count: int
+    data: Any
+    duration_seconds: float
+    request_id: str

report_service.py

@@ -0,0 +1,256 @@
+import pandas as pd
+import matplotlib.pyplot as plt
+import seaborn as sns
+import datetime as dt
+import os
+import uuid
+import json
+import logging
+import sys
+import traceback
+import shutil
+import ast  # <--- Added for syntax checking
+from io import StringIO
+from typing import List, Dict, Any, Optional
+from pydantic import BaseModel
+from starlette.concurrency import run_in_threadpool
+import numpy as np
+
+# --- Internal Imports ---
+from supabase_service import upload_file_to_supabase
+
+# Configure Logging
+logging.basicConfig(level=logging.INFO)
+logger = logging.getLogger("Report_Generator")
+
+# Configure Matplotlib backend for thread safety (Headless)
+os.environ['MPLBACKEND'] = 'agg'
+
+# Disable plt.show globally to prevent blocking
+plt.show = lambda: None
+
+# ==============================================================================
+# REQUEST MODELS
+# ==============================================================================
+class ReportRequest(BaseModel):
+    code: str
+    csv_url: str
+    chat_id: str
+
+# ==============================================================================
+# RESPONSE MODELS
+# ==============================================================================
+
+class FileProps(BaseModel):
+    fileName: str
+    filePath: str
+    fileType: str  # 'csv' | 'image'
+
+class Files(BaseModel):
+    csv_files: List[FileProps]
+    image_files: List[FileProps]
+
+class FileBoxProps(BaseModel):
+    files: Files
+
+# ==============================================================================
+# EXECUTION ENGINE
+# ==============================================================================
+
+class PythonREPL:
+    """Secure Python REPL with file generation tracking"""
+    
+    def __init__(self, df: pd.DataFrame):
+        self.df = df
+        # Create a unique directory for this execution to avoid thread collisions
+        self.output_dir = os.path.abspath(f'generated_outputs/{uuid.uuid4()}')
+        os.makedirs(self.output_dir, exist_ok=True)
+        
+        self.local_env = {
+            "pd": pd,
+            "df": self.df.copy(),
+            "plt": plt,
+            "os": os,
+            "uuid": uuid,
+            "sns": sns,
+            "json": json,
+            "dt": dt,
+            "np": np,
+            "output_dir": self.output_dir # INJECT output_dir so code can use it
+        }
+        
+    def execute(self, code: str) -> Dict[str, Any]:
+        logger.info(f'Executing code in: {self.output_dir}')
+        old_stdout = sys.stdout
+        sys.stdout = mystdout = StringIO()
+        
+        file_tracker = {
+            'csv_files': set(),
+            'image_files': set()
+        }
+
+        # Wrap code to ensure non-interactive plotting inside the exec scope
+        wrapped_code = f"""
+import matplotlib.pyplot as plt
+plt.switch_backend('agg')
+{code}
+plt.close('all')
+"""
+        error = False
+        error_msg = None
+
+        try:
+            # --- IMPROVEMENT: Syntax Validation ---
+            # This checks if the generated code is valid Python before running it.
+            # It catches the "missing parenthesis" error gracefully.
+            try:
+                ast.parse(wrapped_code)
+            except SyntaxError as e:
+                raise SyntaxError(f"Generated code is incomplete or invalid: {e}")
+
+            # Execute the code
+            exec(wrapped_code, self.local_env)
+            
+            # Check for generated files in the specific directory
+            if os.path.exists(self.output_dir):
+                for fname in os.listdir(self.output_dir):
+                    if fname.endswith('.csv'):
+                        file_tracker['csv_files'].add(fname)
+                    elif fname.lower().endswith(('.png', '.jpg', '.jpeg')):
+                        file_tracker['image_files'].add(fname)
+            
+        except Exception as e:
+            error = True
+            error_msg = traceback.format_exc()
+            
+            # --- IMPROVEMENT: Better Logging ---
+            # Log the code causing the error so you can debug the LLM prompt
+            logger.error("============= CODE EXECUTION FAILED =============")
+            logger.error(f"Error Message: {str(e)}")
+            logger.error("---------------- FAILED CODE ----------------")
+            for i, line in enumerate(wrapped_code.split('\n')):
+                logger.error(f"{i+1}: {line}")
+            logger.error("=================================================")
+            
+        finally:
+            sys.stdout = old_stdout
+            
+        return {
+            "output": mystdout.getvalue(),
+            "error": error,
+            "error_message": error_msg,
+            "output_dir": self.output_dir,
+            "files": {
+                "csv": [os.path.join(self.output_dir, f) for f in file_tracker['csv_files']],
+                "images": [os.path.join(self.output_dir, f) for f in file_tracker['image_files']]
+            }
+        }
+    
+    def cleanup(self):
+        """Remove the temporary directory"""
+        if os.path.exists(self.output_dir):
+            try:
+                shutil.rmtree(self.output_dir)
+            except Exception as e:
+                logger.warning(f"Cleanup failed: {e}")
+
+# ==============================================================================
+# MAIN LOGIC (Synchronous Worker)
+# ==============================================================================
+
+def _generate_report_sync(code: str, csv_url: str, chat_id: str) -> FileBoxProps:
+    """
+    Blocking worker function.
+    1. Reads CSV.
+    2. Runs Code.
+    3. Uploads files synchronously.
+    """
+    repl = None
+    csv_props = []
+    image_props = []
+    
+    try:
+        # 1. Load Data
+        df = pd.read_csv(csv_url)
+        
+        # 2. Initialize REPL
+        repl = PythonREPL(df)
+        
+        # 3. Execute Code
+        result = repl.execute(code)
+        
+        if result['error']:
+            # Log is already handled in repl.execute
+            return FileBoxProps(files=Files(csv_files=[], image_files=[]))
+
+        # 4. Process & Upload CSVs
+        for csv_path in result['files']['csv']:
+            if os.path.exists(csv_path):
+                file_name = os.path.basename(csv_path)
+                unique_name = f"{uuid.uuid4()}_{file_name}"
+                try:
+                    # Sync call - NO AWAIT
+                    public_url = upload_file_to_supabase(
+                        file_path=csv_path, 
+                        file_name=unique_name, 
+                        chat_id=chat_id
+                    )
+                    csv_props.append(FileProps(
+                        fileName=file_name,
+                        filePath=public_url,
+                        fileType="csv"
+                    ))
+                except Exception as e:
+                    logger.error(f"Failed upload CSV {file_name}: {e}")
+
+        # 5. Process & Upload Images
+        for img_path in result['files']['images']:
+            if os.path.exists(img_path):
+                file_name = os.path.basename(img_path)
+                unique_name = f"{uuid.uuid4()}_{file_name}"
+                try:
+                    # Sync call - NO AWAIT
+                    public_url = upload_file_to_supabase(
+                        file_path=img_path, 
+                        file_name=unique_name, 
+                        chat_id=chat_id
+                    )
+                    image_props.append(FileProps(
+                        fileName=file_name,
+                        filePath=public_url,
+                        fileType="image"
+                    ))
+                except Exception as e:
+                    logger.error(f"Failed upload Image {file_name}: {e}")
+
+        return FileBoxProps(
+            files=Files(
+                csv_files=csv_props,
+                image_files=image_props
+            )
+        )
+
+    except Exception as e:
+        logger.error(f"System Error in Report Generator: {e}")
+        return FileBoxProps(files=Files(csv_files=[], image_files=[]))
+        
+    finally:
+        # 6. Cleanup local files
+        if repl:
+            repl.cleanup()
+
+# ==============================================================================
+# ASYNC WRAPPER
+# ==============================================================================
+
+async def execute_report_generation(code: str, csv_url: str, chat_id: str) -> FileBoxProps:
+    """
+    Async entry point that runs the blocking logic in a separate thread.
+    This enables concurrency.
+    """
+    return await run_in_threadpool(
+        _generate_report_sync,
+        code=code,
+        csv_url=csv_url,
+        chat_id=chat_id
+    )

supabase_service.py

@@ -0,0 +1,105 @@
+import logging
+import os
+from typing import Optional
+from dotenv import load_dotenv
+from supabase import create_client, Client
+
+logging.basicConfig(
+    format="%(asctime)s - %(name)s - %(levelname)s - %(message)s",
+    level=logging.INFO
+)
+logger = logging.getLogger("supabase_service")
+
+load_dotenv()
+
+# --- Supabase Setup ---
+SUPABASE_URL = os.getenv("SUPABASE_URL")
+SUPABASE_KEY = os.getenv("SUPABASE_KEY")
+BUCKET_NAME = "csvcharts"
+
+# Initialize Supabase Client
+supabase: Optional[Client] = None
+if SUPABASE_URL and SUPABASE_KEY:
+    try:
+        supabase = create_client(SUPABASE_URL, SUPABASE_KEY)
+        logger.info("Supabase client initialized successfully.")
+    except Exception as e:
+        logger.error(f"Failed to initialize Supabase: {e}")
+else:
+    logger.warning("WARNING: SUPABASE_URL or SUPABASE_KEY not set. Uploads will fail.")
+    
+    
+
+def upload_file_to_supabase(file_path: str, file_name: str, chat_id: str) -> str:
+    """
+    Uploads an image to Supabase Storage and returns the public URL.
+    Saves the mapping (url, name, chat_id) in the DB.
+    """
+    if not supabase:
+        raise Exception("Supabase client is not initialized. Check .env variables.")
+
+    if not os.path.exists(file_path):
+        raise FileNotFoundError(f"The file {file_path} does not exist.")
+
+    with open(file_path, "rb") as f:
+        file_data = f.read()
+
+    # 1. Upload to Storage
+    try:
+        res = supabase.storage.from_(BUCKET_NAME).upload(file_name, file_data)
+        logger.info(f"Supabase upload response: {res}")
+    except Exception as e:
+        raise Exception(f"Failed to upload file to Supabase Storage: {e}")
+
+    # 2. Get Public URL
+    public_url = supabase.storage.from_(BUCKET_NAME).get_public_url(file_name)
+    logger.info(f"Generated Public URL: {public_url}")
+
+    # 3. Save mapping to Database
+    try:
+        supabase.table("chart_mappings").insert({
+            "public_url": public_url,
+            "chart_name": file_name,
+            "chat_id": chat_id
+        }).execute()
+    except Exception as e:
+        logger.error(f"Failed to save mapping to DB: {e}")
+        # Cleanup uploaded file to avoid orphans
+        try:
+            supabase.storage.from_(BUCKET_NAME).remove([file_name])
+        except:
+            pass
+        raise Exception(f"Failed to save mapping to database: {e}")
+
+    return public_url
+
+
+def upload_bytes_to_supabase(image_bytes: bytes, file_name: str, chat_id: str) -> str:
+    """
+    Uploads raw bytes directly to Supabase.
+    """
+    if not supabase:
+        raise Exception("Supabase client not initialized.")
+
+    # 1. Upload bytes directly
+    try:
+        # Supabase Python SDK accepts bytes directly
+        supabase.storage.from_(BUCKET_NAME).upload(
+            path=file_name,
+            file=image_bytes,
+            file_options={"content-type": "image/png"}
+        )
+    except Exception as e:
+        raise Exception(f"Supabase Storage error: {e}")
+
+    # 2. Get Public URL
+    public_url = supabase.storage.from_(BUCKET_NAME).get_public_url(file_name)
+
+    # 3. Save mapping to Database
+    supabase.table("chart_mappings").insert({
+        "public_url": public_url,
+        "chart_name": file_name,
+        "chat_id": chat_id
+    }).execute()
+
+    return public_url