update dataconvert and source

soleprint/station/tools/dataconvert/readers.py

@@ -17,7 +17,7 @@ from pathlib import Path, PurePosixPath
 import pandas as pd
 
 from progress import say, seconds
-from sqlgen import human_bytes
+from sqlgen import human_bytes, sanitize_identifier
 
 SUPPORTED = {".csv", ".xlsx", ".xls", ".ods"}
 
@@ -68,13 +68,33 @@ def read_sheet(read, config):
         if layout is None:
             layout = config.fallback
     if layout is None:
-        return read(), None
+        data = read()
+        data.columns = usable_columns(data.columns)
+        return data, 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]]
+    data.columns = usable_columns(str(c).strip() for c in raw.iloc[layout.header_row - 1])
     return data, layout
 
 
+def usable_columns(names) -> list:
+    """
+    Header cells as SQL can take them: an empty one (or one of symbols only, which
+    sanitizes to nothing) becomes column_<n>, n counted from 1 as the spreadsheet does,
+    and a name that repeats once sanitized gets _2, _3. Every other name is kept as written.
+    """
+    out, seen = [], set()
+    for n, name in enumerate(names, start=1):
+        name = str(name)
+        base = name if sanitize_identifier(name) else f"column_{n}"
+        candidate, k = base, 2
+        while sanitize_identifier(candidate) in seen:
+            candidate, k = f"{base}_{k}", k + 1
+        seen.add(sanitize_identifier(candidate))
+        out.append(candidate)
+    return out
+
+
 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()

soleprint/station/tools/dataconvert/sqlgen.py

@@ -62,7 +62,11 @@ def ddl_statement(table_name: str, df: pd.DataFrame) -> str:
         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'
+    # Several sources can feed one table (same-named files in different folders), each
+    # block with its own CREATE; only the first creates. Each block then adds the columns
+    # only it has, so the table ends up with every source's columns, NULL where one lacks them.
+    adds = "".join(f'ALTER TABLE "{table_name}" ADD COLUMN IF NOT EXISTS{line};\n' for line in cols_def)
+    return f'CREATE TABLE IF NOT EXISTS "{table_name}" (\n{cols_sql}\n);\n{adds}\n'
 
 
 def sql_value(v) -> str:
@@ -114,6 +118,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:
+            # IF NOT EXISTS says "skipping" once per column per block; keep the load log to errors.
+            text += "SET LOCAL client_min_messages = warning;\n"
             text += ddl_statement(self.table_name, self.df)
         return text + self.copy if self.fmt == "copy" else text
 

soleprint/station/tools/datasource/README.md

@@ -23,4 +23,8 @@ tables:
       is_active: "COALESCE(etl_utils.safe_cast_boolean(status), TRUE)"
 ```
 
+A target that does not exist yet is created from the mapping: its schema, then a table whose
+columns are the mapping's names, typed by its expressions (`safe_cast_date` gives a `date`). A
+target that already exists, from a real DDL with keys and constraints, is used as it is.
+
 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/migrate_schema.py

@@ -25,16 +25,20 @@ def main():
 
                 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())
 
+                # A missing target is made from the mapping itself: its schema, then a table
+                # shaped by the expressions (safe_cast_date gives a date column, and so on).
+                # A target that exists, from a real DDL with keys and constraints, is left alone.
+                if "." in target:
+                    cur.execute(f"CREATE SCHEMA IF NOT EXISTS {target.split('.', 1)[0]};")
+                shaped = ", ".join(f"{expr} AS {col}" for col, expr in columns.items())
+                cur.execute(f"CREATE TABLE IF NOT EXISTS {target} AS SELECT {shaped} FROM {source} WITH NO DATA;")
+
+                if table.get('truncate', False):
+                    cur.execute(f"TRUNCATE TABLE {target} RESTART IDENTITY CASCADE;")
+
                 query = f"""
                     INSERT INTO {target} ({target_cols})
                     SELECT {source_exprs}