From 7b70b7edc6cd1d9a3d89394bfa80c82809ace3c5 Mon Sep 17 00:00:00 2001 From: buenosairesam Date: Wed, 16 Sep 2026 09:13:47 -0300 Subject: [PATCH] adapter updates --- soleprint/station/tools/dataconvert/README.md | 1 + soleprint/station/tools/dataconvert/config.py | 15 +++-- .../station/tools/dataconvert/dataconvert.py | 6 +- soleprint/station/tools/dataconvert/output.py | 3 +- .../station/tools/dataconvert/progress.py | 39 +++++++++++++ .../station/tools/dataconvert/readers.py | 55 +++++++++++++------ soleprint/station/tools/dataconvert/sqlgen.py | 7 ++- 7 files changed, 97 insertions(+), 29 deletions(-) create mode 100644 soleprint/station/tools/dataconvert/progress.py diff --git a/soleprint/station/tools/dataconvert/README.md b/soleprint/station/tools/dataconvert/README.md index 9cbea8a..8529e62 100644 --- a/soleprint/station/tools/dataconvert/README.md +++ b/soleprint/station/tools/dataconvert/README.md @@ -89,6 +89,7 @@ what pandas read: a starting point, not DDL. `--no-schema` skips the file. | `dataconvert.py` | command line | | `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 | | `schema.py` | `SCHEMA.md` | diff --git a/soleprint/station/tools/dataconvert/config.py b/soleprint/station/tools/dataconvert/config.py index 3c5b0e3..97a7386 100644 --- a/soleprint/station/tools/dataconvert/config.py +++ b/soleprint/station/tools/dataconvert/config.py @@ -101,17 +101,16 @@ class Config: self.out_dir = out_dir @property - def detects(self): - return self.forced is None and (bool(self.rules) or self.fallback is not None) + def probe_rows(self): + """How many rows the layouts' marker cells need to be read.""" + return max((rule.row for rule in self.rules), default=0) - def layout_for(self, raw_df): - """The layout for a sheet read without a header, or None for the default read.""" - if self.forced is not None: - return None if self.forced.is_default else self.forced + def match(self, probe_df): + """The first layout whose marker cell is in these top rows, or None.""" for rule in self.rules: - if rule.matches(raw_df): + if rule.matches(probe_df): return rule.layout - return self.fallback + return None def settings(self): """What the file set, for the line the run prints.""" diff --git a/soleprint/station/tools/dataconvert/dataconvert.py b/soleprint/station/tools/dataconvert/dataconvert.py index 4ff2f05..d2f1dae 100644 --- a/soleprint/station/tools/dataconvert/dataconvert.py +++ b/soleprint/station/tools/dataconvert/dataconvert.py @@ -28,6 +28,7 @@ import argparse from pathlib import Path from output import write_tables +from progress import say import config as cfg from readers import expand_inputs, iter_sources from schema import SchemaReport @@ -100,7 +101,7 @@ def main(): if config.source is not None and config.source not in announced: announced.add(config.source) said = config.settings() - print(f"[dataconvert] config: {config.source}" + (f" ({said})" if said else "")) + say(f"config: {config.source}" + (f" ({said})" if said else "")) # The command line wins, then the config. out_dir = Path(args.out_dir) if args.out_dir else config.out_dir @@ -121,7 +122,8 @@ def main(): # The sampling note names the cap only when every input here used # the same one; otherwise each table's row still says what it kept. cap = next(iter(caps)) if len(caps) == 1 else None - print(f"[dataconvert] Generated: {report.write(out_dir, cap)}") + say(f"writing SCHEMA.md for {len(report.tables())} tables") + say(f"Generated: {report.write(out_dir, cap)}") if __name__ == "__main__": diff --git a/soleprint/station/tools/dataconvert/output.py b/soleprint/station/tools/dataconvert/output.py index 4af5c05..7fd202a 100644 --- a/soleprint/station/tools/dataconvert/output.py +++ b/soleprint/station/tools/dataconvert/output.py @@ -4,6 +4,7 @@ Output: one .sql file per table or sheet, named after it. from pathlib import Path +from progress import say from sqlgen import render_table, sanitize_identifier @@ -43,4 +44,4 @@ def write_tables(dfs: dict, out_dir: Path, source_name: str, max_rows=None, repo 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}") + say(f"Generated: {out_file}{suffix}") diff --git a/soleprint/station/tools/dataconvert/progress.py b/soleprint/station/tools/dataconvert/progress.py new file mode 100644 index 0000000..0c21cfb --- /dev/null +++ b/soleprint/station/tools/dataconvert/progress.py @@ -0,0 +1,39 @@ +""" +Progress lines: enough to see that a long run is alive and where it is, not so +many that printing them slows it down. + +Every line is flushed, so progress shows up as it happens even when the output +is piped or tee'd into a log. +""" + +import time + +PREFIX = "[dataconvert]" + + +def say(message): + print(f"{PREFIX} {message}", flush=True) + + +def seconds(start): + return f"{time.monotonic() - start:.0f}s" + + +class Ticker: + """How far a long loop has got, at most once every `every` seconds.""" + + # How often the loop asks. Checking the clock on every row would itself be + # a measurable cost on a million-row table. + CHECK = 2000 + + def __init__(self, label, total, every=10.0): + self.label = label + self.total = total + self.every = every + self.started = self.last = time.monotonic() + + def tick(self, done): + now = time.monotonic() + if now - self.last >= self.every: + self.last = now + say(f" {self.label}: {done} of {self.total} rows ({now - self.started:.0f}s)") diff --git a/soleprint/station/tools/dataconvert/readers.py b/soleprint/station/tools/dataconvert/readers.py index df363ba..1cfd565 100644 --- a/soleprint/station/tools/dataconvert/readers.py +++ b/soleprint/station/tools/dataconvert/readers.py @@ -8,11 +8,15 @@ one per spreadsheet or CSV. Nothing here knows about SQL or output files. import glob import os import tempfile +import time import zipfile from pathlib import Path import pandas as pd +from progress import say, seconds +from sqlgen import human_bytes + SUPPORTED = {".csv", ".xlsx", ".xls", ".ods"} @@ -23,7 +27,7 @@ def expand_inputs(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}'") + print(f"[Warning] No files matched wildcard pattern: '{in_str}'", flush=True) paths.extend(Path(m) for m in sorted(matched)) else: paths.append(Path(in_str)) @@ -34,16 +38,22 @@ def read_sheet(read, config): """ One sheet or CSV, laid out as the config says. - With nothing to detect, a plain read, exactly as before. Otherwise the sheet - is read once without a header, checked against the layouts, and when one - matches the header and data rows are sliced out of that same read. + With no layout to apply, a plain read, exactly as before. The layouts' + marker cells are checked on only the top few rows, so a big sheet that + matches none of them is read once, not twice. When a layout applies, the + sheet is read without a header and the header and data rows are sliced out. """ - if not config.detects and config.forced is None: - return read(), None - raw = read(header=None) - layout = config.layout_for(raw) + layout = None + if config.forced is not None: + layout = None if config.forced.is_default else config.forced + else: + if config.rules: + layout = config.match(read(header=None, nrows=config.probe_rows)) + if layout is None: + layout = config.fallback if layout is None: return read(), None + raw = read(header=None) data = raw.iloc[layout.data_row - 1:].copy() data.columns = [str(c).strip() for c in raw.iloc[layout.header_row - 1]] return data, layout @@ -54,26 +64,37 @@ def load_dataframes_from_file(file_path: Path, config) -> dict: ext = file_path.suffix.lower() dfs = {} + # Said before the read, not after: a big workbook can take minutes inside + # pandas, and this line is what says which file that is. + say(f"reading {file_path.name} ({human_bytes(file_path.stat().st_size)})") try: if ext == ".csv": - dfs[file_path.stem], layout = read_sheet(lambda **kw: pd.read_csv(file_path, **kw), config) - note_layout(file_path.name, None, layout) + start = time.monotonic() + # low_memory=False: by default pandas types a big CSV chunk by + # 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) + dfs[file_path.stem] = df + say(f" {len(df)} rows{layout_note(layout)} ({seconds(start)})") elif ext in [".xlsx", ".xls", ".ods"]: xls = pd.ExcelFile(file_path, engine="odf") if ext == ".ods" else pd.ExcelFile(file_path) for sheet in xls.sheet_names: - dfs[sheet], layout = read_sheet( + start = time.monotonic() + df, layout = read_sheet( lambda sheet=sheet, **kw: pd.read_excel(xls, sheet_name=sheet, **kw), config) - note_layout(file_path.name, sheet, layout) + dfs[sheet] = df + say(f" sheet {sheet}: {len(df)} rows{layout_note(layout)} ({seconds(start)})") except Exception as e: - print(f"[Warning] Could not read '{file_path.name}': {e}") + print(f"[Warning] Could not read '{file_path.name}': {e}", flush=True) return dfs -def note_layout(file_name, sheet, layout): - if layout is not None: - where = f"{file_name} [{sheet}]" if sheet else file_name - print(f"[dataconvert] {where}: layout '{layout.name}', header row {layout.header_row}, data from row {layout.data_row}") +def layout_note(layout): + if layout is None: + return "" + return f", layout '{layout.name}' (header row {layout.header_row}, data from row {layout.data_row})" def iter_sources(path: Path, config): diff --git a/soleprint/station/tools/dataconvert/sqlgen.py b/soleprint/station/tools/dataconvert/sqlgen.py index 11fdb13..a9d8b96 100644 --- a/soleprint/station/tools/dataconvert/sqlgen.py +++ b/soleprint/station/tools/dataconvert/sqlgen.py @@ -10,6 +10,8 @@ import re import pandas as pd +from progress import Ticker + def sanitize_identifier(identifier: str) -> str: """Sanitize names for SQL tables, columns, and filenames.""" @@ -45,11 +47,14 @@ def render_table(df: pd.DataFrame, table_name: str, max_rows=None): 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 _, row in shown.iterrows(): + 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"