dataconvert

This commit is contained in:
2026-09-16 07:51:37 -03:00
parent e48e1a1e1c
commit 48ab1e663b
9 changed files with 434 additions and 168 deletions

View File

@@ -0,0 +1,45 @@
# dataconvert
Spreadsheets and CSV into schema-agnostic SQL seed files, one per table or sheet,
plus a `SCHEMA.md` describing every table.
```bash
uv run dataconvert.py --input data/ --out-dir seed/ # full seeds
uv run dataconvert.py --input data/ "*.xlsx" --out-dir seed/ # dirs, files, globs, .zip
uv run dataconvert.py --input data/ --out-dir sample/ --max-rows 20 # to understand the data
```
Reads `.csv`, `.xlsx`, `.xls` and `.ods`: single files, directories (recursively),
ZIP archives and wildcard patterns. Sheets of a multi-sheet workbook are written as
`<workbook>_<sheet>.sql` unless the sheet is already `drm_` or `study_`. Several
sources feeding the same table accumulate in one file, and so does re-running into
the same `--out-dir`, so point a fresh run at an empty directory.
## Sampling for a web LLM
Full seed files get large fast, and a model only needs to see the shape of the data.
With `--max-rows N`:
- every table still gets its `.sql` file, holding only its first N rows, with a header
line such as `-- SAMPLE: first 20 of 184233 rows. The full file would be ~48.1M`;
- `SCHEMA.md` lists every table with its total rows and the size the full seed file
would be, then each table's columns: original header, inferred type, null count
and one example value.
`SCHEMA.md` alone is often enough to hand to the model. Full sizes of sampled tables
are estimated from the rows written, so they carry a `~`. Types are inferred from
what pandas read: a starting point, not DDL. `--no-schema` skips the file.
## Layout
| file | does |
|---|---|
| `dataconvert.py` | command line |
| `readers.py` | files, directories, ZIPs and globs into DataFrames |
| `sqlgen.py` | DataFrames into INSERT statements, with the row cap |
| `output.py` | file naming and writing |
| `schema.py` | `SCHEMA.md` |
The modules import each other by name, so the folder works wherever it is copied:
`uv run dataconvert.py` from inside it, or `python3 path/to/dataconvert.py` with
pandas, openpyxl and odfpy installed.

View File

@@ -0,0 +1,58 @@
#!/usr/bin/env python3
"""
dataconvert
Converts CSV, Excel (.xlsx, .xls), OpenDocument (.ods), directories, ZIP archives,
or wildcard file patterns into schema-agnostic, individual SQL seed files, plus a
SCHEMA.md describing every table.
Usage:
python3 dataconvert.py --input "data/*" "*.xlsx" --out-dir seed/
python3 dataconvert.py --input path/to/folder/ --out-dir seed/
python3 dataconvert.py --input path/to/folder/ --out-dir sample/ --max-rows 20
--max-rows is for understanding the data rather than loading it: every table
still gets its file and its SCHEMA.md entry, with only the first N rows, and a
note of how many rows there are and how big the full file would be.
The modules beside this file are imported by name, so the folder works wherever
it is copied: run this script from anywhere, no install step.
"""
import argparse
from pathlib import Path
from output import write_tables
from readers import expand_inputs, iter_sources
from schema import SchemaReport
def positive_int(value):
n = int(value)
if n < 1:
raise argparse.ArgumentTypeError("must be 1 or more")
return n
def main():
parser = argparse.ArgumentParser(description="Convert data sources into schema-agnostic SQL seed files.")
parser.add_argument("--input", nargs="+", required=True, help="Input file(s), directory, wildcard pattern(s), or ZIP archive(s)")
parser.add_argument("--out-dir", default="seed", help="Directory where individual .sql files will be written")
parser.add_argument("--max-rows", type=positive_int, default=None,
help="Write at most N rows per table; the header and SCHEMA.md note the full row count and size")
parser.add_argument("--no-schema", action="store_true", help="Do not write SCHEMA.md")
args = parser.parse_args()
out_dir = Path(args.out_dir)
out_dir.mkdir(parents=True, exist_ok=True)
report = None if args.no_schema else SchemaReport()
for in_path in expand_inputs(args.input):
for source_name, dfs in iter_sources(in_path):
write_tables(dfs, out_dir, source_name, args.max_rows, report)
if report is not None and report.entries:
print(f"[dataconvert] Generated: {report.write(out_dir, args.max_rows)}")
if __name__ == "__main__":
main()

View File

