rmancinas
27118f0df2
Next.js + NestJS + Prisma (MySQL) monorepo replacing the legacy PHP internal app. Includes a session-based auth module with Argon2 password hashing and global input validation (replacing the old app's SQL injection and plaintext password comparison), the full target Prisma schema for customers/insurance/utilities/shared ledger/bank register, Docker Compose + Dockerfiles, and an Access-to-staging migration pipeline (migration/) already run against the real source databases. See PLAN.md and RESUME.md for the full architecture and session history.
95 lines
3.1 KiB
Python
95 lines
3.1 KiB
Python
"""
|
|
Reads tables out of an Access database via pyodbc.
|
|
|
|
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.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import re
|
|
import pyodbc
|
|
import pandas as pd
|
|
|
|
ACCESS_DRIVER = "Microsoft Access Driver (*.mdb, *.accdb)"
|
|
|
|
|
|
def connect(path) -> pyodbc.Connection:
|
|
conn_str = f"DRIVER={{{ACCESS_DRIVER}}};DBQ={path};"
|
|
return pyodbc.connect(conn_str, autocommit=True)
|
|
|
|
|
|
def list_tables(cnxn: pyodbc.Connection) -> list[str]:
|
|
cursor = cnxn.cursor()
|
|
tables = []
|
|
for row in cursor.tables(tableType="TABLE"):
|
|
name = row.table_name
|
|
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:");
|
|
Postgres staging columns need to be predictable identifiers. 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: 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]
|
|
|
|
# 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)")
|
|
|
|
df = pd.DataFrame(data, columns=original_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").
|
|
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
|