jorgecuadros-platform/migration/extract.py

"""
Reads tables out of an Access database via mdbtools.

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 shutil
import subprocess
from pathlib import Path

import pandas as pd

# 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 _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 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 name in result.stdout.splitlines():
        name = name.strip()
        if not name:
            continue
        if name.startswith("MSys") or name.startswith("~"):
            continue
        tables.append(name)
    return sorted(tables)


def sanitize_column_name(name: str) -> str:
    """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:
        cleaned = "col"
    if cleaned[0].isdigit():
        cleaned = f"c_{cleaned}"
    return cleaned


def read_table(cnxn, table_name: str) -> pd.DataFrame:
    path = Path(cnxn)

    # -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,
    )

    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 sanitization could introduce (e.g. "NUM id" and "NUM_id"
    # both -> "num_id").
    seen: dict[str, int] = {}
    sanitized = []
    for col in original_columns:
        base = sanitize_column_name(col)
        if base in seen:
            seen[base] += 1
            base = f"{base}_{seen[base]}"
        else:
            seen[base] = 0
        sanitized.append(base)
    df.columns = sanitized
    df.attrs["original_columns"] = original_columns

    return df