@@ -0,0 +1,39 @@
"""
Output: one .sql file per table or sheet, named after it.
"""
from pathlib import Path
from sqlgen import render_table, sanitize_identifier
def table_filename(raw_name: str, source_name: str, sheet_count: int) -> str:
"""Sheets of a multi-sheet workbook are prefixed with the workbook, unless already namespaced."""
clean = sanitize_identifier(raw_name)
if sheet_count > 1 and not clean.startswith("drm_") and not clean.startswith("study_"):
return f"{sanitize_identifier(source_name)}_{clean}.sql"
return f"{clean}.sql"
def write_tables(dfs: dict, out_dir: Path, source_name: str, max_rows=None, report=None):
"""Write individual .sql files per table/sheet into the output directory."""
for raw_name, df in dfs.items():
if df.empty:
continue
table = sanitize_identifier(raw_name)
filename = table_filename(raw_name, source_name, len(dfs))
sql, total, full_bytes, exact = render_table(df, table, max_rows)
out_file = out_dir / filename
# Several sources can feed the same table; they accumulate in one file.
mode = "a" if out_file.exists() else "w"
with open(out_file, mode, encoding="utf-8") as f:
f.write(sql)
shown = total if exact else max_rows
if report is not None:
report.add(source_name, table, filename, df, total, full_bytes, exact, shown)
suffix = "" if exact else f" ({shown} of {total} rows)"
print(f"[dataconvert] Generated: {out_file}{suffix}")

View File

