Port extraction layer from pyodbc to mdbtools for macOS

macOS has no Access ODBC driver, so the pyodbc-based extract.py could not run. Rewrite it to shell out to mdbtools (mdb-tables/mdb-export) while keeping the public interface (connect/list_tables/read_table/ sanitize_column_name) unchanged, so load_staging.py and config.py are untouched. connect() now returns the file path as the handle (mdbtools has no persistent connection). Behavior details: - -b strip drops LONGBINARY/OLE bytes so blobs never corrupt the CSV (documents are extracted separately in migration step 4). - ISO date/datetime output (-D/-T); staging read as text (dtype=str), only empty fields treated as null (keep_default_na=False) so literal "NA"/"NULL" data strings survive. - mdbtools reads deleted/corrupted records mdbtools omits rather than aborting, so the old per-row skip loop is no longer needed. Verified end-to-end: full staging load reproduces the original Windows run (82 tables, 0 errors) with exact row counts (datos2 16000, EFECTIVO 13697, DATGRAL 1172/1070, DATMEX 1520) and recovers all 764 MULT rows (the pyodbc path lost 1 to HY109 corruption). Accented-column tables (PROPANO) read cleanly. requirements.txt: drop pyodbc/pywin32 (Windows-only), keep pandas/pyarrow/ sqlalchemy/pymysql; document the Windows-only DAO catalog as historical. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
2026-07-22 17:15:02 -07:00
parent 22a53bc562
commit dc9866c24a
2 changed files with 124 additions and 48 deletions
@@ -1,32 +1,80 @@
 """
-Reads tables out of an Access database via pyodbc.
+Reads tables out of an Access database via mdbtools.

-Column metadata is read from cursor.description after a SELECT * rather than
-via cursor.columns() — the latter hits a UTF-16 decode bug in pyodbc/the
-Access ODBC driver on a subset of tables whose column names contain certain
-accented characters (confirmed against PROPANO, FALTANTES AGUA, TIT in this
-session). SELECT * + description does not hit that path.
+The project originally used pyodbc + the Windows "Microsoft Access Driver
+(*.mdb, *.accdb)". That driver is Windows-only; after the move to macOS the
+extraction layer was reworked to shell out to mdbtools (`mdb-tables`,
+`mdb-export`) instead — install with `brew install mdbtools`.
+
+The public interface (connect / list_tables / read_table / sanitize_column_name)
+is unchanged so load_staging.py and config.py did not need to change. mdbtools
+has no persistent connection object, so `connect` just returns the file path and
+the other functions take that "handle".
+
+Notes carried over from the pyodbc era:
+  - Some tables have accented column names ("AÑO1" etc.) that broke pyodbc's
+    UTF-16 path. mdb-export emits UTF-8 and reads those cleanly (verified
+    against PROPANO / FALTANTES AGUA).
+  - MULT has a storage-level corrupted/deleted record that made pyodbc abort
+    the whole table (HY109). mdbtools reads pages directly and simply omits
+    deleted records, so no special per-row skip loop is needed here; any
+    warnings mdb-export prints to stderr are logged, not fatal.
+  - LONGBINARY blob columns are exported with `-b strip` (binary content
+    dropped) so blob bytes never pollute the CSV. Document/blob extraction is a
+    separate step (migration plan step 4) with its own tooling.
 """

 from __future__ import annotations

+import io
 import re
-import pyodbc
+import shutil
+import subprocess
+from pathlib import Path
+
 import pandas as pd

-ACCESS_DRIVER = "Microsoft Access Driver (*.mdb, *.accdb)"
+# mdb-export date/datetime format: emit ISO so downstream parsing is unambiguous.
+_DATE_FMT = "%Y-%m-%d"
+_DATETIME_FMT = "%Y-%m-%d %H:%M:%S"


-def connect(path) -> pyodbc.Connection:
-    conn_str = f"DRIVER={{{ACCESS_DRIVER}}};DBQ={path};"
-    return pyodbc.connect(conn_str, autocommit=True)
+def _require_mdbtools() -> None:
+    if shutil.which("mdb-export") is None or shutil.which("mdb-tables") is None:
+        raise RuntimeError(
+            "mdbtools not found on PATH (need mdb-tables and mdb-export). "
+            "Install with: brew install mdbtools"
+        )


-def list_tables(cnxn: pyodbc.Connection) -> list[str]:
-    cursor = cnxn.cursor()
+def connect(path):
+    """No real connection with mdbtools — return the source path as the handle.
+
+    Kept for interface compatibility with the old pyodbc-based module (and so
+    load_staging.py's connect()/list_tables()/read_table() flow is unchanged).
+    """
+    _require_mdbtools()
+    p = Path(path)
+    if not p.exists():
+        raise FileNotFoundError(f"Access file not found: {p}")
+    return p
+
+
+def list_tables(cnxn) -> list[str]:
+    path = Path(cnxn)
+    result = subprocess.run(
+        ["mdb-tables", "-1", str(path)],
+        capture_output=True,
+        text=True,
+        encoding="utf-8",
+        errors="replace",
+        check=True,
+    )
    tables = []
