Initial scaffold: unified customer/insurance/utilities platform
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.
This commit is contained in:
@@ -0,0 +1,84 @@
|
||||
"""
|
||||
Source-database manifest for the staging load (migration plan step 1).
|
||||
|
||||
Exclusions here are deliberately conservative: only tables that are either
|
||||
(a) confirmed empty (0 rows — nothing is lost by skipping them) or (b) have
|
||||
rows but are structurally not customer/business data (mail-merge document
|
||||
templates, materialized Access query results) are excluded. Anything with
|
||||
real rows and an ambiguous purpose (e.g. PROPANO, datosfreak, pagos email)
|
||||
is loaded into staging anyway — the reconciliation pass decides what to do
|
||||
with it, per the migration plan's "don't guess the rule up front" principle.
|
||||
See C:\\Users\\ricar\\.claude\\plans\\logical-yawning-tome.md for the full
|
||||
rationale per table group.
|
||||
"""
|
||||
|
||||
from pathlib import Path
|
||||
|
||||
SOURCE_ROOT = Path(r"C:\Users\ricar\Downloads\Jorge")
|
||||
|
||||
SOURCES = {
|
||||
"utilities": {
|
||||
"path": SOURCE_ROOT / "UTILITIES.accdb",
|
||||
"schema": "stg_utilities",
|
||||
# Confirmed empty (0 rows) working/scratch tables.
|
||||
"exclude": {
|
||||
"BANCO EDITOR",
|
||||
"Errores de pegado",
|
||||
"TABLE1",
|
||||
"PARA BILLING SIN",
|
||||
"PARA BILLING1",
|
||||
"PARA BILLING2",
|
||||
"PARA EDO",
|
||||
"FALTANTES AGUA",
|
||||
"FALTANTES TEL",
|
||||
"LUZ TODOS",
|
||||
"TELEFONOS FECHAS",
|
||||
"TRUSTHFEE",
|
||||
"faltantes luz",
|
||||
"billing", # 0 rows; superseded by datos2/FEE ANUAL/fee15
|
||||
"TIT", # 1 row, default Access "Contacts" template shell — not real data
|
||||
},
|
||||
},
|
||||
"seguros": {
|
||||
# SEGUROS 16.mdb is an empty linked front-end; all data lives in _be.
|
||||
"path": SOURCE_ROOT / "SEGUROS 16_be.mdb",
|
||||
"schema": "stg_seguros",
|
||||
"exclude": {
|
||||
# Mail-merge document templates (letters/certificates), not data.
|
||||
"AMPL MENS",
|
||||
"AMPL R MENS",
|
||||
"IN MENS",
|
||||
"LIC MENS",
|
||||
"MCA2 MENS",
|
||||
"ME MENS",
|
||||
"MF MENS",
|
||||
"RC MENSAJE",
|
||||
"RC R MENS",
|
||||
# Materialized Access query results, not source-of-truth data.
|
||||
"TODOSJC",
|
||||
"TODOS",
|
||||
"vigenta casa y auto unicos",
|
||||
# Confirmed empty scratch tables.
|
||||
"ID TABLA",
|
||||
"ID TABLATLAS",
|
||||
"TABLA LIQUIDA MCA2",
|
||||
"TABLA LIQUIDA MF",
|
||||
"TABLA LIQUIDA RES",
|
||||
"TABLA LIQUIDA TUR",
|
||||
"TABLA LIQUIDA TUR ENDOSO",
|
||||
"TABLA AUTOS LIMIT R",
|
||||
"BORRA",
|
||||
"GENERICO_OLD",
|
||||
"TIT",
|
||||
},
|
||||
},
|
||||
"scothia": {
|
||||
"path": SOURCE_ROOT / "SCOTHIA.mdb",
|
||||
"schema": "stg_scothia",
|
||||
"exclude": {
|
||||
"INFORME",
|
||||
"INFORME BA",
|
||||
"FECHAIF", # date-range UI parameter table, not data
|
||||
},
|
||||
},
|
||||
}
|
||||
@@ -0,0 +1,94 @@
|
||||
"""
|
||||
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
|
||||
@@ -0,0 +1,120 @@
|
||||
"""
|
||||
Migration plan step 1: raw staging load.
|
||||
|
||||
Dumps every non-scratch table from all four source Access files 1:1 into
|
||||
either MySQL (one database per source system, per config.SOURCES — e.g.
|
||||
`stg_utilities`, `stg_seguros`, `stg_scothia`, matching a MySQL "schema" to
|
||||
a MySQL "database" 1:1) or, if --output-dir is given, to Parquet files —
|
||||
useful for environments (like this one) that have Access + pyodbc available
|
||||
but not a live MySQL instance reachable to load into yet.
|
||||
|
||||
Every staged table gets two extra columns: _legacy_source_table (the
|
||||
original Access table name) and _row_num (ordinal position in the source
|
||||
table) so rows are traceable even before any real key is identified.
|
||||
|
||||
Usage:
|
||||
python load_staging.py --output-dir ./output # Parquet, no DB needed
|
||||
python load_staging.py --database-url mysql+pymysql://user:pass@host/db # load into MySQL
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import sys
|
||||
from pathlib import Path
|
||||
|
||||
import pandas as pd
|
||||
|
||||
from config import SOURCES
|
||||
import extract
|
||||
|
||||
|
||||
def stage_source(source_name: str, source_cfg: dict, sink) -> None:
|
||||
path = source_cfg["path"]
|
||||
if not Path(path).exists():
|
||||
print(f" [skip] {source_name}: file not found at {path}", file=sys.stderr)
|
||||
return
|
||||
|
||||
print(f"=== {source_name} ({path}) ===")
|
||||
cnxn = extract.connect(path)
|
||||
tables = extract.list_tables(cnxn)
|
||||
excluded = source_cfg["exclude"]
|
||||
|
||||
for table_name in tables:
|
||||
if table_name in excluded:
|
||||
print(f" [exclude] {table_name}")
|
||||
continue
|
||||
|
||||
try:
|
||||
df = extract.read_table(cnxn, table_name)
|
||||
except Exception as exc: # noqa: BLE001 - report and keep going
|
||||
print(f" [ERROR] {table_name}: {exc}", file=sys.stderr)
|
||||
continue
|
||||
|
||||
df.insert(0, "_legacy_source_table", table_name)
|
||||
df.insert(1, "_row_num", range(len(df)))
|
||||
|
||||
sink(source_cfg["schema"], table_name, df)
|
||||
print(f" [ok] {table_name}: {len(df)} rows, {len(df.columns) - 2} columns")
|
||||
|
||||
|
||||
def make_parquet_sink(output_dir: Path):
|
||||
def sink(schema: str, table_name: str, df: pd.DataFrame) -> None:
|
||||
target_dir = output_dir / schema
|
||||
target_dir.mkdir(parents=True, exist_ok=True)
|
||||
safe_name = extract.sanitize_column_name(table_name)
|
||||
df.to_parquet(target_dir / f"{safe_name}.parquet", index=False)
|
||||
|
||||
return sink
|
||||
|
||||
|
||||
def make_mysql_sink(database_url: str):
|
||||
from sqlalchemy import create_engine, text
|
||||
|
||||
# A MySQL "schema" is a database - CREATE SCHEMA is a synonym for
|
||||
# CREATE DATABASE. Connect without a default database in the URL so
|
||||
# each staging area (stg_utilities/stg_seguros/stg_scothia) can be
|
||||
# created and written to independently by one engine.
|
||||
engine = create_engine(database_url)
|
||||
created_schemas: set[str] = set()
|
||||
|
||||
def sink(schema: str, table_name: str, df: pd.DataFrame) -> None:
|
||||
if schema not in created_schemas:
|
||||
with engine.begin() as conn:
|
||||
conn.execute(text(f"CREATE SCHEMA IF NOT EXISTS `{schema}`"))
|
||||
created_schemas.add(schema)
|
||||
|
||||
safe_name = extract.sanitize_column_name(table_name)
|
||||
df.to_sql(
|
||||
safe_name,
|
||||
engine,
|
||||
schema=schema,
|
||||
if_exists="replace",
|
||||
index=False,
|
||||
)
|
||||
|
||||
return sink
|
||||
|
||||
|
||||
def main() -> None:
|
||||
parser = argparse.ArgumentParser(description=__doc__)
|
||||
parser.add_argument("--output-dir", type=Path, help="Write Parquet files here instead of a database")
|
||||
parser.add_argument("--database-url", type=str, help="MySQL connection string, e.g. mysql+pymysql://user:pass@host:3306/ (sqlalchemy format)")
|
||||
parser.add_argument("--only", type=str, help="Comma-separated subset of source names to run (utilities,seguros,scothia)")
|
||||
args = parser.parse_args()
|
||||
|
||||
if not args.output_dir and not args.database_url:
|
||||
parser.error("one of --output-dir or --database-url is required")
|
||||
|
||||
sink = make_parquet_sink(args.output_dir) if args.output_dir else make_mysql_sink(args.database_url)
|
||||
|
||||
only = set(args.only.split(",")) if args.only else None
|
||||
|
||||
for source_name, source_cfg in SOURCES.items():
|
||||
if only and source_name not in only:
|
||||
continue
|
||||
stage_source(source_name, source_cfg, sink)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
@@ -0,0 +1,5 @@
|
||||
pyodbc>=5.0
|
||||
pandas>=2.2
|
||||
pyarrow>=15.0
|
||||
sqlalchemy>=2.0
|
||||
pymysql>=1.1
|
||||
Reference in New Issue
Block a user