@@ -1,7 +1,7 @@
[project] [project]
name = "seed-converter" name = "dataconvert"
version = "0.1.0" version = "0.1.0"
description = "SQL Seed Converter Tool" description = "Spreadsheets and CSV into SQL seed files and a schema summary"
readme = "README.md" readme = "README.md"
requires-python = ">=3.11" requires-python = ">=3.11"
dependencies = [ dependencies = [
@@ -11,4 +11,4 @@ dependencies = [
] ]
[tool.uv] [tool.uv]
package = false package = false

View File

@@ -0,0 +1,85 @@
"""
Readers: files, directories, ZIP archives and wildcards into DataFrames.
Each input becomes zero or more (source_name, {entity_name: DataFrame}) pairs,
one per spreadsheet or CSV. Nothing here knows about SQL or output files.
"""
import glob
import os
import tempfile
import zipfile
from pathlib import Path
import pandas as pd
SUPPORTED = {".csv", ".xlsx", ".xls", ".ods"}
# DRM catalogue sheets put the real column names on row 1 and start the data
# on row 5; the rows in between are descriptions. Recognised by the first cell
# of the header row.
DRM_HEADER_MARKERS = {
"CONTROLLED_TERMINOLOGY", "DATA_ELEMENT", "DATA_DOMAIN", "DATA_STATE", "DATA_CATEGORY",
}
def expand_inputs(inputs):
"""Wildcard patterns become the paths they match; everything else passes through."""
paths = []
for in_str in inputs:
if any(c in in_str for c in ["*", "?", "["]):
matched = glob.glob(in_str, recursive=True)
if not matched:
print(f"[Warning] No files matched wildcard pattern: '{in_str}'")
paths.extend(Path(m) for m in sorted(matched))
else:
paths.append(Path(in_str))
return paths
def load_dataframes_from_file(file_path: Path) -> dict:
"""Load a file (.csv, .xlsx, .xls, .ods) into {entity_name: DataFrame}."""
ext = file_path.suffix.lower()
dfs = {}
try:
if ext == ".csv":
dfs[file_path.stem] = pd.read_csv(file_path)
elif ext in [".xlsx", ".xls"]:
xls = pd.ExcelFile(file_path)
for sheet in xls.sheet_names:
df_raw = pd.read_excel(xls, sheet_name=sheet, header=None)
if len(df_raw) > 1 and str(df_raw.iloc[1, 0]).strip() in DRM_HEADER_MARKERS:
cols = [str(c).strip() for c in df_raw.iloc[1]]
data_df = df_raw.iloc[5:].copy()
data_df.columns = cols
dfs[sheet] = data_df
else:
dfs[sheet] = pd.read_excel(xls, sheet_name=sheet)
elif ext == ".ods":
xls = pd.ExcelFile(file_path, engine="odf")
for sheet in xls.sheet_names:
dfs[sheet] = pd.read_excel(xls, sheet_name=sheet)
except Exception as e:
print(f"[Warning] Could not read '{file_path.name}': {e}")
return dfs
def iter_sources(path: Path):
"""Yield (source_name, dfs) for a file, a directory (recursively) or a ZIP archive."""
if path.is_file() and path.suffix.lower() == ".zip":
with tempfile.TemporaryDirectory() as tmp_dir:
with zipfile.ZipFile(path, "r") as zip_ref:
zip_ref.extractall(tmp_dir)
yield from iter_sources(Path(tmp_dir))
elif path.is_dir():
for root, _, files in os.walk(path):
for f in sorted(files):
f_path = Path(root) / f
if f_path.suffix.lower() in SUPPORTED:
yield f_path.stem, load_dataframes_from_file(f_path)
elif path.is_file() and path.suffix.lower() in SUPPORTED:
yield path.stem, load_dataframes_from_file(path)

View File

@@ -0,0 +1,122 @@
"""
SCHEMA.md: what each table looks like, without its rows.
Written for reading, by a person or a web LLM that has to understand the data
before anything else: columns, an inferred type, how many are empty, one
example value, and how big the table really is. Types are inferred from what
pandas read, so they are a starting point, not a DDL.
"""
from pathlib import Path
import pandas as pd
from sqlgen import human_bytes, sanitize_identifier
EXAMPLE_MAX = 40
def infer_type(series: pd.Series) -> str:
values = series.dropna()
if values.empty:
return "unknown (all null)"
if pd.api.types.is_bool_dtype(values):
return "boolean"
if pd.api.types.is_integer_dtype(values):
return "integer"
if pd.api.types.is_float_dtype(values):
return "integer" if (values % 1 == 0).all() else "numeric"
if pd.api.types.is_datetime64_any_dtype(values):
return "timestamp"
# Object columns from spreadsheets mix types; name the one that is there.
kinds = {type(v).__name__ for v in values}
if kinds <= {"int", "bool"}:
return "integer"
if kinds <= {"int", "float"}:
return "numeric"
if kinds <= {"datetime", "Timestamp"}:
return "timestamp"
longest = values.astype(str).str.len().max()
return f"text (max {longest})" if kinds == {"str"} else f"mixed ({', '.join(sorted(kinds))})"
def example(series: pd.Series) -> str:
values = series.dropna()
if values.empty:
return ""
text = str(values.iloc[0]).replace("\n", " ").replace("|", "\\|")
return text if len(text) <= EXAMPLE_MAX else text[: EXAMPLE_MAX - 1] + ""
class SchemaReport:
"""Collects one entry per written table, then writes them as one document."""
def __init__(self):
self.entries = []
def add(self, source, table, filename, df, total_rows, full_bytes, exact, shown_rows):
self.entries.append(dict(
source=source, table=table, filename=filename, df=df,
total_rows=total_rows, full_bytes=full_bytes, exact=exact, shown_rows=shown_rows,
))
def tables(self):
"""
One entry per output file. Several sources can feed the same table, and
the file holds all of them, so the report does too: rows and sizes are
summed, and the columns come from the first source.
"""
merged = {}
for e in self.entries:
m = merged.get(e["filename"])
if m is None:
merged[e["filename"]] = dict(e, sources=[e["source"]])
continue
m["sources"].append(e["source"])
m["total_rows"] += e["total_rows"]
m["full_bytes"] += e["full_bytes"]
m["shown_rows"] += e["shown_rows"]
m["exact"] = m["exact"] and e["exact"]
return list(merged.values())
def write(self, out_dir: Path, max_rows):
tables = self.tables()
lines = ["# Data schema", ""]
total_rows = sum(t["total_rows"] for t in tables)
total_bytes = sum(t["full_bytes"] for t in tables)
lines.append(
f"{len(tables)} tables · {total_rows} rows · full seed files "
f"{'~' if any(not t['exact'] for t in tables) else ''}{human_bytes(total_bytes)}"
)
lines.append("")
if max_rows is not None:
lines += [
f"The .sql files beside this one hold at most {max_rows} rows per table from each",
"source: they are samples for understanding the data, not seeds to load. Sizes",
"marked ~ are estimated from the rows that were written.",
"",
]
lines += ["| table | file | rows | full size | in the file |", "|---|---|---:|---:|---|"]
for t in tables:
size = ("" if t["exact"] else "~") + human_bytes(t["full_bytes"])
kept = "all" if t["exact"] else f"{t['shown_rows']} rows"
lines.append(f"| `{t['table']}` | `{t['filename']}` | {t['total_rows']} | {size} | {kept} |")
lines.append("")
for t in tables:
df = t["df"]
sources = ", ".join(f"`{src}`" for src in t["sources"])
lines += [f"## {t['table']}", "", f"From {sources} · {t['total_rows']} rows · {len(df.columns)} columns", ""]
lines += ["| column | source header | type | nulls | example |", "|---|---|---|---:|---|"]
for col in df.columns:
series = df[col]
header = str(col).replace("|", "\\|")
lines.append(
f"| `{sanitize_identifier(col)}` | {header} | {infer_type(series)} | "
f"{int(series.isna().sum())} | {example(series)} |"
)
lines.append("")
path = out_dir / "SCHEMA.md"
path.write_text("\n".join(lines), encoding="utf-8")
return path

View File

@@ -1,164 +0,0 @@
#!/usr/bin/env python3
"""
Modular Seed Converter Tool
Converts CSV, Excel (.xlsx, .xls), OpenDocument (.ods), directories, ZIP archives,
or wildcard file patterns into schema-agnostic, individual SQL seed files.
Usage:
python3 seed_converter.py --input "data/*" "*.xlsx" --out-dir seed/
python3 seed_converter.py --input path/to/folder/ --out-dir seed/
"""
import os
import re
import glob
import argparse
import zipfile
import tempfile
import pandas as pd
from pathlib import Path
def sanitize_identifier(identifier: str) -> str:
"""Sanitize names for SQL tables, columns, and filenames."""
clean = re.sub(r"[^\w]", "_", str(identifier).strip().lower())
clean = re.sub(r"_+", "_", clean)
return clean.strip("_")
def dataframe_to_sql(df: pd.DataFrame, table_name: str) -> str:
"""Generate SQL INSERT statements for a given DataFrame without schema qualification."""
if df.empty:
return ""
table_ref = f'"{table_name}"'
cols = ", ".join([f'"{sanitize_identifier(c)}"' for c in df.columns])
lines = [
f"-- Generated seed data for table: {table_ref}\n",
"BEGIN;\n\n"
]
for _, row in df.iterrows():
vals = []
for v in row:
if pd.isna(v):
vals.append("NULL")
elif isinstance(v, (bool, int)):
vals.append(str(v))
elif isinstance(v, float):
vals.append(str(int(v)) if v.is_integer() else str(v))
else:
escaped = str(v).replace("'", "''")
vals.append(f"'{escaped}'")
val_str = ", ".join(vals)
lines.append(f"INSERT INTO {table_ref} ({cols}) VALUES ({val_str}) ON CONFLICT DO NOTHING;\n")
lines.append("\nCOMMIT;\n")
return "".join(lines)
def load_dataframes_from_file(file_path: Path) -> dict:
"""Load a file (.csv, .xlsx, .ods) into a dictionary of {entity_name: DataFrame}."""
ext = file_path.suffix.lower()
dfs = {}
try:
if ext == ".csv":
dfs[file_path.stem] = pd.read_csv(file_path)
elif ext in [".xlsx", ".xls"]:
xls = pd.ExcelFile(file_path)
for sheet in xls.sheet_names:
df_raw = pd.read_excel(xls, sheet_name=sheet, header=None)
if len(df_raw) > 1 and str(df_raw.iloc[1, 0]).strip() in [
"CONTROLLED_TERMINOLOGY", "DATA_ELEMENT", "DATA_DOMAIN", "DATA_STATE", "DATA_CATEGORY"
]:
cols = [str(c).strip() for c in df_raw.iloc[1]]
data_df = df_raw.iloc[5:].copy()
data_df.columns = cols
dfs[sheet] = data_df
else:
dfs[sheet] = pd.read_excel(xls, sheet_name=sheet)
elif ext == ".ods":
xls = pd.ExcelFile(file_path, engine="odf")
for sheet in xls.sheet_names:
dfs[sheet] = pd.read_excel(xls, sheet_name=sheet)
except Exception as e:
print(f"[Warning] Could not read '{file_path.name}': {e}")
return dfs
def process_and_write(dfs: dict, out_dir: Path, source_name: str):
"""Write individual .sql files per table/sheet into the output directory."""
for raw_name, df in dfs.items():
if df.empty:
continue
clean_table_name = sanitize_identifier(raw_name)
if len(dfs) > 1 and not clean_table_name.startswith("drm_") and not clean_table_name.startswith("study_"):
filename = f"{sanitize_identifier(source_name)}_{clean_table_name}.sql"
else:
filename = f"{clean_table_name}.sql"
sql_content = dataframe_to_sql(df, clean_table_name)
out_file = out_dir / filename
mode = "a" if out_file.exists() else "w"
with open(out_file, mode, encoding="utf-8") as f:
f.write(sql_content)
print(f"[seed_converter] Generated: {out_file}")
def process_path(path: Path, out_dir: Path):
"""Process files, directories, or zip archives recursively."""
supported = {".csv", ".xlsx", ".xls", ".ods"}
if path.is_file() and path.suffix.lower() == ".zip":
with tempfile.TemporaryDirectory() as tmp_dir:
with zipfile.ZipFile(path, "r") as zip_ref:
zip_ref.extractall(tmp_dir)
process_path(Path(tmp_dir), out_dir)
elif path.is_dir():
for root, _, files in os.walk(path):
for f in sorted(files):
f_path = Path(root) / f
if f_path.suffix.lower() in supported:
dfs = load_dataframes_from_file(f_path)
process_and_write(dfs, out_dir, f_path.stem)
elif path.is_file() and path.suffix.lower() in supported:
dfs = load_dataframes_from_file(path)
process_and_write(dfs, out_dir, path.stem)
def main():
parser = argparse.ArgumentParser(description="Convert data sources into schema-agnostic SQL seed files.")
parser.add_argument("--input", nargs="+", required=True, help="Input file(s), directory, wildcard pattern(s), or ZIP archive(s)")
parser.add_argument("--out-dir", default="seed", help="Directory where individual .sql files will be written")
args = parser.parse_args()
out_dir = Path(args.out_dir)
out_dir.mkdir(parents=True, exist_ok=True)
expanded_paths = []
for in_str in args.input:
if any(c in in_str for c in ["*", "?", "["]):
matched = glob.glob(in_str, recursive=True)
if not matched:
print(f"[Warning] No files matched wildcard pattern: '{in_str}'")
for m in sorted(matched):
expanded_paths.append(Path(m))
else:
expanded_paths.append(Path(in_str))
for in_path in expanded_paths:
process_path(in_path, out_dir)
if __name__ == "__main__":
main()

View File

@@ -0,0 +1,81 @@
"""
SQL rendering: DataFrames into schema-agnostic INSERT statements.
With a row cap, only the first rows are rendered, and the size the full output
would have had is estimated from them, so a sample still says how big the real
thing is.
"""
import re
import pandas as pd
def sanitize_identifier(identifier: str) -> str:
"""Sanitize names for SQL tables, columns, and filenames."""
clean = re.sub(r"[^\w]", "_", str(identifier).strip().lower())
clean = re.sub(r"_+", "_", clean)
return clean.strip("_")
def sql_value(v) -> str:
if pd.isna(v):
return "NULL"
if isinstance(v, (bool, int)):
return str(v)
if isinstance(v, float):
return str(int(v)) if v.is_integer() else str(v)
escaped = str(v).replace("'", "''")
return f"'{escaped}'"
def render_table(df: pd.DataFrame, table_name: str, max_rows=None):
"""
Return (sql_text, total_rows, full_bytes, exact).
full_bytes is what the file would weigh with every row: measured when every
row was rendered, extrapolated from the average rendered row otherwise.
"""
total = len(df)
if total == 0:
return "", 0, 0, True
table_ref = f'"{table_name}"'
cols = ", ".join(f'"{sanitize_identifier(c)}"' for c in df.columns)
shown = df if max_rows is None else df.head(max_rows)
rows = []
# iterrows, not itertuples: it hands values over the way the original tool
# did, and seed files people already load depend on exactly that quoting.
for _, row in shown.iterrows():
vals = ", ".join(sql_value(v) for v in row)
rows.append(f"INSERT INTO {table_ref} ({cols}) VALUES ({vals}) ON CONFLICT DO NOTHING;\n")
head = f"-- Generated seed data for table: {table_ref}\n"
begin, commit = "BEGIN;\n\n", "\nCOMMIT;\n"
rows_bytes = sum(len(r.encode("utf-8")) for r in rows)
fixed = len((head + begin + commit).encode("utf-8"))
exact = len(rows) == total
if exact:
full_bytes = fixed + rows_bytes
else:
full_bytes = fixed + round(rows_bytes / len(rows) * total)
note = ""
if not exact:
note = (
f"-- SAMPLE: first {len(rows)} of {total} rows. The full file would be "
f"~{human_bytes(full_bytes)}; run without --max-rows for all of it.\n"
)
return head + note + begin + "".join(rows) + commit, total, full_bytes, exact
def human_bytes(n: int) -> str:
size = float(n)
for unit in ("B", "K", "M", "G"):
if size < 1000 or unit == "G":
return f"{size:.0f}{unit}" if unit == "B" else f"{size:.1f}{unit}"
size /= 1000
return f"{n}B"

View File

@@ -285,7 +285,7 @@ wheels = [
] ]
[[package]] [[package]]
name = "seed-converter" name = "dataconvert"
version = "0.1.0" version = "0.1.0"
source = { virtual = "." } source = { virtual = "." }
dependencies = [ dependencies = [