-    for row in cursor.tables(tableType="TABLE"):
-        name = row.table_name
+    for name in result.stdout.splitlines():
+        name = name.strip()
+        if not name:
+            continue
        if name.startswith("MSys") or name.startswith("~"):
            continue
        tables.append(name)
@@ -34,10 +82,10 @@ def list_tables(cnxn: pyodbc.Connection) -> list[str]:


 def sanitize_column_name(name: str) -> str:
-    """Access column names are free-form ("NUM id", "A�O1", "TRUST NUM:");
-    Postgres staging columns need to be predictable identifiers. Original
-    name is preserved separately as metadata, this is only for the column
-    identifier itself."""
+    """Access column names are free-form ("NUM id", "AÑO1", "TRUST NUM:");
+    staging columns need predictable identifiers. The original name is
+    preserved separately as metadata — this is only for the column identifier
+    itself."""
    cleaned = re.sub(r"[^0-9a-zA-Z]+", "_", name).strip("_")
    cleaned = cleaned.lower()
    if not cleaned:
@@ -47,37 +95,59 @@ def sanitize_column_name(name: str) -> str:
    return cleaned


-def read_table(cnxn: pyodbc.Connection, table_name: str) -> pd.DataFrame:
-    cursor = cnxn.cursor()
-    cursor.execute(f"SELECT * FROM [{table_name}]")
-    original_columns = [d[0] for d in cursor.description]
+def read_table(cnxn, table_name: str) -> pd.DataFrame:
+    path = Path(cnxn)

-    # fetchall() aborts the whole table on the first bad row. Some legacy
-    # tables (confirmed: MULT) have Jet/ACE-level corruption — a record
-    # marked deleted at the storage level that the driver still enumerates
-    # but can't SQLGetData from ("Record is deleted", HY109). Fetch one row
-    # at a time so a corrupted row is skipped and logged instead of losing
-    # the entire table.
-    data = []
-    skipped = 0
-    while True:
-        try:
-            row = cursor.fetchone()
-        except pyodbc.Error as exc:
-            skipped += 1
-            print(f"    [skip row] {table_name}: {exc}")
-            continue
-        if row is None:
-            break
-        data.append(list(row))
-    if skipped:
-        print(f"    [{table_name}] skipped {skipped} corrupted row(s)")
+    # -b strip : drop LONGBINARY/OLE bytes so blobs never corrupt the CSV
+    #            (documents are extracted separately in migration step 4).
+    # -D / -T  : ISO date/datetime output.
+    # Default comma delimiter + double-quote quoting (doublequote escaping),
+    # which pandas.read_csv parses with its defaults.
+    result = subprocess.run(
+        [
+            "mdb-export",
+            "-b", "strip",
+            "-D", _DATE_FMT,
+            "-T", _DATETIME_FMT,
+            str(path),
+            table_name,
+        ],
+        capture_output=True,
+        text=True,
+        encoding="utf-8",
+        errors="replace",
+        check=True,
+    )

-    df = pd.DataFrame(data, columns=original_columns)
+    if result.stderr.strip():
+        # mdb-export prints warnings (e.g. about odd/deleted records) to
+        # stderr but still exits 0 — surface them without failing the table.
+        for line in result.stderr.splitlines():
+            if line.strip():
+                print(f"    [mdb-export warn] {table_name}: {line.strip()}")
+
+    csv_text = result.stdout
+    if not csv_text.strip():
+        # Empty table: mdb-export still emits a header row, but guard anyway.
+        return pd.DataFrame()
+
+    # Staging is a 1:1 audit copy — read everything as text (dtype=str) and
+    # treat only an empty field as null. keep_default_na=False is deliberate:
+    # real data can contain literal "NA"/"NULL"/"None" strings that pandas
+    # would otherwise silently turn into missing values.
+    df = pd.read_csv(
+        io.StringIO(csv_text),
+        dtype=str,
+        na_values=[""],
+        keep_default_na=False,
+        low_memory=False,
+    )
+
+    original_columns = list(df.columns)

    # Track original -> sanitized name mapping for the loader; dedupe any
-    # collisions that sanitization could introduce (e.g. "NUM id" and
-    # "NUM_id" both -> "num_id").
+    # collisions sanitization could introduce (e.g. "NUM id" and "NUM_id"
+    # both -> "num_id").
    seen: dict[str, int] = {}
    sanitized = []
    for col in original_columns:
@@ -1,6 +1,12 @@
-pyodbc>=5.0
+# Extraction on macOS uses mdbtools (brew install mdbtools) via subprocess,
+# not pyodbc — no Access ODBC driver exists on macOS. See extract.py.
 pandas>=2.2
 pyarrow>=15.0
 sqlalchemy>=2.0
 pymysql>=1.1
-pywin32>=306  # Windows only - needed for catalog_objects.py (DAO COM automation)
+
+# Windows-only, historical — the DAO/COM object catalog (catalog_objects.py)
+# was already run on Windows and its output is committed (objects.json). Not
+# needed on macOS; left documented for provenance.
+#   pyodbc>=5.0
+#   pywin32>=306