Spaces:
Runtime error
Runtime error
File size: 1,675 Bytes
1914b78 | 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 | import sqlite3
import os
import pandas as pd
import logging
logger = logging.getLogger(__name__)
def import_sql_to_db(sql_path, db_path="output.db"):
if not os.path.exists(sql_path):
raise FileNotFoundError(f"{sql_path} not found")
conn = sqlite3.connect(db_path)
cursor = conn.cursor()
with open(sql_path, "r", encoding="utf-8") as f:
sql_script = f.read()
try:
cursor.executescript(sql_script) # ✅ executes full SQL dump
conn.commit()
print(f"✅ Database created at {db_path}")
return db_path
except Exception as e:
print(f"❌ Error: {e}")
finally:
conn.close()
def convert_csv_to_sqlite(csv_path, db_path="output.db", table_name=None):
if table_name is None:
table_name = os.path.splitext(os.path.basename(csv_path))[0]
df = pd.read_csv(csv_path)
conn = sqlite3.connect(db_path)
df.to_sql(table_name, conn, if_exists="replace", index=False)
conn.close()
print(f"✅ CSV converted to SQLite DB ({table_name} table)")
return db_path
def convert_to_sqlite(input_path, output_db="converted.db"):
ext = os.path.splitext(input_path)[1].lower()
if os.path.exists(output_db):
os.remove(output_db)
if ext == ".sql":
return import_sql_to_db(input_path, output_db)
elif ext in [".db", ".sqlite"]:
import shutil
shutil.copy(input_path, output_db)
print(f"✅ Copied DB to {output_db}")
return output_db
elif ext == ".csv":
return convert_csv_to_sqlite(input_path, output_db)
else:
raise ValueError(f"❌ Unsupported format: {ext}")
|