diff --git a/soleprint/station/tools/dataconvert/README.md b/soleprint/station/tools/dataconvert/README.md index 8529e62..41cf30c 100644 --- a/soleprint/station/tools/dataconvert/README.md +++ b/soleprint/station/tools/dataconvert/README.md @@ -7,6 +7,7 @@ plus a `SCHEMA.md` describing every table. 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 +uv run dataconvert.py --input data/ --out-dir seed/ --sql-format batch --max-file-size 100M # a full dump uv run dataconvert.py --input export.xlsx --header-row 2 --data-row 6 uv run dataconvert.py --input data/ --config their-exports.json ``` @@ -29,8 +30,12 @@ produced the files. Start from `dataconvert-example.json`. ```json { - "out_dir": "sample", - "max_rows": 20, + "out_dir": "seed", + "all_text": true, + "keep_folders": true, + "exclude": ["*.xlsx.ods"], + "sql_format": "batch", + "max_file_size": "100M", "schema": true, "header_row": 1, "layouts": [ @@ -44,8 +49,11 @@ produced the files. Start from `dataconvert-example.json`. Every key is optional. -- **out_dir, max_rows, schema** are the run settings: the same as `--out-dir`, - `--max-rows` and `--no-schema`, and a flag on the command line wins over the file. +- **out_dir, max_rows, schema, sql_format, batch_rows, max_file_size, all_text, + keep_folders, exclude** are the run settings: the same as `--out-dir`, `--max-rows`, + `--no-schema`, `--sql-format`, `--batch-rows`, `--max-file-size`, `--all-text`, + `--keep-folders` and `--exclude`, and a flag on the command line wins over the file. + `--exclude` patterns are added to the file's. There is no default output directory: without `--out-dir` or an `out_dir` in the config, the run stops before writing anything. `out_dir` is relative to the folder the config is in, so `"sample"` beside the data @@ -67,6 +75,65 @@ such as `max_row` is not silently ignored. For a one-off, `--header-row 2 --data-row 6` applies one layout to every file in the run and skips the config's layouts. +## Values as they are: all_text + +By default pandas guesses a type per column, and for real exports that guess does +damage: a site code `0012` becomes `12`, `NA` becomes NULL, and a column that is +numbers for the first 200,000 rows and text after that is typed differently in +different rows. `all_text` reads every cell as the string in the file. Only an empty +cell becomes NULL; `NA`, `N/A` and `null` stay text. Converting types is then the +loader's job, where the real schema is decided. + +To help decide it, `SCHEMA.md` describes what each text column looks like: + +| type column says | meaning | +|---|---| +| `text, integer-like (max 6)` | every value is a whole number | +| `text, code with leading zeros (max 4)` | whole numbers, some with leading zeros: keep as text | +| `text, decimal-like` / `timestamp-like (YYYY-MM-DD hh:mm)` / `date-like (DDMONYYYY)` / `boolean-like` | every value has that shape | +| `text, mostly date-like (DDMONYYYY): 300 others` | 95% or more fit; the others are what needs a rule (partial dates, `UNK`) | +| `text, 67% integer-like` | at least half fit | +| `text (max 40)` | no common shape | + +Shapes are judged on up to 200,000 values spread through the column. + +## Folders and exclusions + +Output is flat by default: a table's file is named after the file or sheet, and two +source folders with the same table name (two studies' `drm_lb.csv`) write into the +same file. `keep_folders` mirrors the source folders under `out_dir` instead, so each +keeps its own `.sql`; `SCHEMA.md` stays one document at the top, with the folder in +each table's heading. + +`exclude` skips files by glob. A pattern with no `/` matches the file name at any +depth (`"*.xlsx.ods"`); one with a `/` matches the path under the input folder +(`"Old/*"`). Each skipped file is named in the run's output. + +## Full dumps: format and splitting + +One `INSERT` per row repeats the table and column list on every line, so the SQL can +be many times the size of the spreadsheets, which are compressed on disk to begin +with. `sql_format` picks how rows are written for PostgreSQL: + +| format | shape | load with | re-runnable | size | +|---|---|---|---|---| +| `insert` (default) | one `INSERT ... ON CONFLICT DO NOTHING` per row | any client | yes | largest | +| `batch` | one `INSERT` per `batch_rows` rows (500), same `ON CONFLICT` | any client | yes | about a quarter | +| `copy` | `COPY ... FROM stdin`, tab-separated | `psql` only | no: a second load fails on duplicate keys | smallest | + +`max_file_size` (`"100M"`, decimal k/M/G) splits a table's output into `table.001.sql`, +`table.002.sql`, ... once it passes that size. Splits fall between rows (or batches), +never inside one, and every part is its own transaction, so the parts load one at a time +in name order: + +```bash +for f in seed/*.sql; do psql -v ON_ERROR_STOP=1 -f "$f" || break; done +``` + +A table under the limit keeps its plain `table.sql` name. Rows are written to disk as +they are rendered, so a table larger than memory converts; reading the source still +needs it in memory. Samples (`max_rows`) are never split. + ## Sampling for a web LLM Full seed files get large fast, and a model only needs to see the shape of the data. @@ -90,8 +157,8 @@ what pandas read: a starting point, not DDL. `--no-schema` skips the file. | `config.py` | `dataconvert.json`: layouts, sheet naming and run settings | | `readers.py` | files, directories, ZIPs and globs into DataFrames | | `progress.py` | progress lines: which file is being read, and how far a long table has got | -| `sqlgen.py` | DataFrames into INSERT statements, with the row cap | -| `output.py` | file naming and writing | +| `sqlgen.py` | rows as `insert`, `batch` or `copy` statements, and size estimates | +| `output.py` | file naming, streaming rows to disk, and splitting into parts | | `schema.py` | `SCHEMA.md` | The modules import each other by name, so the folder works wherever it is copied: diff --git a/soleprint/station/tools/dataconvert/config.py b/soleprint/station/tools/dataconvert/config.py index 97a7386..536a0cf 100644 --- a/soleprint/station/tools/dataconvert/config.py +++ b/soleprint/station/tools/dataconvert/config.py @@ -12,8 +12,10 @@ Start from dataconvert-example.json. With neither, every file is read with its header on row 1. { - "out_dir": "sample", - "max_rows": 20, + "out_dir": "seed", + "sql_format": "batch", + "batch_rows": 500, + "max_file_size": "100M", "schema": true, "header_row": 1, "layouts": [ @@ -32,15 +34,21 @@ applies to a sheet (or CSV) when the cell at match.row/match.column, trimmed, is one of match.in. The first layout that matches wins; a sheet none matches uses the top-level header_row/data_row, which default to 1 and 2. -The run settings (out_dir, max_rows, schema) are the command line's flags with -the same names, and a flag given on the command line wins over the file. out_dir -is relative to the folder the config file is in. Keys starting with _ are +The run settings (out_dir, max_rows, schema, sql_format, batch_rows, +max_file_size, all_text, keep_folders, exclude) are the command line's flags +with the same names, and a flag given on the command line wins over the file; +--exclude patterns are added to the file's. out_dir is relative to the folder +the config file is in; max_file_size takes a number of bytes or a suffixed size +such as "100M" (decimal: k, M, G). Keys starting with _ are comments; any other unknown key is an error, so a typo is not silently ignored. """ import json +import re from pathlib import Path +from sqlgen import FORMATS + class ConfigError(Exception): pass @@ -86,7 +94,9 @@ class Rule: class Config: def __init__(self, rules=(), bare_sheet_prefixes=(), source=None, forced=None, - fallback=None, max_rows=None, schema=None, out_dir=None): + fallback=None, max_rows=None, schema=None, out_dir=None, + sql_format=None, batch_rows=None, max_file_size=None, + all_text=None, keep_folders=None, exclude=()): self.rules = list(rules) self.bare_sheet_prefixes = tuple(bare_sheet_prefixes) self.source = source @@ -99,6 +109,12 @@ class Config: self.max_rows = max_rows self.schema = schema self.out_dir = out_dir + self.sql_format = sql_format + self.batch_rows = batch_rows + self.max_file_size = max_file_size + self.all_text = all_text + self.keep_folders = keep_folders + self.exclude = tuple(exclude) @property def probe_rows(self): @@ -121,6 +137,18 @@ class Config: said.append(f"max_rows {self.max_rows}") if self.schema is not None: said.append(f"schema {'on' if self.schema else 'off'}") + if self.sql_format is not None: + said.append(f"sql_format {self.sql_format}") + if self.batch_rows is not None: + said.append(f"batch_rows {self.batch_rows}") + if self.max_file_size is not None: + said.append(f"max_file_size {self.max_file_size} bytes") + if self.all_text: + said.append("all_text") + if self.keep_folders: + said.append("keep_folders") + if self.exclude: + said.append(f"exclude {', '.join(self.exclude)}") if self.fallback is not None: said.append(f"header row {self.fallback.header_row}, data from row {self.fallback.data_row}") if self.rules: @@ -129,7 +157,24 @@ class Config: FILENAME = "dataconvert.json" -KNOWN_KEYS = {"layouts", "bare_sheet_prefixes", "max_rows", "schema", "out_dir", "header_row", "data_row"} +KNOWN_KEYS = {"layouts", "bare_sheet_prefixes", "max_rows", "schema", "out_dir", "header_row", "data_row", + "sql_format", "batch_rows", "max_file_size", "all_text", "keep_folders", "exclude"} + + +def parse_size(value): + """Bytes from 100000000, "100000000" or "100M". Decimal, as distill counts: 1k is 1000.""" + if isinstance(value, bool): + raise ValueError(value) + if isinstance(value, int): + n = value + else: + m = re.fullmatch(r"\s*(\d+)\s*([kKmMgG]?)[bB]?\s*", str(value)) + if not m: + raise ValueError(value) + n = int(m[1]) * {"": 1, "k": 1000, "m": 1000 ** 2, "g": 1000 ** 3}[m[2].lower()] + if n < 1: + raise ValueError(value) + return n def source_config_path(in_path: Path): @@ -175,6 +220,28 @@ def load(path=None, forced=None): raise ConfigError(f"{path}: out_dir must be a path") out_dir = path.parent / Path(out_dir).expanduser() + sql_format = raw.get("sql_format") + if sql_format is not None and sql_format not in FORMATS: + raise ConfigError(f"{path}: sql_format must be one of {', '.join(FORMATS)}") + batch_rows = raw.get("batch_rows") + if batch_rows is not None and (isinstance(batch_rows, bool) or not isinstance(batch_rows, int) or batch_rows < 1): + raise ConfigError(f"{path}: batch_rows must be a whole number, 1 or more") + max_file_size = raw.get("max_file_size") + if max_file_size is not None: + try: + max_file_size = parse_size(max_file_size) + except ValueError: + raise ConfigError(f'{path}: max_file_size must be a size such as 100000000 or "100M"') + + flags = {} + for key in ("all_text", "keep_folders"): + flags[key] = raw.get(key) + if flags[key] is not None and not isinstance(flags[key], bool): + raise ConfigError(f"{path}: {key} must be true or false") + exclude = raw.get("exclude", []) + if not isinstance(exclude, list) or not all(isinstance(p, str) and p for p in exclude): + raise ConfigError(f'{path}: exclude must be a list of patterns such as "*.xlsx.ods"') + fallback = None if "header_row" in raw or "data_row" in raw: try: @@ -183,4 +250,6 @@ def load(path=None, forced=None): raise ConfigError(f"{path}: header_row and data_row must be row numbers") return Config(rules, [str(p) for p in prefixes], source=path, forced=forced, - fallback=fallback, max_rows=max_rows, schema=schema, out_dir=out_dir) + fallback=fallback, max_rows=max_rows, schema=schema, out_dir=out_dir, + sql_format=sql_format, batch_rows=batch_rows, max_file_size=max_file_size, + all_text=flags["all_text"], keep_folders=flags["keep_folders"], exclude=exclude) diff --git a/soleprint/station/tools/dataconvert/dataconvert-example.json b/soleprint/station/tools/dataconvert/dataconvert-example.json index b8e30d5..8fdfcfd 100644 --- a/soleprint/station/tools/dataconvert/dataconvert-example.json +++ b/soleprint/station/tools/dataconvert/dataconvert-example.json @@ -1,9 +1,17 @@ { "_comment": "Template for one set of spreadsheets. Copy it to dataconvert.json in the folder that holds them, where it is picked up automatically, or pass any file with --config FILE. Every key is optional; keys starting with _ are comments, any other unknown key is an error. Rows and columns count from 1, as the spreadsheet shows them.", - "_run": "The same settings as the command-line flags, and a flag given there wins. out_dir is relative to this file's folder. max_rows samples each table (null for all rows). schema false skips SCHEMA.md.", - "out_dir": "sample", - "max_rows": 20, + "_run": "The same settings as the command-line flags, and a flag given there wins. out_dir is relative to this file's folder. max_rows samples each table (leave it out for all rows). schema false skips SCHEMA.md. sql_format: insert (a statement per row, default), batch (one INSERT per batch_rows rows), or copy (COPY FROM stdin: smallest, psql only, load once). max_file_size splits a table into numbered parts past that size, e.g. \"100M\".", + "out_dir": "seed", + + "_values": "all_text reads every cell as the string in the file (only empty cells become NULL), so codes keep leading zeros and NA stays NA; SCHEMA.md then says what each column looks like. keep_folders mirrors the source folders under out_dir, so same-named tables from different folders do not share a file. exclude skips files by glob: no / matches the file name at any depth, with a / the path under the input folder.", + "all_text": true, + "keep_folders": true, + "exclude": ["*.xlsx.ods"], + + "sql_format": "batch", + "batch_rows": 500, + "max_file_size": "100M", "schema": true, "_rows": "Where the column names and the data are, for sheets no layout below matches. Defaults: 1 and the row after it.", diff --git a/soleprint/station/tools/dataconvert/dataconvert.py b/soleprint/station/tools/dataconvert/dataconvert.py index d2f1dae..c58dc82 100644 --- a/soleprint/station/tools/dataconvert/dataconvert.py +++ b/soleprint/station/tools/dataconvert/dataconvert.py @@ -28,12 +28,20 @@ import argparse from pathlib import Path from output import write_tables +from sqlgen import DEFAULT_BATCH_ROWS, FORMATS from progress import say import config as cfg from readers import expand_inputs, iter_sources from schema import SchemaReport +def size(value): + try: + return cfg.parse_size(value) + except ValueError: + raise argparse.ArgumentTypeError('a size such as 100000000 or "100M"') + + def positive_int(value): n = int(value) if n < 1: @@ -49,6 +57,22 @@ def main(): 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 (default: the config's max_rows, else all)") parser.add_argument("--no-schema", action="store_true", help="Do not write SCHEMA.md, whatever the config says") + parser.add_argument("--sql-format", choices=FORMATS, default=None, + help="insert: a statement per row (default). batch: one INSERT per --batch-rows rows. " + "copy: COPY FROM stdin, smallest, psql only, load once") + parser.add_argument("--batch-rows", type=positive_int, default=None, + help=f"Rows per INSERT with --sql-format batch (default {DEFAULT_BATCH_ROWS})") + parser.add_argument("--max-file-size", type=size, default=None, + help='Split a table into numbered parts past this size, e.g. "100M" (default: never split; samples are never split)') + parser.add_argument("--all-text", action="store_true", default=None, + help="Read every column as text, exactly as in the file; only empty cells become NULL. " + "Types are then the loader's business; SCHEMA.md says what each column looks like") + parser.add_argument("--keep-folders", action="store_true", default=None, + help="Mirror the source folders under the output directory, instead of one flat folder " + "where same-named tables from different folders share a file") + parser.add_argument("--exclude", action="append", default=[], metavar="GLOB", + help='Skip matching files, e.g. "*.xlsx.ods"; a pattern with no / matches the file name ' + "at any depth. Repeatable, and added to the config's") parser.add_argument("--header-row", type=positive_int, default=1, help="Spreadsheet row holding the column names, for every file; overrides the config's layouts (default 1)") parser.add_argument("--data-row", type=positive_int, default=None, @@ -107,6 +131,12 @@ def main(): out_dir = Path(args.out_dir) if args.out_dir else config.out_dir max_rows = args.max_rows if args.max_rows is not None else config.max_rows schema = not args.no_schema and config.schema is not False + fmt = args.sql_format or config.sql_format or "insert" + batch_rows = args.batch_rows or config.batch_rows or DEFAULT_BATCH_ROWS + max_bytes = args.max_file_size or config.max_file_size + all_text = bool(args.all_text or config.all_text) + keep_folders = bool(args.keep_folders or config.keep_folders) + exclude = tuple(config.exclude) + tuple(args.exclude) key = out_dir.resolve() if key not in reports: @@ -114,8 +144,9 @@ def main(): _, report, caps = reports[key] caps.add(max_rows) - for source_name, dfs in iter_sources(in_path, config): - write_tables(dfs, out_dir, source_name, max_rows, report if schema else None, config.bare_sheet_prefixes) + for source_name, folder, dfs in iter_sources(in_path, config, all_text, exclude): + write_tables(dfs, out_dir, source_name, max_rows, report if schema else None, config.bare_sheet_prefixes, + fmt, batch_rows, max_bytes, folder if keep_folders else "") for out_dir, report, caps in reports.values(): if report.entries and out_dir.is_dir(): diff --git a/soleprint/station/tools/dataconvert/output.py b/soleprint/station/tools/dataconvert/output.py index 7fd202a..efbec72 100644 --- a/soleprint/station/tools/dataconvert/output.py +++ b/soleprint/station/tools/dataconvert/output.py @@ -1,11 +1,14 @@ """ -Output: one .sql file per table or sheet, named after it. +Output: one .sql file per table or sheet, named after it, split into numbered +parts when a size limit is set and the table outgrows it. """ +import glob +import re from pathlib import Path from progress import say -from sqlgen import render_table, sanitize_identifier +from sqlgen import DEFAULT_BATCH_ROWS, Shape, render_sample, sanitize_identifier, utf8_len def table_filename(raw_name: str, source_name: str, sheet_count: int, bare_prefixes=()) -> str: @@ -20,28 +23,127 @@ def table_filename(raw_name: str, source_name: str, sheet_count: int, bare_prefi return f"{clean}.sql" -def write_tables(dfs: dict, out_dir: Path, source_name: str, max_rows=None, report=None, bare_prefixes=()): - """Write individual .sql files per table/sheet into the output directory.""" +class TableFile: + """ + Where one table's statements go: .sql, or once it passes max_bytes, +
.001.sql,
.002.sql, ... in load order. + + Several sources can feed the same table and they accumulate: writing resumes + at the last part that exists, as the unsplit file always did. + """ + + def __init__(self, out_dir: Path, filename: str, max_bytes=None): + self.out_dir = out_dir + self.stem = filename[:-len(".sql")] + self.max_bytes = max_bytes + self.handle = None + self.touched = [] + parts = self.parts() + self.index = int(parts[-1].name[len(self.stem) + 1:-len(".sql")]) if parts else 0 + self.size = self.path.stat().st_size if self.path.exists() else 0 + + @property + def path(self) -> Path: + name = f"{self.stem}.sql" if self.index == 0 else f"{self.stem}.{self.index:03d}.sql" + return self.out_dir / name + + def parts(self): + pattern = re.compile(re.escape(self.stem) + r"\.\d{3}\.sql$") + if not self.out_dir.is_dir(): + return [] + return sorted(p for p in self.out_dir.glob(glob.escape(self.stem) + ".*.sql") if pattern.match(p.name)) + + def write(self, text: str): + if self.handle is None: + # Created on the first write, so a run that finds nothing to + # convert leaves no empty directory behind. + self.out_dir.mkdir(parents=True, exist_ok=True) + self.handle = open(self.path, "a", encoding="utf-8") + if self.path not in self.touched: + self.touched.append(self.path) + self.handle.write(text) + self.size += utf8_len(text) + + def full(self, adding: int) -> bool: + return self.max_bytes is not None and self.size > 0 and self.size + adding > self.max_bytes + + def next_part(self): + """Move on to the next numbered part; the unnumbered file becomes part 001.""" + self.close() + if self.index == 0: + base, first = self.path, self.out_dir / f"{self.stem}.001.sql" + base.rename(first) + self.touched = [first if p == base else p for p in self.touched] + self.index = 1 + self.index += 1 + self.size = 0 + + def close(self): + if self.handle is not None: + self.handle.close() + self.handle = None + + +def write_full(shape: Shape, df, table_file: TableFile) -> int: + """Every row, streamed; a new part, as its own transaction, whenever the limit is reached.""" + opening, closing = shape.open(), shape.close() + open_bytes, close_bytes = utf8_len(opening), utf8_len(closing) + if table_file.full(open_bytes + close_bytes): + table_file.next_part() + table_file.write(opening) + written = open_bytes + units_in_part = 0 + for unit in shape.units(df): + size = utf8_len(unit) + # Only between units, and never leaving a part empty: a single unit + # bigger than the limit gets a part of its own instead of looping. + if units_in_part and table_file.full(size + close_bytes): + table_file.write(closing) + table_file.next_part() + table_file.write(opening) + written += close_bytes + open_bytes + units_in_part = 0 + table_file.write(unit) + written += size + units_in_part += 1 + table_file.write(closing) + return written + close_bytes + + +def write_tables(dfs: dict, out_dir: Path, source_name: str, max_rows=None, report=None, bare_prefixes=(), + fmt="insert", batch_rows=DEFAULT_BATCH_ROWS, max_bytes=None, folder=""): + """ + Write individual .sql files per table/sheet into the output directory, or + into its `folder` subdirectory when the source folders are kept. + """ + target = out_dir / folder if folder else out_dir 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), bare_prefixes) - sql, total, full_bytes, exact = render_table(df, table, max_rows) - out_file = out_dir / filename - # Created on the first file, so a run that finds nothing to convert - # leaves no empty directory behind. - out_dir.mkdir(parents=True, exist_ok=True) + shape = Shape(fmt, table, df.columns, batch_rows) + total = len(df) + exact = max_rows is None or max_rows >= total + # A sample is never split: it is small, and it is not for loading. + table_file = TableFile(target, filename, max_bytes if exact else None) - # 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) + try: + if exact: + full_bytes = write_full(shape, df, table_file) + else: + text, full_bytes = render_sample(shape, df, max_rows) + table_file.write(text) + finally: + table_file.close() shown = total if exact else max_rows + parts = len(table_file.parts()) or 1 if report is not None: - report.add(source_name, table, filename, df, total, full_bytes, exact, shown) + report.add(source_name, table, f"{folder}/{filename}" if folder else filename, + df, total, full_bytes, exact, shown, parts) suffix = "" if exact else f" ({shown} of {total} rows)" - say(f"Generated: {out_file}{suffix}") + for path in table_file.touched: + say(f"Generated: {path}{suffix}") diff --git a/soleprint/station/tools/dataconvert/readers.py b/soleprint/station/tools/dataconvert/readers.py index 1cfd565..0a593a4 100644 --- a/soleprint/station/tools/dataconvert/readers.py +++ b/soleprint/station/tools/dataconvert/readers.py @@ -1,16 +1,18 @@ """ 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. +Each input becomes zero or more (source_name, folder, {entity_name: DataFrame}) +triples, one per spreadsheet or CSV, where folder is where the file sits relative +to the input folder. Nothing here knows about SQL or output files. """ +import fnmatch import glob import os import tempfile import time import zipfile -from pathlib import Path +from pathlib import Path, PurePosixPath import pandas as pd @@ -19,6 +21,20 @@ from sqlgen import human_bytes SUPPORTED = {".csv", ".xlsx", ".xls", ".ods"} +# all_text: every cell as the string in the file. Only an empty cell is missing; +# pandas' default list would also turn "NA", "N/A", "null" and "nan" into NULL, +# and in real data "NA" is as often a value as an absence. +TEXT_READ = dict(dtype=str, keep_default_na=False, na_values=[""]) + + +def excluded(rel_path: str, patterns) -> bool: + """ + A pattern with a / is matched against the path under the input folder; one + without, against the file name alone, so "*.xlsx.ods" works at any depth. + """ + name = rel_path.rsplit("/", 1)[-1] + return any(fnmatch.fnmatch(rel_path if "/" in p else name, p) for p in patterns) + def expand_inputs(inputs): """Wildcard patterns become the paths they match; everything else passes through.""" @@ -59,10 +75,11 @@ def read_sheet(read, config): return data, layout -def load_dataframes_from_file(file_path: Path, config) -> dict: +def load_dataframes_from_file(file_path: Path, config, all_text=False) -> dict: """Load a file (.csv, .xlsx, .xls, .ods) into {entity_name: DataFrame}.""" ext = file_path.suffix.lower() dfs = {} + extra = TEXT_READ if all_text else {} # Said before the read, not after: a big workbook can take minutes inside # pandas, and this line is what says which file that is. @@ -74,7 +91,7 @@ def load_dataframes_from_file(file_path: Path, config) -> dict: # chunk, so one column can come out as numbers in one chunk and # text in the next, quoted differently row to row in the SQL. Read # whole, each column gets one type. Costs memory on huge files. - df, layout = read_sheet(lambda **kw: pd.read_csv(file_path, low_memory=False, **kw), config) + df, layout = read_sheet(lambda **kw: pd.read_csv(file_path, low_memory=False, **extra, **kw), config) dfs[file_path.stem] = df say(f" {len(df)} rows{layout_note(layout)} ({seconds(start)})") elif ext in [".xlsx", ".xls", ".ods"]: @@ -82,7 +99,7 @@ def load_dataframes_from_file(file_path: Path, config) -> dict: for sheet in xls.sheet_names: start = time.monotonic() df, layout = read_sheet( - lambda sheet=sheet, **kw: pd.read_excel(xls, sheet_name=sheet, **kw), config) + lambda sheet=sheet, **kw: pd.read_excel(xls, sheet_name=sheet, **extra, **kw), config) dfs[sheet] = df say(f" sheet {sheet}: {len(df)} rows{layout_note(layout)} ({seconds(start)})") except Exception as e: @@ -97,20 +114,37 @@ def layout_note(layout): return f", layout '{layout.name}' (header row {layout.header_row}, data from row {layout.data_row})" -def iter_sources(path: Path, config): - """Yield (source_name, dfs) for a file, a directory (recursively) or a ZIP archive.""" +def iter_sources(path: Path, config, all_text=False, exclude=(), prefix=""): + """ + Yield (source_name, folder, dfs) for a file, a directory (recursively) or a + ZIP archive given as an input. folder is the file's directory under the + input folder (or inside the ZIP), "" at the top. + """ if path.is_file() and path.suffix.lower() == ".zip": + if excluded(prefix + path.name, exclude): + say(f"skipping {prefix + path.name} (excluded)") + return 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), config) + yield from iter_sources(Path(tmp_dir), config, all_text, exclude, prefix) elif path.is_dir(): - for root, _, files in os.walk(path): + for root, dirs, files in os.walk(path): + dirs.sort() 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, config) + if f_path.suffix.lower() not in SUPPORTED: + continue + rel = prefix + f_path.relative_to(path).as_posix() + if excluded(rel, exclude): + say(f"skipping {rel} (excluded)") + continue + folder = str(PurePosixPath(rel).parent) if "/" in rel else "" + yield f_path.stem, folder, load_dataframes_from_file(f_path, config, all_text) elif path.is_file() and path.suffix.lower() in SUPPORTED: - yield path.stem, load_dataframes_from_file(path, config) + if excluded(prefix + path.name, exclude): + say(f"skipping {prefix + path.name} (excluded)") + return + yield path.stem, prefix.rstrip("/"), load_dataframes_from_file(path, config, all_text) diff --git a/soleprint/station/tools/dataconvert/schema.py b/soleprint/station/tools/dataconvert/schema.py index a088381..0d1bd4d 100644 --- a/soleprint/station/tools/dataconvert/schema.py +++ b/soleprint/station/tools/dataconvert/schema.py @@ -5,8 +5,14 @@ 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. + +Text columns, which is every column with all_text, are described by what their +values look like: integer-like, a code with leading zeros, a date in a given +format. Nothing is converted; the point is to have the evidence in one place +when the real schema is decided, in the loader. """ +import re from pathlib import Path import pandas as pd @@ -14,6 +20,52 @@ import pandas as pd from sqlgen import human_bytes, sanitize_identifier EXAMPLE_MAX = 40 +# Text shapes are judged on at most this many values, spread through the column. +SHAPE_SAMPLE = 200_000 +# A shape that fits this share of the values is reported as "mostly". +MOSTLY = 0.95 + +SHAPES = ( + # (name, pattern), most specific first. + ("integer-like", r"[+-]?\d+"), + ("decimal-like", r"[+-]?(?:\d+\.\d*|\.\d+|\d+)(?:[eE][+-]?\d+)?"), + ("timestamp-like (YYYY-MM-DD hh:mm)", r"\d{4}-\d{2}-\d{2}[ T]\d{2}:\d{2}(?::\d{2}(?:\.\d+)?)?(?:Z|[+-]\d{2}:?\d{2})?"), + ("date-like (YYYY-MM-DD)", r"\d{4}-\d{2}-\d{2}"), + ("date-like (DDMONYYYY)", r"\d{1,2}[A-Za-z]{3}\d{4}"), + ("timestamp-like (DDMONYYYY:hh:mm)", r"\d{1,2}[A-Za-z]{3}\d{4}[: ]\d{2}:\d{2}(?::\d{2})?"), + ("date-like (D/M/Y or M/D/Y)", r"\d{1,2}/\d{1,2}/\d{2,4}"), + ("boolean-like", r"(?i:y|n|yes|no|true|false)"), +) + + +def text_shape(values: pd.Series) -> str: + """What a column of strings looks like, for choosing its real type later.""" + text = values.astype(str) + longest = int(text.str.len().max()) + if len(text) > SHAPE_SAMPLE: + text = text.iloc[:: len(text) // SHAPE_SAMPLE] + stripped = text.str.strip() + stripped = stripped[stripped != ""] + if stripped.empty: + return f"text, blank (max {longest})" + best = None + for name, pattern in SHAPES: + fits = stripped.str.fullmatch(pattern) + share = fits.mean() + if name == "integer-like" and share and stripped[fits].str.fullmatch(r"[+-]?0\d+").any(): + name = "code with leading zeros" + if share == 1: + return f"text, {name} (max {longest})" + if share >= MOSTLY: + others = int((~fits).sum()) + return f"text, mostly {name}: {others} other{'s' if others != 1 else ''} (max {longest})" + if best is None or share > best[1]: + best = (name, share) + # Half or more is still worth knowing when the type is being decided: a + # visit number that is "UNS" a third of the time is a decision, not text. + if best is not None and best[1] >= 0.5: + return f"text, {best[1]:.0%} {best[0]} (max {longest})" + return f"text (max {longest})" def infer_type(series: pd.Series) -> str: @@ -36,8 +88,9 @@ def infer_type(series: pd.Series) -> str: 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))})" + if kinds == {"str"}: + return text_shape(values) + return f"mixed ({', '.join(sorted(kinds))})" def example(series: pd.Series) -> str: @@ -54,10 +107,11 @@ class SchemaReport: def __init__(self): self.entries = [] - def add(self, source, table, filename, df, total_rows, full_bytes, exact, shown_rows): + def add(self, source, table, filename, df, total_rows, full_bytes, exact, shown_rows, parts=1): 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, + parts=parts, )) def tables(self): @@ -77,6 +131,9 @@ class SchemaReport: m["full_bytes"] += e["full_bytes"] m["shown_rows"] += e["shown_rows"] m["exact"] = m["exact"] and e["exact"] + # The part count is read off the disk after each write, so the + # latest one already includes everything written before it. + m["parts"] = max(m["parts"], e["parts"]) return list(merged.values()) def write(self, out_dir: Path, max_rows): @@ -100,13 +157,19 @@ class SchemaReport: 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} |") + file = f"`{t['filename']}`" + if t["parts"] > 1: + stem = t["filename"][:-len(".sql")] + file = f"`{stem}.001.sql` … `{stem}.{t['parts']:03d}.sql` ({t['parts']} parts)" + lines.append(f"| `{t['table']}` | {file} | {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", ""] + folder = t["filename"].rsplit("/", 1)[0] if "/" in t["filename"] else "" + heading = f"{t['table']} · {folder}" if folder else t["table"] + lines += [f"## {heading}", "", 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] diff --git a/soleprint/station/tools/dataconvert/sqlgen.py b/soleprint/station/tools/dataconvert/sqlgen.py index a9d8b96..8b96299 100644 --- a/soleprint/station/tools/dataconvert/sqlgen.py +++ b/soleprint/station/tools/dataconvert/sqlgen.py @@ -1,17 +1,35 @@ """ -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. +SQL rendering: DataFrames into schema-agnostic PostgreSQL seed statements. + +Three shapes, the same rows: + + insert one INSERT per row, ON CONFLICT DO NOTHING. Runs in any client and can + be re-run; the column list repeats on every row, so it is the largest. + batch one INSERT per batch_rows rows, still ON CONFLICT DO NOTHING. Same + guarantees as insert, a fraction of the size, and far faster to load. + copy COPY ... FROM stdin, tab-separated. Smallest and fastest, but only psql + runs it, and COPY has no ON CONFLICT: loading it twice fails on the + first duplicate key. + +A block is one transaction: open(), units, close(). Units are the pieces a file +may be split between (a row, or a batch), so every part of a split file is a +complete, loadable script. Getting them onto disk is output.py's business, and +nothing here holds a whole table's text, except a capped sample, which is small. """ +import math import re import pandas as pd from progress import Ticker +FORMATS = ("insert", "batch", "copy") +DEFAULT_BATCH_ROWS = 500 +# Rows measured to estimate a full table's size, spread evenly through it: the +# first rows of a table are often shorter (small ids, early dates) than the rest. +MEASURE_ROWS = 1000 + def sanitize_identifier(identifier: str) -> str: """Sanitize names for SQL tables, columns, and filenames.""" @@ -31,50 +49,107 @@ def sql_value(v) -> str: return f"'{escaped}'" -def render_table(df: pd.DataFrame, table_name: str, max_rows=None): +def copy_value(v) -> str: + """The same value in COPY's text format: \\N for NULL, backslash escapes.""" + if pd.isna(v): + return r"\N" + if isinstance(v, (bool, int)): + return str(v) + if isinstance(v, float): + return str(int(v)) if v.is_integer() else str(v) + return (str(v).replace("\\", "\\\\").replace("\t", "\\t") + .replace("\n", "\\n").replace("\r", "\\r")) + + +def utf8_len(text: str) -> int: + return len(text.encode("utf-8")) + + +class Shape: + """How one table's rows are framed in one format.""" + + def __init__(self, fmt, table_name, columns, batch_rows=DEFAULT_BATCH_ROWS): + if fmt not in FORMATS: + raise ValueError(f"unknown sql format: {fmt}") + self.fmt = fmt + self.table_name = table_name + self.batch_rows = batch_rows + table_ref = f'"{table_name}"' + cols = ", ".join(f'"{sanitize_identifier(c)}"' for c in columns) + self.head = f"-- Generated seed data for table: {table_ref}\n" + if fmt == "copy": + self.head += "-- COPY FROM stdin: load with psql. COPY has no ON CONFLICT, so load it once.\n" + self.insert = f"INSERT INTO {table_ref} ({cols}) VALUES" + self.copy = f"COPY {table_ref} ({cols}) FROM stdin;\n" + + def open(self, note="") -> str: + text = self.head + note + "BEGIN;\n\n" + return text + self.copy if self.fmt == "copy" else text + + def close(self) -> str: + return ("\\.\n" if self.fmt == "copy" else "") + "\nCOMMIT;\n" + + def _row(self, values) -> str: + if self.fmt == "copy": + return "\t".join(copy_value(v) for v in values) + "\n" + vals = ", ".join(sql_value(v) for v in values) + if self.fmt == "insert": + return f"{self.insert} ({vals}) ON CONFLICT DO NOTHING;\n" + return f"({vals})" + + def _batch(self, rows) -> str: + return f"{self.insert}\n" + ",\n".join(rows) + "\nON CONFLICT DO NOTHING;\n\n" + + def units(self, df: pd.DataFrame): + """The rows as splittable pieces: one per row, or one per batch.""" + ticker = Ticker(self.table_name, len(df)) + batch = [] + # iterrows, not itertuples: it hands values over the way the original + # tool did, and seed files people already load depend on that quoting. + for i, (_, row) in enumerate(df.iterrows(), 1): + text = self._row(row) + if self.fmt == "batch": + batch.append(text) + if len(batch) == self.batch_rows: + yield self._batch(batch) + batch = [] + else: + yield text + if i % Ticker.CHECK == 0: + ticker.tick(i) + if batch: + yield self._batch(batch) + + def estimate(self, df: pd.DataFrame) -> int: + """Bytes one unsplit file of the whole table would take, from rows spread through it.""" + total = len(df) + n = min(total, MEASURE_ROWS) + picks = sorted({round(i * (total - 1) / (n - 1)) for i in range(n)}) if n > 1 else [0] + measured = df.iloc[picks] + per_row = sum(utf8_len(self._row(row)) for _, row in measured.iterrows()) / len(measured) + fixed = utf8_len(self.open() + self.close()) + if self.fmt == "batch": + batches = math.ceil(total / self.batch_rows) + framing = utf8_len(f"{self.insert}\n") + utf8_len("\nON CONFLICT DO NOTHING;\n\n") + return fixed + round((per_row + 2) * total) + batches * framing + return fixed + round(per_row * total) + + +def render_sample(shape: Shape, df: pd.DataFrame, max_rows: int): """ - Return (sql_text, total_rows, full_bytes, exact). + A capped table as one block of text, with a note of what was left out. - full_bytes is what the file would weigh with every row: measured when every - row was rendered, extrapolated from the average rendered row otherwise. + Return (text, full_bytes). Small by definition, so it is built in memory: + the note at the top needs the estimate, which needs the rows. """ - 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 = [] - ticker = Ticker(table_name, len(shown)) - # 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 i, (_, row) in enumerate(shown.iterrows(), 1): - vals = ", ".join(sql_value(v) for v in row) - rows.append(f"INSERT INTO {table_ref} ({cols}) VALUES ({vals}) ON CONFLICT DO NOTHING;\n") - if i % Ticker.CHECK == 0: - ticker.tick(i) - - 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 + shown = df.head(max_rows) + body = "".join(shape.units(shown)) + full_bytes = shape.estimate(df) + note = ( + f"-- SAMPLE: first {len(shown)} of {len(df)} rows. The full file would be " + f"~{human_bytes(full_bytes)}; run without --max-rows for all of it.\n" + ) + return shape.open(note) + body + shape.close(), full_bytes def human_bytes(n: int) -> str: