Spaces:
Runtime error
Runtime error
File size: 8,150 Bytes
37a6ee1 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 | import os
import sqlite3
import pandas as pd
from typing import List, Dict, Any
from backend import config
class CSVService:
def __init__(self):
print("[CSVService] Initializing CSVService singleton...")
# Ensure directories exist
os.makedirs(config.DB_DIR, exist_ok=True)
os.makedirs(config.SAMPLE_DATA_DIR, exist_ok=True)
def _infer_column_type(self, series: pd.Series) -> str:
"""
Helper method to infer SQLite-compatible column types from a pandas Series.
"""
try:
# If dtype is numeric, check if integer or float
if pd.api.types.is_integer_dtype(series):
return "INTEGER"
elif pd.api.types.is_float_dtype(series):
return "REAL"
# Check if it looks like a datetime
# We try to convert to datetime. If it succeeds without producing more than 10% NaNs (on non-empty series), it's likely a date/time.
non_null_series = series.dropna()
if not non_null_series.empty:
try:
converted = pd.to_datetime(non_null_series, errors='coerce')
null_pct = converted.isna().sum() / len(non_null_series)
if null_pct < 0.1:
return "DATE"
except Exception:
pass
return "TEXT"
except Exception as e:
print(f"[CSVService] Error inferring type: {e}. Defaulting to TEXT.")
return "TEXT"
def load_csv_to_sqlite(self, file_path: str, db_name: str) -> dict:
"""
Reads a CSV with pandas, infers column types, creates a SQLite database,
loads the dataframe into a table named 'data', and returns schema metadata.
"""
print(f"[CSVService] Loading CSV from {file_path} into database {db_name}.db")
try:
# Read CSV
df = pd.read_csv(file_path)
# Remove any leading/trailing whitespaces in string columns and header names
df.columns = [col.strip() for col in df.columns]
for col in df.select_dtypes(include=['object']):
df[col] = df[col].astype(str).str.strip()
db_path = os.path.join(config.DB_DIR, f"{db_name}.db")
conn = sqlite3.connect(db_path)
# Load into SQLite table 'data'
df.to_sql("data", conn, if_exists="replace", index=False)
# Determine columns metadata
columns_metadata = []
for col in df.columns:
inferred_type = self._infer_column_type(df[col])
# Convert numpy values to native Python types for JSON serialization
sample_vals = df[col].dropna().unique()[:3]
sample_vals_list = []
for val in sample_vals:
if hasattr(val, "item"):
sample_vals_list.append(val.item())
else:
sample_vals_list.append(val)
columns_metadata.append({
"name": col,
"type": inferred_type,
"sample_values": sample_vals_list
})
row_count = len(df)
conn.close()
result = {
"db_name": db_name,
"table_name": "data",
"columns": columns_metadata,
"row_count": row_count
}
print(f"[CSVService] Successfully loaded {row_count} rows into {db_name}.db table 'data'")
return result
except Exception as e:
print(f"[CSVService] Failed to load CSV {file_path} to SQLite: {e}")
raise ValueError(f"Failed to load CSV to SQLite: {str(e)}")
def get_table_schema(self, db_name: str) -> dict:
"""
Connects to db_name.db and retrieves the table 'data' schema details.
"""
print(f"[CSVService] Retrieving table schema for {db_name}.db")
try:
db_path = os.path.join(config.DB_DIR, f"{db_name}.db")
if not os.path.exists(db_path):
raise FileNotFoundError(f"Database {db_name}.db does not exist.")
conn = sqlite3.connect(db_path)
cursor = conn.cursor()
# Check row count
cursor.execute("SELECT COUNT(*) FROM data")
row_count = cursor.fetchone()[0]
# Retrieve table schema info
cursor.execute("PRAGMA table_info(data)")
columns_info = cursor.fetchall()
columns = []
for col_info in columns_info:
col_name = col_info[1]
col_type = col_info[2]
# Fetch up to 3 unique non-null sample values from SQLite
cursor.execute(f'SELECT DISTINCT "{col_name}" FROM data WHERE "{col_name}" IS NOT NULL LIMIT 3')
sample_vals = [row[0] for row in cursor.fetchall()]
columns.append({
"name": col_name,
"type": col_type,
"sample_values": sample_vals
})
conn.close()
schema = {
"table_name": "data",
"columns": columns,
"row_count": row_count
}
print(f"[CSVService] Retrieved schema for {db_name}.db with {row_count} rows.")
return schema
except Exception as e:
print(f"[CSVService] Failed to retrieve table schema for {db_name}: {e}")
raise ValueError(f"Failed to retrieve table schema: {str(e)}")
def get_sample_datasets(self) -> List[dict]:
"""
Reads all CSVs in sample_data/ folder. Load them to SQLite if not already loaded,
and returns details of each.
"""
print("[CSVService] Fetching sample datasets...")
try:
datasets = []
descriptions = {
"sales": "Monthly sales data across products, regions, and salespeople",
"employees": "Company HR data with salaries, departments, and performance",
"ecommerce": "Online store orders with products, customers, and payments"
}
if not os.path.exists(config.SAMPLE_DATA_DIR):
print(f"[CSVService] Sample data directory {config.SAMPLE_DATA_DIR} does not exist.")
return []
for filename in os.listdir(config.SAMPLE_DATA_DIR):
if filename.endswith(".csv"):
name = os.path.splitext(filename)[0]
file_path = os.path.join(config.SAMPLE_DATA_DIR, filename)
db_path = os.path.join(config.DB_DIR, f"{name}.db")
display_name = name.capitalize()
description = descriptions.get(name, f"Sample dataset containing {name} records")
# If not already loaded to SQLite, load it
if not os.path.exists(db_path):
print(f"[CSVService] Pre-loading sample CSV: {filename}")
schema_info = self.load_csv_to_sqlite(file_path, name)
else:
schema_info = self.get_table_schema(name)
datasets.append({
"name": name,
"display_name": display_name,
"description": description,
"columns": schema_info["columns"],
"row_count": schema_info["row_count"]
})
return datasets
except Exception as e:
print(f"[CSVService] Failed to fetch sample datasets: {e}")
raise ValueError(f"Failed to fetch sample datasets: {str(e)}")
# Singleton instance
csv_service = CSVService()
|