Merge branch 'rig-work'

soleprint/station/tools/dataconvert/.env-example

@@ -0,0 +1,17 @@
+# Environment Variables for dataconvert tool
+
+# Input settings
+DATACONVERT_INPUT="data/*"
+DATACONVERT_EXCLUDE="*.tmp"
+DATACONVERT_CONFIG=
+
+# Output settings
+DATACONVERT_OUT_DIR=datadir/sql
+DATACONVERT_MAX_ROWS=
+DATACONVERT_SQL_FORMAT=batch
+DATACONVERT_BATCH_ROWS=500
+DATACONVERT_MAX_FILE_SIZE=100M
+DATACONVERT_ALL_TEXT=false
+DATACONVERT_KEEP_FOLDERS=true
+DATACONVERT_NO_SCHEMA=false
+DATACONVERT_NO_DDL=false

soleprint/station/tools/dataconvert/README.md

@@ -100,7 +100,7 @@ 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
+source folders with the same table name (two exports' `users.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.

soleprint/station/tools/dataconvert/config.py

@@ -17,6 +17,8 @@ header on row 1.
       "batch_rows": 500,
       "max_file_size": "100M",
       "schema": true,
+      "include_ddl": true,
+      "load_order": ["ref_data", "users", "orders"],
       "header_row": 1,
       "layouts": [
         {
@@ -35,11 +37,11 @@ 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, sql_format, batch_rows,
-max_file_size, all_text, keep_folders, exclude) are the command line's flags
+max_file_size, all_text, keep_folders, exclude, include_ddl, load_order) are the
-with the same names, and a flag given on the command line wins over the file;
+command line's flags with the same names, and a flag given on the command line
---exclude patterns are added to the file's. out_dir is relative to the folder
+wins over the file; --exclude patterns are added to the file's. out_dir is relative
-the config file is in; max_file_size takes a number of bytes or a suffixed size
+to the folder the config file is in; max_file_size takes a number of bytes or a
-such as "100M" (decimal: k, M, G). Keys starting with _ are
+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.
 """
 
@@ -96,7 +98,8 @@ class Config:
     def __init__(self, rules=(), bare_sheet_prefixes=(), source=None, forced=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=()):
+                 all_text=None, keep_folders=None, exclude=(), include_ddl=None,
+                 load_order=()):
         self.rules = list(rules)
         self.bare_sheet_prefixes = tuple(bare_sheet_prefixes)
         self.source = source
@@ -115,6 +118,8 @@ class Config:
         self.all_text = all_text
         self.keep_folders = keep_folders
         self.exclude = tuple(exclude)
+        self.include_ddl = include_ddl
+        self.load_order = tuple(load_order)
 
     @property
     def probe_rows(self):
@@ -137,6 +142,8 @@ 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.include_ddl is not None:
+            said.append(f"include_ddl {'on' if self.include_ddl else 'off'}")
         if self.sql_format is not None:
             said.append(f"sql_format {self.sql_format}")
         if self.batch_rows is not None:
@@ -149,6 +156,8 @@ class Config:
             said.append("keep_folders")
         if self.exclude:
             said.append(f"exclude {', '.join(self.exclude)}")
+        if self.load_order:
+            said.append(f"load_order {', '.join(self.load_order)}")
         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:
@@ -158,7 +167,8 @@ class Config:
 
 FILENAME = "dataconvert.json"
 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"}
+              "sql_format", "batch_rows", "max_file_size", "all_text", "keep_folders", "exclude",
+              "include_ddl", "load_order"}
 
 
 def parse_size(value):
@@ -214,6 +224,9 @@ def load(path=None, forced=None):
     schema = raw.get("schema")
     if schema is not None and not isinstance(schema, bool):
         raise ConfigError(f"{path}: schema must be true or false")
+    include_ddl = raw.get("include_ddl")
+    if include_ddl is not None and not isinstance(include_ddl, bool):
+        raise ConfigError(f"{path}: include_ddl must be true or false")
     out_dir = raw.get("out_dir")
     if out_dir is not None:
         if not isinstance(out_dir, str) or not out_dir.strip():
@@ -242,6 +255,10 @@ def load(path=None, forced=None):
     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"')
 
+    load_order = raw.get("load_order", [])
+    if not isinstance(load_order, list) or not all(isinstance(p, str) and p for p in load_order):
+        raise ConfigError(f'{path}: load_order must be a list of table identifiers')
+
     fallback = None
     if "header_row" in raw or "data_row" in raw:
         try:
@@ -252,4 +269,5 @@ def load(path=None, forced=None):
     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,
                   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)
+                  all_text=flags["all_text"], keep_folders=flags["keep_folders"], exclude=exclude,
+                  include_ddl=include_ddl, load_order=load_order)

soleprint/station/tools/dataconvert/dataconvert-example.json

@@ -2,7 +2,7 @@
   "_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 (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",
+  "out_dir": "datadir/sql",
 
   "_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,

soleprint/station/tools/dataconvert/dataconvert.py

@@ -25,6 +25,7 @@ it is copied: run this script from anywhere, no install step.
 """
 
 import argparse
+import os
 from pathlib import Path
 
 from output import write_tables
@@ -35,6 +36,24 @@ from readers import expand_inputs, iter_sources
 from schema import SchemaReport
 
 
+def load_dotenv():
+    env_path = Path(".env")
+    if env_path.is_file():
+        for line in env_path.read_text(encoding="utf-8").splitlines():
+            line = line.strip()
+            if line and not line.startswith("#") and "=" in line:
+                key, val = line.split("=", 1)
+                if key not in os.environ:
+                    os.environ[key] = val.strip().strip('"\'')
+
+
+def get_bool_env(key, default=False):
+    val = os.getenv(key)
+    if val is None:
+        return default
+    return val.strip().lower() in ('true', '1', 'yes', 'on')
+
+
 def size(value):
     try:
         return cfg.parse_size(value)
@@ -50,35 +69,41 @@ def positive_int(value):
 
 
 def main():
+    load_dotenv()
+
+    default_input = os.getenv("DATACONVERT_INPUT")
+    default_exclude = os.getenv("DATACONVERT_EXCLUDE", "").split(",") if os.getenv("DATACONVERT_EXCLUDE") else []
+    
     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("--input", nargs="+", default=[default_input] if default_input else None, required=not bool(default_input), 
-    parser.add_argument("--out-dir", default=None,
+                        help="Input file(s), directory, wildcard pattern(s), or ZIP archive(s)")
-                        help="Directory where individual .sql files will be written (default: the config's out_dir; one of the two is required)")
+    parser.add_argument("--out-dir", default=os.getenv("DATACONVERT_OUT_DIR"),
-    parser.add_argument("--max-rows", type=positive_int, default=None,
+                        help="Directory where individual .sql files will be written (default: the config's out_dir or env)")
-                        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("--max-rows", type=positive_int, default=os.getenv("DATACONVERT_MAX_ROWS"),
-    parser.add_argument("--no-schema", action="store_true", help="Do not write SCHEMA.md, whatever the config says")
+                        help="Write at most N rows per table; the header and SCHEMA.md note the full row count and size")
-    parser.add_argument("--sql-format", choices=FORMATS, default=None,
+    parser.add_argument("--no-schema", action="store_true", default=get_bool_env("DATACONVERT_NO_SCHEMA"), 
+                        help="Do not write SCHEMA.md, whatever the config says")
+    parser.add_argument("--no-ddl", action="store_true", default=get_bool_env("DATACONVERT_NO_DDL"), 
+                        help="Do not emit CREATE TABLE IF NOT EXISTS statements before seeding")
+    parser.add_argument("--sql-format", choices=FORMATS, default=os.getenv("DATACONVERT_SQL_FORMAT"),
                         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,
+    parser.add_argument("--batch-rows", type=positive_int, default=os.getenv("DATACONVERT_BATCH_ROWS", DEFAULT_BATCH_ROWS),
                         help=f"Rows per INSERT with --sql-format batch (default {DEFAULT_BATCH_ROWS})")
-    parser.add_argument("--max-file-size", type=size, default=None,
+    parser.add_argument("--max-file-size", type=size, default=os.getenv("DATACONVERT_MAX_FILE_SIZE"),
-                        help='Split a table into numbered parts past this size, e.g. "100M" (default: never split; samples are never split)')
+                        help='Split a table into numbered parts past this size, e.g. "100M"')
-    parser.add_argument("--all-text", action="store_true", default=None,
+    parser.add_argument("--all-text", action="store_true", default=get_bool_env("DATACONVERT_ALL_TEXT", None),
-                        help="Read every column as text, exactly as in the file; only empty cells become NULL. "
+                        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=get_bool_env("DATACONVERT_KEEP_FOLDERS", None),
-    parser.add_argument("--keep-folders", action="store_true", default=None,
+                        help="Mirror the source folders under the output directory")
-                        help="Mirror the source folders under the output directory, instead of one flat folder "
+    parser.add_argument("--exclude", action="append", default=default_exclude, metavar="GLOB",
-                             "where same-named tables from different folders share a file")
+                        help='Skip matching files, e.g. "*.xlsx.ods". Repeatable, and added to the config\'s')
-    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,
                         help="First row of data, when rows sit between it and the header (default: the row after the header)")
-    parser.add_argument("--config", default=None,
+    parser.add_argument("--config", default=os.getenv("DATACONVERT_CONFIG"),
-                        help="JSON with layouts, naming and run settings, for every input (default: dataconvert.json in each input's folder, if there is one)")
+                        help="JSON with layouts, naming and run settings, for every input")
     args = parser.parse_args()
 
     forced = None
@@ -129,14 +154,16 @@ def main():
 
         # The command line wins, then the config.
         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
+        max_rows = int(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
+        include_ddl = not args.no_ddl and config.include_ddl 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
+        batch_rows = int(args.batch_rows) if args.batch_rows else (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)
+        load_order = config.load_order
 
         key = out_dir.resolve()
         if key not in reports:
@@ -146,7 +173,8 @@ def main():
 
         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 "")
+                         fmt, batch_rows, max_bytes, folder if keep_folders else "", include_ddl=include_ddl,
+                         load_order=load_order)
 
     for out_dir, report, caps in reports.values():
         if report.entries and out_dir.is_dir():

soleprint/station/tools/dataconvert/output.py

@@ -11,16 +11,16 @@ from progress import say
 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:
+def table_filename(raw_name: str, source_name: str, sheet_count: int, bare_prefixes=(), prefix="") -> str:
     """
     Sheets of a multi-sheet workbook are prefixed with the workbook, so two
     workbooks cannot collide, unless the config names the sheet as already
-    unique (bare_sheet_prefixes).
+    unique (bare_sheet_prefixes). An optional numeric prefix (e.g. 01_) forces
+    strict load order.
     """
     clean = sanitize_identifier(raw_name)
-    if sheet_count > 1 and not clean.startswith(tuple(bare_prefixes)):
+    base_name = f"{sanitize_identifier(source_name)}_{clean}.sql" if sheet_count > 1 and not clean.startswith(tuple(bare_prefixes)) else f"{clean}.sql"
-        return f"{sanitize_identifier(source_name)}_{clean}.sql"
+    return f"{prefix}{base_name}" if prefix else base_name
-    return f"{clean}.sql"
 
 
 class TableFile:
@@ -111,19 +111,34 @@ def write_full(shape: Shape, df, table_file: TableFile) -> int:
 
 
 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=""):
+                 fmt="insert", batch_rows=DEFAULT_BATCH_ROWS, max_bytes=None, folder="", include_ddl=True,
+                 load_order=()):
     """
     Write individual .sql files per table/sheet into the output directory, or
     into its `folder` subdirectory when the source folders are kept.
+    Respects DDL-first generation and load_order configuration.
     """
     target = out_dir / folder if folder else out_dir
-    for raw_name, df in dfs.items():
+
+    order_map = {sanitize_identifier(name): idx for idx, name in enumerate(load_order, start=1)}
+
+    def table_sort_key(item):
+        raw_name = item[0]
+        sanitized = sanitize_identifier(raw_name)
+        return (order_map.get(sanitized, 9999), sanitized)
+
+    sorted_dfs = sorted(dfs.items(), key=table_sort_key)
+
+    for raw_name, df in sorted_dfs:
         if df.empty:
             continue
 
         table = sanitize_identifier(raw_name)
-        filename = table_filename(raw_name, source_name, len(dfs), bare_prefixes)
+        order_idx = order_map.get(table)
-        shape = Shape(fmt, table, df.columns, batch_rows)
+        file_prefix = f"{order_idx:02d}_" if order_idx is not None else ""
+
+        filename = table_filename(raw_name, source_name, len(dfs), bare_prefixes, prefix=file_prefix)
+        shape = Shape(fmt, table, df.columns, batch_rows, include_ddl=include_ddl, df=df)
         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.

soleprint/station/tools/dataconvert/sqlgen.py

@@ -38,6 +38,33 @@ def sanitize_identifier(identifier: str) -> str:
     return clean.strip("_")
 
 
+def postgres_type(series: pd.Series) -> str:
+    """Infer a PostgreSQL column data type for DDL generation."""
+    vals = series.dropna()
+    if vals.empty:
+        return "TEXT"
+    if pd.api.types.is_bool_dtype(vals):
+        return "BOOLEAN"
+    if pd.api.types.is_integer_dtype(vals):
+        return "BIGINT"
+    if pd.api.types.is_float_dtype(vals):
+        return "NUMERIC"
+    if pd.api.types.is_datetime64_any_dtype(vals):
+        return "TIMESTAMP"
+    return "TEXT"
+
+
+def ddl_statement(table_name: str, df: pd.DataFrame) -> str:
+    """Emit DDL for table creation before seeding data so that loading does not fail."""
+    cols_def = []
+    for col in df.columns:
+        col_id = sanitize_identifier(col)
+        col_type = postgres_type(df[col])
+        cols_def.append(f'    "{col_id}" {col_type}')
+    cols_sql = ",\n".join(cols_def)
+    return f'CREATE TABLE IF NOT EXISTS "{table_name}" (\n{cols_sql}\n);\n\n'
+
+
 def sql_value(v) -> str:
     if pd.isna(v):
         return "NULL"
@@ -68,12 +95,14 @@ def utf8_len(text: str) -> int:
 class Shape:
     """How one table's rows are framed in one format."""
 
-    def __init__(self, fmt, table_name, columns, batch_rows=DEFAULT_BATCH_ROWS):
+    def __init__(self, fmt, table_name, columns, batch_rows=DEFAULT_BATCH_ROWS, include_ddl=True, df=None):
         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
+        self.include_ddl = include_ddl
+        self.df = df
         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"
@@ -84,6 +113,8 @@ class Shape:
 
     def open(self, note="") -> str:
         text = self.head + note + "BEGIN;\n\n"
+        if self.include_ddl and self.df is not None:
+            text += ddl_statement(self.table_name, self.df)
         return text + self.copy if self.fmt == "copy" else text
 
     def close(self) -> str:

soleprint/station/tools/dataconvert/uv.lock

@@ -13,6 +13,23 @@ resolution-markers = [
     "python_full_version < '3.12' and sys_platform != 'emscripten' and sys_platform != 'win32'",
 ]
 
+[[package]]
+name = "dataconvert"
+version = "0.1.0"
+source = { virtual = "." }
+dependencies = [
+    { name = "odfpy" },
+    { name = "openpyxl" },
+    { name = "pandas" },
+]
+
+[package.metadata]
+requires-dist = [
+    { name = "odfpy", specifier = ">=1.4.1" },
+    { name = "openpyxl", specifier = ">=3.1.2" },
+    { name = "pandas", specifier = ">=2.2.0" },
+]
+
 [[package]]
 name = "defusedxml"
 version = "0.7.1"
@@ -284,23 +301,6 @@ wheels = [
     { url = "https://files.pythonhosted.org/packages/ec/57/56b9bcc3c9c6a792fcbaf139543cee77261f3651ca9da0c93f5c1221264b/python_dateutil-2.9.0.post0-py2.py3-none-any.whl", hash = "sha256:a8b2bc7bffae282281c8140a97d3aa9c14da0b136dfe83f850eea9a5f7470427", size = 229892, upload-time = "2024-03-01T18:36:18.57Z" },
 ]
 
-[[package]]
-name = "dataconvert"
-version = "0.1.0"
-source = { virtual = "." }
-dependencies = [
-    { name = "odfpy" },
-    { name = "openpyxl" },
-    { name = "pandas" },
-]
-
-[package.metadata]
-requires-dist = [
-    { name = "odfpy", specifier = ">=1.4.1" },
-    { name = "openpyxl", specifier = ">=3.1.2" },
-    { name = "pandas", specifier = ">=2.2.0" },
-]
-
 [[package]]
 name = "six"
 version = "1.17.0"

soleprint/station/tools/datasource/.env-example

@@ -0,0 +1,20 @@
+# Environment Variables for datasource tools
+
+# Database Credentials
+POSTGRES_USER=app
+POSTGRES_PASSWORD=secret
+POSTGRES_DB=app
+
+# Full Database URI (Overrides user/pass/db if provided)
+TARGET_DB_URI=postgresql://app:secret@postgres:5432/app
+
+# Data Dump configuration
+DUMP_OUT_DIR=/backups/cleaned_db_dump
+DUMP_IN_DIR=/backups/cleaned_db_dump
+# Pass schemas exactly as required by pg_dump, e.g., "--schema=core --schema=sales"
+DUMP_SCHEMAS="--schema=core"
+
+# Data Load / Seeding Configuration
+SEED_DIR=/rig/datadir/sql
+TOOLS_DIR=/rig/tools/datasource
+CONF_DIR=/rig/datadir/datasource

soleprint/station/tools/datasource/.gitignore

@@ -0,0 +1,4 @@
+# The real mapping: how one producer's raw tables become the logical domains names
+# their tables and columns. It lives in the data folder (datadir/datasource/), not
+# here; this only stops a stray copy being committed. Start from schema_mapping-example.yaml.
+schema_mapping.yaml

soleprint/station/tools/datasource/README.md

@@ -0,0 +1,26 @@
+# datasource
+
+Scripts to dump, restore, and seed the database, including schema migration mapping.
+
+## Schema Mapping
+
+The data loading process converges raw seed tables into logical domains using a YAML configuration.
+This configuration is injected at runtime from `datadir/datasource/schema_mapping.yaml` (never
+committed: it names the producer's tables) rather than hardcoded in the tool. Start from
+`schema_mapping-example.yaml`.
+
+Example mapping structure:
+
+```yaml
+tables:
+  - source: raw_users
+    target: core.users
+    truncate: true
+    columns:
+      user_id: id
+      username: name
+      created_at: etl_utils.safe_cast_date(created_date)
+      is_active: "COALESCE(etl_utils.safe_cast_boolean(status), TRUE)"
+```
+
+All operational variables (database credentials, dump directories, active schemas) are read from the environment or a `.env` file at the root. See `.env-example` for available configuration keys.

soleprint/station/tools/datasource/casting_helpers.sql

@@ -0,0 +1,30 @@
+-- Create dedicated schema for ETL migration helpers
+CREATE SCHEMA IF NOT EXISTS etl_utils;
+
+-- 1. Safe String to Boolean Converter
+CREATE OR REPLACE FUNCTION etl_utils.safe_cast_boolean(val TEXT)
+RETURNS BOOLEAN AS $$ BEGIN     IF val IS NULL OR TRIM(val) = '' OR UPPER(TRIM(val)) IN ('NULL', 'N/A', '.') THEN         RETURN NULL;     ELSIF UPPER(TRIM(val)) IN ('Y', 'YES', '1', 'TRUE') THEN         RETURN TRUE;     ELSIF UPPER(TRIM(val)) IN ('N', 'NO', '0', 'FALSE') THEN         RETURN FALSE;     ELSE         RETURN NULL;     END IF; END; $$ LANGUAGE plpgsql IMMUTABLE;
+
+-- 2. Safe String to Numeric/Integer Converter
+CREATE OR REPLACE FUNCTION etl_utils.safe_cast_numeric(val TEXT)
+RETURNS NUMERIC AS $$ DECLARE     cleaned TEXT; BEGIN     cleaned := REGEXP_REPLACE(TRIM(val), '[^0-9.-]', '', 'g');     IF cleaned = '' OR cleaned = '.' OR cleaned = '-' THEN         RETURN NULL;     END IF;     RETURN cleaned::NUMERIC; EXCEPTION WHEN OTHERS THEN     RETURN NULL; END; $$ LANGUAGE plpgsql IMMUTABLE;
+
+-- 3. Safe String to Date/Timestamp Converter
+CREATE OR REPLACE FUNCTION etl_utils.safe_cast_date(val TEXT)
+RETURNS DATE AS $func$
+BEGIN
+    IF val IS NULL OR TRIM(val) = '' OR val ~* '^(UN|UNK|UNKNOWN)' THEN
+        RETURN NULL;
+    END IF;
+
+    BEGIN
+        RETURN val::DATE;
+    EXCEPTION WHEN OTHERS THEN
+        BEGIN
+            RETURN TO_DATE(val, 'DD-MON-YY');
+        EXCEPTION WHEN OTHERS THEN
+            RETURN NULL;
+        END;
+    END;
+END;
+$func$ LANGUAGE plpgsql IMMUTABLE;

soleprint/station/tools/datasource/dump_trasformed.sh

@@ -0,0 +1,20 @@
+#!/bin/bash
+set -euo pipefail
+
+# Load environment variables from .env if present
+if [ -f .env ]; then
+    export $(grep -v '^#' .env | xargs)
+fi
+
+DB_USER="${POSTGRES_USER:-postgres}"
+DB_NAME="${POSTGRES_DB:-production_db}"
+OUT_DIR="${DUMP_OUT_DIR:-/backups/cleaned_db_dump}"
+SCHEMAS="${DUMP_SCHEMAS:-}"
+
+# Dump only the transformed, strictly-typed schemas (skipping raw staging tables)
+# Note: Pass schemas in DUMP_SCHEMAS as e.g. "--schema=schema1 --schema=schema2"
+pg_dump -U "$DB_USER" -d "$DB_NAME" \
+  ${SCHEMAS} \
+  --format=directory \
+  --jobs=4 \
+  --file="$OUT_DIR"

soleprint/station/tools/datasource/load_seed_data.sh

@@ -0,0 +1,59 @@
+#!/bin/bash
+set -euo pipefail
+
+# Load environment variables from .env if present
+if [ -f .env ]; then
+    export $(grep -v '^#' .env | xargs)
+fi
+
+# Build the URI dynamically using the injected Postgres secrets
+# Fall back to the default 'app' user if the variables are missing
+USER="${POSTGRES_USER:-app}"
+PASS="${POSTGRES_PASSWORD:-}"
+DB="${POSTGRES_DB:-app}"
+
+DB_URI="${TARGET_DB_URI:-postgresql://${USER}:${PASS}@postgres:5432/${DB}}"
+
+# Point to the containerPaths mapped in your kind-config.yaml.tpl
+SEED_DIR="${SEED_DIR:-/rig/datadir/sql}"
+TOOLS_DIR="${TOOLS_DIR:-/rig/tools/datasource}"
+CONF_DIR="${CONF_DIR:-/rig/datadir/datasource}"
+
+echo "Optimizing PostgreSQL for bulk load..."
+psql -v ON_ERROR_STOP=1 "$DB_URI" <<-EOSQL
+    ALTER SYSTEM SET maintenance_work_mem = '1GB';
+    ALTER SYSTEM SET synchronous_commit = 'off';
+    ALTER SYSTEM SET checkpoint_timeout = '15min';
+    ALTER SYSTEM SET max_wal_size = '2GB';
+    SELECT pg_reload_conf();
+EOSQL
+
+echo "Loading casting helpers..."
+psql -v ON_ERROR_STOP=1 "$DB_URI" -f "$TOOLS_DIR/casting_helpers.sql"
+
+echo "Loading seed data files in single transactions..."
+if [ -d "$SEED_DIR" ]; then
+    for f in "$SEED_DIR"/*.sql; do
+        if [ -f "$f" ]; then
+            echo "Processing $f..."
+            psql -v ON_ERROR_STOP=1 -1 "$DB_URI" -f "$f"
+        fi
+    done
+else
+    echo "Warning: Seed directory $SEED_DIR not found. Skipping flat file loads."
+fi
+
+echo "Converging schemas into logical domains using configured mapping..."
+MAPPING_FILE="${SCHEMA_MAPPING_FILE:-$CONF_DIR/schema_mapping.yaml}"
+python3 "$TOOLS_DIR/migrate_schema.py" "$MAPPING_FILE"
+
+echo "Restoring standard PostgreSQL settings..."
+psql -v ON_ERROR_STOP=1 "$DB_URI" <<-EOSQL
+    ALTER SYSTEM RESET maintenance_work_mem;
+    ALTER SYSTEM RESET synchronous_commit;
+    ALTER SYSTEM RESET checkpoint_timeout;
+    ALTER SYSTEM RESET max_wal_size;
+    SELECT pg_reload_conf();
+EOSQL
+
+echo "Data load completed."

soleprint/station/tools/datasource/migrate_schema.py

@@ -0,0 +1,54 @@
+import os
+import sys
+import yaml
+import psycopg
+from psycopg import sql
+
+def main():
+    mapping_file = os.getenv("SCHEMA_MAPPING_FILE", "/rig/datadir/datasource/schema_mapping.yaml")
+    
+    if len(sys.argv) >= 2:
+        mapping_file = sys.argv[1]
+
+    db_uri = os.getenv("TARGET_DB_URI", "postgresql://postgres:postgres@postgres:5432/postgres")
+
+    print(f"Loading schema mapping from {mapping_file}...")
+    with open(mapping_file, 'r') as f:
+        config = yaml.safe_load(f)
+
+    with psycopg.connect(db_uri) as conn:
+        with conn.cursor() as cur:
+            for table in config.get('tables', []):
+                source = table['source']
+                target = table['target']
+                columns = table['columns']
+
+                print(f"Migrating {source} to {target}...")
+
+                if table.get('truncate', False):
+                    try:
+                        cur.execute(f"TRUNCATE TABLE {target} RESTART IDENTITY CASCADE;")
+                    except psycopg.errors.UndefinedTable:
+                        print(f"  Target table {target} does not exist yet; skipping truncate.")
+                        conn.rollback()
+
+                target_cols = ", ".join(columns.keys())
+                source_exprs = ", ".join(columns.values())
+
+                query = f"""
+                    INSERT INTO {target} ({target_cols})
+                    SELECT {source_exprs}
+                    FROM {source}
+                """
+                try:
+                    cur.execute(query)
+                    print(f"  Successfully migrated {source}.")
+                except Exception as e:
+                    print(f"  Failed to migrate {source}: {e}")
+                    conn.rollback()
+                    raise
+        conn.commit()
+    print("Schema migration complete.")
+
+if __name__ == '__main__':
+    main()

soleprint/station/tools/datasource/restore_transformed.sh

@@ -0,0 +1,18 @@
+#!/bin/bash
+set -euo pipefail
+
+# Load environment variables from .env if present
+if [ -f .env ]; then
+    export $(grep -v '^#' .env | xargs)
+fi
+
+DB_USER="${POSTGRES_USER:-postgres}"
+DB_NAME="${POSTGRES_DB:-production_db}"
+IN_DIR="${DUMP_IN_DIR:-/backups/cleaned_db_dump}"
+
+# Multi-threaded restore into the target PostgreSQL cluster pod
+pg_restore -U "$DB_USER" -d "$DB_NAME" \
+  --jobs=4 \
+  --clean \
+  --if-exists \
+  "$IN_DIR"

soleprint/station/tools/datasource/schema_mapping-example.yaml

@@ -0,0 +1,21 @@
+# Schema Conversion and Type Cast Registry
+# Maps raw seed tables to the unified logical domains.
+
+tables:
+  - source: raw_users
+    target: core.users
+    truncate: true
+    columns:
+      user_id: id
+      username: name
+      created_at: etl_utils.safe_cast_date(created_date)
+      is_active: "COALESCE(etl_utils.safe_cast_boolean(status), TRUE)"
+
+  - source: raw_orders
+    target: core.orders
+    truncate: true
+    columns:
+      order_id: id
+      user_id: user_id
+      total_amount: etl_utils.safe_cast_numeric(amount)
+      order_date: etl_utils.safe_cast_date(date)

soleprint/station/tools/distill/distill-example.json

@@ -1,29 +1,97 @@
 {
-  "_comment": "Template for distill.json, the job distill.sh reads when no repo is named on the command line. Copy this to distill.json (gitignored) and edit that. Everything that shapes the run is at the top; 'repos' is just a list of paths. Preview any change with: ./distill.sh list",
+  "_comment": "Template for distill.json, the job distill.sh reads when no repo is named on the command line. Copy this to distill.json (gitignored) and edit that — which repos a machine distils is not a fact about the tool. Every key the file understands appears below, once, with the value it would fall back to if you deleted the line. Preview any change with: ./distill.sh list  — or ./distill.sh check, which resolves every path and ref in about a second.",
 
-  "_command": "tree = a directory per repo. digest = one .md per repo, flattened into a single readable file. both = each repo as a directory AND a .md. list = write nothing, just report what would be kept.",
+  "_command": "tree = a directory per repo, a copy you can open, grep and build. digest = one .md per repo, the whole thing flattened into a single readable file (header, file tree, manifest, then every file under its path). both = each repo comes out as a directory AND a .md — they answer different questions and the selection work is shared, so the second costs only a copy. list = write nothing, just report what would be kept, weighed. check = write nothing and read nothing but git's index: does every path exist, does every ref resolve, is there room. Every other command runs check first anyway.",
   "command": "list",
 
+  "_out": "Where the run writes. Relative to wherever you invoke the script from, so an absolute path is the safer thing to write down; ~ is expanded. Ignored by list and check.",
   "out": "distilled",
 
-  "_branch_mode": "full = each branch is a complete, standalone copy. diff = only the files that differ from diff_base (it gets a _PARTIAL.md saying so).",
+  "_root": "Where a bare slug in 'repos' resolves, so entries can read \"path\": \"myrepo\" instead of the whole absolute path. Leave it out and slugs resolve against the current directory, or DISTILL_ROOT. Absolute paths never touch it.",
+  "root": "~/wdir",
+
+  "_branch_mode": "full = each branch is a complete, standalone copy. diff = only the files that differ from diff_base, which is far smaller but is a changelist, not the branch (it gets a _PARTIAL.md saying so). diff_base must name one of the refs actually listed, and it is the one that comes out whole. Set per entry to override this default.",
   "branch_mode": "full",
   "diff_base": "main",
 
-  "_filters": "Applied to every repo. exclude/include are path globs; a pattern with no / also matches basenames at any depth. 'all' keeps the noise (lockfiles, images, minified, maps). max_bytes skips anything larger and names it in MANIFEST.md.",
+  "_filters": "Applied to every repo, and overridable per entry. exclude/include are path globs; a pattern with no / also matches basenames at any depth. 'all' keeps the noise (lockfiles, minified, maps, compiled output, caches) that is dropped by default. max_bytes drops anything larger outright, and names it in MANIFEST.md.",
-  "exclude": [],
+  "exclude": ["*.ipynb", "fixtures/**"],
   "include": [],
   "all": false,
-  "max_bytes": null,
+  "max_bytes": "2M",
 
-  "_split_tokens": "A digest over this many tokens is written as NAME.md (an index: the tree, and the manifest saying which part holds each file) plus NAME.part-01.md, NAME.part-02.md... holding the files, cut between files. Each part stands on its own. Default 100k; 0 never splits. Can also be set per entry.",
+  "_budget": "Three different answers to 'this is too much to read', and they do not compete. clip_bytes inlines only the head and tail of any file over that size, with a marker saying how much was cut. max_tokens holds each digest to roughly that many tokens by clipping largest-first — one shared ceiling, lowered until the total fits — so the biggest file pays and the hundred small ones that describe the project keep every line. Both of those lose text; split_tokens loses none, it just stops writing one enormous file: over ~N tokens the digest becomes NAME.md (an index holding the tree and the manifest, saying which part each file is in) plus NAME.part-01.md, NAME.part-02.md… cut between files in path order, each part standing on its own. Default 100k; 0 never splits. Clipping is a statement about the document, not the files: the tree copy always has them whole. Numbers take k/M/G, decimal.",
+  "clip_bytes": null,
+  "max_tokens": null,
   "split_tokens": "100k",
+
+  "_with_root": "Only matters when an entry names a subpath. Keeps the repo's top-level files (README, pyproject.toml, package.json) alongside the subtree, so the copy still says which project it is a part of.",
+  "with_root": false,
+
+  "_keep_secrets": ".env files, private keys, keystores and the like are dropped by default and named in MANIFEST.md. 'all' does not bring them back — only this does, and only deliberately.",
+  "keep_secrets": false,
+
+  "_run": "skip_unchanged leaves alone anything whose source and settings have not moved since the last run into this destination. prune deletes what an earlier run into 'out' wrote and this run no longer produces, so dropping a repo from the list drops its output too. Anything distill did not write (it keeps the list in out/.distill-owned) is never touched, so 'out' can be a folder shared with other things.",
   "skip_unchanged": true,
   "prune": true,
 
-  "_repos": "'path' is absolute, or a name resolved under --root. 'branches' is optional — leave it out for the working tree as it stands, uncommitted changes included. A path may appear more than once.",
+  "_bundle": "Also write _BUNDLE.md: every .md in the destination concatenated, in one document, behind an index. A directory of files is right for grep and for reading; an upload box takes one file. Costs a second copy of the text on disk and nothing else. Note that it concatenates whatever split_tokens produced, parts included — bundle and split are opposite answers to the same question, and you can have both on disk.",
+  "bundle": false,
+
+  "_refs_patch": "Put the full diff, not just the diffstat, in NAME@REFS.md. Answers 'what changed' instead of 'how much'.",
+  "refs_patch": false,
+
+  "_raw_fences": "Write runs of backticks and tildes into the digest as they are, instead of escaping them as ⟪BT3⟫ / ⟪TL3⟫. Escaping is the default because a fence inside a file otherwise closes the fence the digest is being read inside; explode.sh puts the characters back. Turn this off only when nothing is going to render the result as markdown.",
+  "raw_fences": false,
+
+  "_top": "'list' only: how many heavy files and directories to show. 0 for none.",
+  "top": 10,
+
+  "_repos": "One entry per thing you want out. 'path' is absolute, or a name resolved under 'root'. The same path may appear as many times as you like, which is why this is a list rather than a keyed object — one entry per repo could not hold two branches of the same repo. Every filter and budget key above may be repeated inside an entry, where it replaces the top-level value for that entry alone; a command-line option still wins over both.",
   "repos": [
-    { "path": "/path/to/some-repo" },
+    {
-    { "path": "/path/to/another-repo", "branches": ["origin/main", "origin/feature/example"] }
+      "_note": "The plainest entry: the working tree as it stands right now, uncommitted changes included, under every setting from the top of the file.",
+      "path": "/path/to/some-repo"
+    },
+
+    {
+      "_note": "Named refs. Each is read with ls-tree/archive, so nothing is checked out and a dirty working tree is never touched. In full mode each of these is a complete copy. A hash, a tag or HEAD~3 works anywhere a branch name does.",
+      "path": "/path/to/another-repo",
+      "branches": ["origin/main", "v1.4.0", "3f2a9c1"]
+    },
+
+    {
+      "_note": "Many branches. In full mode this would be one complete copy per branch, most of them the same bytes — the right answer for an archive and the wrong one for anything with a context window. So: origin/main whole, and the rest as changelists against it.",
+      "path": "/path/to/busy-repo",
+      "branch_mode": "diff",
+      "diff_base": "origin/main",
+      "branches": ["origin/main", "origin/topic-a", "origin/topic-b"]
+    },
+
+    {
+      "_note": "A subtree, and only it — with the repo's own ignore rules, refs and deltas still applying, because the repo root is found for you. subpath may be a list: 'the part I am working on' is usually the module plus the tests or docs beside it, and two entries for that would be two outputs to read separately. 'name' overrides what the output is called, which is what you want when the same repo appears more than once.",
+      "path": "slug-resolved-under-root",
+      "subpath": ["src/api", "tests/api"],
+      "with_root": true,
+      "name": "api-only"
+    },
+
+    {
+      "_note": "Per-entry filters and budget. One generated file does not get to spend the whole digest; this entry also keeps the lockfiles the top of the file drops, and splits later than everything else.",
+      "path": "/path/to/heavy-repo",
+      "include": ["src/**", "*.md"],
+      "exclude": ["src/generated/**"],
+      "all": true,
+      "max_bytes": "5M",
+      "clip_bytes": "64k",
+      "max_tokens": "80k",
+      "split_tokens": "250k"
+    },
+
+    {
+      "_note": "Kept in the file, out of the run. Better than deleting the entry when it is something you distil occasionally — the paths and the branch list stay written down.",
+      "path": "/path/to/retired-repo",
+      "enabled": false
+    }
   ]
 }

soleprint/station/tools/distill/distill.sh

@@ -128,8 +128,12 @@
 #   --strict        refuse a dirty working tree (only affects worktree copies)
 #   --skip-unchanged  leave alone anything whose source and settings have not
 #                   moved since the last run into this destination
-#   --prune         delete anything in DEST this run did not produce, so
+#   --prune         delete what an earlier run into DEST wrote and this run no
-#                   dropping a repo from the list drops its output too
+#                   longer produces, so dropping a repo from the list drops its
+#                   output too. Only what distill created is ever removed
+#                   (DEST/.distill-owned records it): other things in DEST, and
+#                   folders that existed before a run first synced into them,
+#                   are never touched, so DEST can be a shared folder
 #   --bundle        also write DEST/_BUNDLE.md: every digest concatenated into
 #                   one document, for anything that takes a single file
 #   --refs-patch    put the full diff, not just the diffstat, in NAME@REFS.md
@@ -1350,6 +1354,17 @@ STATE_FILE=""
 STATE_OLD="$TMP/state.old"
 STATE_NEW="$TMP/state.new"
 PRODUCED=()
+# Which top-level entries of DEST distill wrote, as of the previous run. Prune
+# reads this, not the directory: DEST may be a folder shared with things that
+# were never distill's, and those must survive a prune however it is set.
+#
+# Owned means created, not merely written to. An entry that was already there
+# the first time a run wrote into it — a folder being synced into, holding files
+# of its own — is updated but never claimed, so dropping it from the list can
+# never delete what was there before distill.
+OWNED_FILE=""
+OWNED_OLD="$TMP/owned.old"
+EXISTED_BEFORE="$TMP/existed.before"
 
 # Saved after every entry, not once at the end. A run over a dozen repos and
 # their branches takes long enough to get interrupted, and a record written only
@@ -1847,6 +1862,10 @@ if [ "$CMD" != list ] && [ "$CMD" != check ] && [ -z "$DRY" ]; then
     : > "$STATE_NEW"
     produced "$DEST/MANIFEST.md"
     produced "$STATE_FILE"
+    OWNED_FILE="$DEST/.distill-owned"
+    [ -f "$OWNED_FILE" ] && cp "$OWNED_FILE" "$OWNED_OLD"
+    find "$DEST" -mindepth 1 -maxdepth 1 -printf '%f\n' > "$EXISTED_BEFORE"
+    produced "$OWNED_FILE"
 fi
 
 # Repos named on the command line run under the global options, as before.
@@ -1950,16 +1969,19 @@ if [ -n "$BUNDLE" ] && [ "$CMD" != tree ] && [ "$CMD" != list ] && [ -z "$DRY" ]
     fi
 fi
 
-# Anything at the top of the destination this run did not produce came from a
+# Something distill wrote on an earlier run and this run did not produce came
-# previous, longer list. Scoped to depth 1 and to a destination we just wrote
+# from a previous, longer list. Only that: an entry distill never wrote is not
-# to: this deletes, so it should never go hunting.
+# an orphan but someone else's, and DEST may well be a folder shared with them.
-if [ -n "$PRUNE" ] && [ "$CMD" != list ] && [ -z "$DRY" ]; then
+# Scoped to depth 1 and to a destination we just wrote to: this deletes, so it
+# should never go hunting.
+if [ -n "$PRUNE" ] && [ "$CMD" != list ] && [ -z "$DRY" ] && [ -f "$OWNED_OLD" ]; then
     while IFS= read -r -d '' entry; do
         keep=""
         # A digest left alone as unchanged registers only its index, not the
         # parts beside it, which belong to it all the same.
         owner="$entry"
         case "$entry" in *.part-[0-9][0-9].md) owner="${entry%.part-[0-9][0-9].md}.md" ;; esac
+        grep -qxF "$(basename "$owner")" "$OWNED_OLD" || continue
         for kept in ${PRODUCED[@]+"${PRODUCED[@]}"}; do
             { [ "$entry" = "$kept" ] || [ "$owner" = "$kept" ]; } && { keep=1; break; }
         done
@@ -1970,6 +1992,27 @@ if [ -n "$PRUNE" ] && [ "$CMD" != list ] && [ -z "$DRY" ]; then
     done < <(find "$DEST" -mindepth 1 -maxdepth 1 -print0)
 fi
 
+# Written whether or not prune is on: what this run created, plus whatever an
+# earlier run created that is still on disk, so turning prune on later still
+# knows everything distill put here.
+if [ -n "$OWNED_FILE" ]; then
+    {
+        for kept in ${PRODUCED[@]+"${PRODUCED[@]}"}; do
+            kept="$(basename "$kept")"
+            if ! grep -qxF "$kept" "$EXISTED_BEFORE" \
+               || { [ -f "$OWNED_OLD" ] && grep -qxF "$kept" "$OWNED_OLD"; }; then
+                printf '%s\n' "$kept"
+            fi
+        done
+        if [ -f "$OWNED_OLD" ]; then
+            while IFS= read -r o; do
+                if [ -n "$o" ] && [ -e "$DEST/$o" ]; then printf '%s\n' "$o"; fi
+            done < "$OWNED_OLD"
+        fi
+    } | LC_ALL=C sort -u > "$OWNED_FILE.partial"
+    mv "$OWNED_FILE.partial" "$OWNED_FILE"
+fi
+
 if [ -n "$SKIP_UNCHANGED" ] && [ "$CMD" != list ] && [ -z "$DRY" ]; then
     mv "$STATE_NEW" "$STATE_FILE"
 fi