CSV to SQL Importer — Python source

"""csv-to-sql — pure CSV → SQL import generator. Python port (canonical TS:
src/lib/csv-to-sql.ts; Go twin: cli/csv-to-sql). RFC 4180 parse, header
sanitizing into SQL identifiers, then batched INSERTs, a Postgres COPY block,
or a MySQL LOAD DATA statement. Numeric-looking text emits bare and verbatim;
empty fields become NULL. Never raises — errors travel in the result dict.
"""
import re

_NUMERIC = re.compile(r"^-?(\d+(\.\d+)?|\.\d+)([eE][+-]?\d+)?$")
_IDENT = re.compile(r"[^A-Za-z0-9_]")


def csv_to_rows(csv):
    """RFC 4180 parser mirroring the TS loop (lenient quotes, CR dropped)."""
    rows, field, row, in_q = [], [], [], False
    i, n = 0, len(csv)
    while i < n:
        ch = csv[i]
        if in_q:
            if ch == '"':
                if i + 1 < n and csv[i + 1] == '"':
                    field.append('"')
                    i += 1
                else:
                    in_q = False
            else:
                field.append(ch)
        elif ch == '"':
            in_q = True
        elif ch == ",":
            row.append("".join(field))
            field = []
        elif ch == "\n":
            row.append("".join(field))
            rows.append(row)
            row, field = [], []
        elif ch != "\r":
            field.append(ch)
        i += 1
    if field or row:
        row.append("".join(field))
        rows.append(row)
    return rows


def escape_sql_string(s, dialect="standard"):
    out = s.replace("'", "''")
    if dialect == "mysql":
        out = (out.replace("\\", "\\\\").replace("\0", "\\0").replace("\n", "\\n")
               .replace("\r", "\\r").replace("\x1a", "\\Z"))
    return out


def field_literal(value, dialect="standard", infer_types=True):
    if infer_types:
        if value == "":
            return "NULL"
        if _NUMERIC.match(value):
            return value  # verbatim — no float round-trip
    return "'" + escape_sql_string(value, dialect) + "'"


def sanitize_ident(name):
    return _IDENT.sub("_", name)


def sanitize_headers(headers):
    out, seen = [], {}
    for i, h in enumerate(headers):
        ident = sanitize_ident(h.strip()) or f"col{i + 1}"
        seen[ident] = seen.get(ident, 0) + 1
        if seen[ident] > 1:
            ident = f"{ident}_{seen[ident]}"
        out.append(ident)
    return out


def csv_escape(field):
    if any(c in field for c in ',\n\r"'):
        return '"' + field.replace('"', '""') + '"'
    return field


def csv_to_sql(csv, table, format="insert", dialect="standard", batch_size=100,
               infer_types=True, quote_identifiers=True, file_name="import.csv"):
    text = csv.strip()
    rows = csv_to_rows(text)
    if not text or len(rows) < 2:
        return {"ok": False, "sql": "", "rows": 0, "error": "No rows to import."}

    cols = sanitize_headers(rows[0])
    data = rows[1:]
    ident_dialect = {"copy": "postgres", "load-data": "mysql"}.get(format, dialect)

    def q(name):
        if not quote_identifiers:
            return name
        return f"`{name}`" if ident_dialect == "mysql" else f'"{name}"'

    tbl = q(sanitize_ident(table) or "tbl")
    col_list = ", ".join(q(c) for c in cols)

    if format in ("copy", "load-data"):
        payload = [",".join(csv_escape(c) for c in cols)]
        payload += [",".join(csv_escape(r[ci] if ci < len(r) else "") for ci in range(len(cols)))
                    for r in data]
        if format == "copy":
            sql = (f"COPY {tbl} ({col_list}) FROM STDIN WITH (FORMAT csv, HEADER true);\n"
                   + "\n".join(payload) + "\n\\.")
        else:
            clean = re.sub(r"[^A-Za-z0-9._\-/]", "", file_name) or "import.csv"
            sql = (f"LOAD DATA LOCAL INFILE '{clean}'\nINTO TABLE {tbl}\n"
                   "FIELDS TERMINATED BY ',' OPTIONALLY ENCLOSED BY '\"'\n"
                   "LINES TERMINATED BY '\\n'\nIGNORE 1 LINES;\n\n" + "\n".join(payload))
        return {"ok": True, "sql": sql, "rows": len(data), "error": None}

    size = max(1, batch_size)
    stmts = []
    for start in range(0, len(data), size):
        values = ["  (" + ", ".join(field_literal(r[ci] if ci < len(r) else "", dialect, infer_types)
                                    for ci in range(len(cols))) + ")"
                  for r in data[start:start + size]]
        stmts.append(f"INSERT INTO {tbl} ({col_list}) VALUES\n" + ",\n".join(values) + ";")
    return {"ok": True, "sql": "\n".join(stmts), "rows": len(data), "error": None}


# Example:
# csv_to_sql("id,name\n1,Ada\n2,", table="users")["sql"]
# → 'INSERT INTO "users" ("id", "name") VALUES\n  (1, \'Ada\'),\n  (2, NULL);'