Fabric - Process all .csv in Lakehouse folder to delta tables. The .csv files are placed in a subfolder called 'input'

notebook.py

import os
import re

files_folder = "/lakehouse/default/Files/input"
csv_files = []

for root, dirs, files in os.walk(files_folder):
    for file in files:
        if file.endswith('.csv'):
            csv_files.append(os.path.join(root, file))

def sanitize_column(col_name):
    # Replace all forbidden characters with underscores
    return re.sub(r'[^A-Za-z0-9_]', '_', col_name)

for file_path in csv_files:
    print(f"Processing file: {file_path}")
    fname = os.path.basename(file_path)
    tablename = os.path.splitext(fname)[0]
    tablename = re.sub(r'[^A-Za-z0-9_]', '_', tablename)

    trimmed_path = file_path.replace("/lakehouse/default/", "")

    try:
        # Try both delimiters and pick the better one
        df_comma = spark.read.option("header", True).option("delimiter", ",").csv(trimmed_path)
        df_semi = spark.read.option("header", True).option("delimiter", ";").csv(trimmed_path)
        # Heuristics: pick the one with more columns (i.e., correctly parsed)
        df = df_comma if len(df_comma.columns) > len(df_semi.columns) else df_semi

        # Sanitize columns
        df = df.toDF(*[sanitize_column(col) for col in df.columns])

        df.write.mode("overwrite").option("overwriteSchema", "true").saveAsTable(f"dbo.{tablename}")
        print(f"Wrote table: dbo.{tablename} | Rows: {df.count()} Columns: {len(df.columns)}")
    except Exception as e:
        print(f"Error processing {file_path}: {str(e)}")