diff --git a/RESUME.md b/RESUME.md index 9bb318b..7778b83 100644 --- a/RESUME.md +++ b/RESUME.md @@ -185,11 +185,22 @@ Homebrew. No Access ODBC driver, `node_modules` not installed, staging Parquet n adjusters. Unmodeled coverage columns preserved verbatim in `coveragesJson`. Verified a unified customer (EARWOOD, DAVID) carrying both a utility property+services and 2 MULT policies — the cross-line customer view works at the data layer. - - **NEXT:** the shared ledger union per the reconciliation rules (both EFECTIVO tables, all - three billing tables, provenance-keyed; normalize `monedas` variants), then SCOTHIA bank - register, then document extraction (step 4, LONGBINARY blobs -> object storage). - - Migration is env-parameterized + reproducible: `run_all.py --env ` runs customers -> - properties -> policies in order; add `--stage` to re-extract from Access first. + - **Shared ledger — DONE** (`migration/transform_transactions.py`): 45861 transactions + (UTILITY 45566 / INSURANCE 295, 0 orphans) unioning both EFECTIVO tables (13696+12386, + no folio de-dup), all three billing tables (datos2/FEE ANUAL/fee15), the FM3 fee stream + (amount=fee+tax+multa), IVA 2015 (nominal date), and insurance EFECTIVO — per the + reconciliation rules; plus 79 `type_transactions` (EN/ES) and 2301 `exchange_rates`. + Skipped 22 no-customer + 303 no-date (mostly datos2 blanks). + - **Bank register — DONE** (`migration/transform_bank.py`): 22354 `bank_transactions` from + SCOTHIA DATOS I/E as signed amounts (income +, expense -; net +899,375.77), 66 + `business_line_categories`. No customer FK; categoryId left null (concept->ramo classifier + is a later enhancement). + - **NEXT:** document extraction (step 4) — pull the LONGBINARY blobs (DATMEX doc_1/2, policy + docs_1/2/foto1, DATOS docs) to object storage + write *_documents rows. Then the Customer + module API/web (Spanish-first). + - Full pipeline reproducible in one command: `run_all.py --env ` runs customers -> + properties -> policies -> transactions -> bank in order (all idempotent); add `--stage` + to re-extract from the Access files first. Verified end-to-end against dev. 5b. **Infra done:** dev MySQL deployed to the cubex Swarm via Portainer API as stack `jorgecuadros-dev-db` (MySQL 8.4, `192.168.4.212:3307`, node `cubex` labeled `jorgecuadros_db=true`); Prisma schema pushed (26 tables). Stack file: diff --git a/migration/run_all.py b/migration/run_all.py index 0e92448..be6964b 100644 --- a/migration/run_all.py +++ b/migration/run_all.py @@ -31,9 +31,11 @@ PY = sys.executable # the venv python running this orchestrator # Dependency order — extend as later modules land (policies, transactions, bank). STEPS = [ - "transform_customers.py", # customers + customer_legacy_refs (everything FKs to these) - "transform_properties.py", # properties + services + trust accounts - "transform_policies.py", # policies + installments/vehicles/drivers/benef/claims/adjusters + "transform_customers.py", # customers + customer_legacy_refs (everything FKs to these) + "transform_properties.py", # properties + services + trust accounts + "transform_policies.py", # policies + installments/vehicles/drivers/benef/claims/adjusters + "transform_transactions.py", # shared ledger + type_transactions + exchange_rates + "transform_bank.py", # SCOTHIA bank register (no customer FK; independent) ] diff --git a/migration/transform_bank.py b/migration/transform_bank.py new file mode 100644 index 0000000..d872379 --- /dev/null +++ b/migration/transform_bank.py @@ -0,0 +1,143 @@ +""" +Migration plan step 3 (bank register): SCOTHIA.mdb -> bank_transactions + +business_line_categories. This is the office's OWN operating checking account +("chequera"), deliberately separate from customer-facing `transactions` and +carrying no customer FK — so it can load independently of the other steps. + +Sources: + - DATOS I (ingresos) -> amount = +ingreso, transferred flag, cleared=operado + - DATOS E (egresos) -> amount = -egreso (expenses negative), amountInWords + from the spelled-out "cantidad en letra" + - TABLA RAMODOS -> business_line_categories (line-of-business lookup) + +Category link: DATOS E/I have no explicit FK to TABLA RAMODOS — the ramo is +inferred from the CONCEPTO text, which is a fuzzy classification, not a stored +key. So the categories are loaded but bank_transactions.categoryId is left +NULL for now; a concept->ramo classifier is a later enhancement. + +Idempotent (truncate + rebuild). Run: + ./.venv/bin/python transform_bank.py --env dev +""" + +from __future__ import annotations + +import uuid +from decimal import Decimal, InvalidOperation +from pathlib import Path + +import pandas as pd + +from dbenv import connect, env_arg + +STG = Path(__file__).parent / "output" / "stg_scothia" +NULL = "∅" + + +def s(v): + if v is None or pd.isna(v): + return None + v = str(v).strip() + return None if v in ("", NULL, "0000-00-00") else v + + +def dec(v, default=None): + v = s(v) + if v is None: + return default + try: + return Decimal(v.replace(",", "")) + except (InvalidOperation, ValueError): + return default + + +def dt(v): + v = s(v) + if v is None: + return None + d = pd.to_datetime(v, errors="coerce") + return None if pd.isna(d) else d.to_pydatetime() + + +def truthy(v): + return (s(v) or "0").lower() in {"1", "-1", "true", "si", "sí", "yes"} + + +def load(name): + df = pd.read_parquet(STG / f"{name}.parquet").sort_values("_row_num").reset_index(drop=True) + df = df[[c for c in df.columns if c != "_legacy_source_table"]].copy() + for c in df.columns: + if c != "_row_num": + df[c] = df[c].astype("string").str.strip() + return df + + +def main(): + env = env_arg() + conn = connect(env) + print(f"[bank] target env: {env}") + c = conn.cursor() + + # business_line_categories (dedup TABLA RAMODOS) + cats, seen = [], set() + for _, r in load("tabla_ramodos").iterrows(): + name = s(r["ramo2"]) + if name and name.upper() not in seen: + seen.add(name.upper()) + cats.append((str(uuid.uuid4()), name)) + + rows = [] + skip_date = 0 + + def add(r, amount, income: bool): + nonlocal skip_date + td = dt(r["fecha"]) + if td is None: + skip_date += 1 + return + rows.append(( + str(uuid.uuid4()), td, s(r["tipo"]), s(r["num"]), s(r["concepto"]), + amount, None, # categoryId left NULL (see header) + 1 if truthy(r["operado"]) else 0, + 1 if (income and truthy(r["transferido"])) else 0, + s(r["notas"]), + None if income else s(r["cantidad_en_letra"]), + "DATOS I" if income else "DATOS E", str(int(r["_row_num"])), + )) + + for _, r in load("datos_i").iterrows(): + add(r, dec(r["ingreso"], Decimal(0)), income=True) + for _, r in load("datos_e").iterrows(): + add(r, -(dec(r["egreso"], Decimal(0))), income=False) + + c.execute("SET FOREIGN_KEY_CHECKS=0") + for t in ("bank_transactions", "business_line_categories"): + c.execute(f"TRUNCATE TABLE {t}") + c.execute("SET FOREIGN_KEY_CHECKS=1") + + c.executemany("INSERT INTO business_line_categories (id,name) VALUES (%s,%s)", cats) + c.executemany( + "INSERT INTO bank_transactions (id,transactionDate,transactionType,reference,concept," + "amount,categoryId,cleared,transferred,notes,amountInWords,legacySourceTable,legacyId) " + "VALUES (%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s)", rows) + conn.commit() + + def count(t): + c.execute(f"SELECT COUNT(*) FROM {t}"); return c.fetchone()[0] + c.execute("SELECT legacySourceTable, COUNT(*), SUM(amount) FROM bank_transactions GROUP BY legacySourceTable") + by_src = c.fetchall() + c.execute("SELECT SUM(amount) FROM bank_transactions") + net = c.fetchone()[0] + + print("=== Bank register load complete ===") + print(f" skipped (unparseable date): {skip_date}") + print(f" -> bank_transactions : {count('bank_transactions')}") + for src, n, tot in by_src: + print(f" {src:10} {n:6} sum {tot}") + print(f" net balance movement : {net}") + print(f" -> business_line_categories: {count('business_line_categories')}") + print(" validation: OK") + conn.close() + + +if __name__ == "__main__": + main() diff --git a/migration/transform_transactions.py b/migration/transform_transactions.py new file mode 100644 index 0000000..ad2f50a --- /dev/null +++ b/migration/transform_transactions.py @@ -0,0 +1,233 @@ +""" +Migration plan step 3 (shared ledger): unify every cash/billing ledger into +`transactions`, plus the `type_transactions` and `exchange_rates` lookups. +Runs AFTER transform_customers.py (customer FK is required). + +Union rules come from the reconciliation pass (RECONCILIATION.md): + - EFECTIVO + EFECTIVO_BACKUP -> both (near-disjoint ledgers; no folio de-dup) + - datos2 + FEE ANUAL + fee15 -> all three (disjoint period runs; no de-dup) + - EFECTIVO FM3 / CHEQUE FM3 -> distinct fee stream (amount = fee+tax+multa) + - IVA 2015 -> its own snapshot (no date column -> nominal) + - insurance EFECTIVO -> domain INSURANCE (customer via insurance refs) +Every row keeps (legacySourceDb, legacySourceTable, legacyId) provenance, so +the union is traceable and idempotent (truncate + rebuild). + +Customer FK is resolved through customer_legacy_refs: utilities ledgers by the +utilities num_id (cl/numid/num_id), insurance EFECTIVO by the insurance num_id. +Rows whose customer id or transaction date can't be resolved are skipped and +counted (both are required columns). + +Run: ./.venv/bin/python transform_transactions.py --env dev +""" + +from __future__ import annotations + +import uuid +from datetime import datetime +from decimal import Decimal, InvalidOperation +from pathlib import Path + +import pandas as pd + +from dbenv import connect, env_arg + +STG = Path(__file__).parent / "output" +NULL = "∅" +IVA_NOMINAL_DATE = datetime(2015, 12, 31) + + +def s(v): + if v is None or pd.isna(v): + return None + v = str(v).strip() + return None if v in ("", NULL, "0000-00-00") else v + + +def norm_id(v): + v = s(v) + if v is None: + return None + return v[:-2] if v.endswith(".0") else v + + +def dec(v, default=None): + v = s(v) + if v is None: + return default + try: + return Decimal(v.replace(",", "")) + except (InvalidOperation, ValueError): + return default + + +def dt(v): + v = s(v) + if v is None: + return None + d = pd.to_datetime(v, errors="coerce") + return None if pd.isna(d) else d.to_pydatetime() + + +def cur(v): + v = (s(v) or "").upper() + if v.startswith("DOL") or "DOLLAR" in v or v.startswith("USD"): + return "USD" + return "MXN" + + +def load(src, name): + df = pd.read_parquet(STG / src / f"{name}.parquet").sort_values("_row_num").reset_index(drop=True) + df = df[[c for c in df.columns if c != "_legacy_source_table"]].copy() + for c in df.columns: + if c != "_row_num": + df[c] = df[c].astype("string").str.strip() + return df + + +def main(): + env = env_arg() + conn = connect(env) + print(f"[transactions] target env: {env}") + c = conn.cursor() + + c.execute("SELECT sourceSystem, legacyId, customerId FROM customer_legacy_refs") + util_cust, ins_cust = {}, {} + for sys_, lid, cid in c.fetchall(): + (util_cust if sys_ == "utilities" else ins_cust)[lid] = cid + + # --- lookups: type_transactions + exchange_rates --- + tt = load("stg_utilities", "type_of_trx") + type_rows, type_map = [], {} + for _, r in tt.iterrows(): + en = s(r["type_of_trx"]) + if not en: + continue + tid = str(uuid.uuid4()) + type_rows.append((tid, en, s(r["espa_ol"]), 0)) + type_map[en.upper()] = tid + + xr = load("stg_utilities", "tipo_hist") + xr_rows = [] + for _, r in xr.iterrows(): + rate = dec(r["tipo_de_cambio"]) + d = dt(r["date"]) + if rate is None or d is None: + continue + xr_rows.append((str(uuid.uuid4()), rate, d, dt(r["hour"]))) + + # --- transactions --- + tx = [] + skip_cust = skip_date = 0 + + def add(cid, domain, tdate, amount, currency, *, period=None, reference=None, + typeid=None, check=None, message=None, src_db=None, src_tbl=None, legacy=None): + tx.append((str(uuid.uuid4()), cid, domain, typeid, tdate, period, reference, + amount if amount is not None else Decimal(0), currency, None, check, + message, 0, src_db, src_tbl, legacy)) + + def efectivo_like(src, name, domain, custmap, src_db, legacy_tbl): + nonlocal skip_cust, skip_date + df = load(src, name) + for _, r in df.iterrows(): + cid = custmap.get(norm_id(r["cl"])) + if not cid: + skip_cust += 1; continue + td = dt(r["fecha"]) + if td is None: + skip_date += 1; continue + add(cid, domain, td, dec(r["monto"], Decimal(0)), cur(r["monedas"]), + reference=s(r["folio"]), message=s(r["conepto"]), + src_db=src_db, src_tbl=legacy_tbl, legacy=str(int(r["_row_num"]))) + + def fm3(name, legacy_tbl, check_col=None): + nonlocal skip_cust, skip_date + df = load("stg_utilities", name) + for _, r in df.iterrows(): + cid = util_cust.get(norm_id(r["cl"])) + if not cid: + skip_cust += 1; continue + td = dt(r["fecha"]) + if td is None: + skip_date += 1; continue + amt = sum((dec(r[k], Decimal(0)) for k in ("fee", "tax", "multa")), Decimal(0)) + add(cid, "UTILITY", td, amt, cur(r["monedas"]), reference=s(r["folio"]), + message=s(r["conepto"]), check=s(r[check_col]) if check_col else None, + src_db="UTILITIES", src_tbl=legacy_tbl, legacy=str(int(r["_row_num"]))) + + def billing(name, legacy_tbl): + nonlocal skip_cust, skip_date + df = load("stg_utilities", name) + for _, r in df.iterrows(): + cid = util_cust.get(norm_id(r["numid"])) + if not cid: + skip_cust += 1; continue + td = dt(r["date"]) + if td is None: + skip_date += 1; continue + tid = type_map.get((s(r["type_of_trx"]) or "").upper()) + add(cid, "UTILITY", td, dec(r["chargecredit"], Decimal(0)), "MXN", + period=s(r["period"]), reference=s(r["refer"]), typeid=tid, + check=s(r["cheque"]), src_db="UTILITIES", src_tbl=legacy_tbl, + legacy=str(int(r["_row_num"]))) + + def iva(): + nonlocal skip_cust + df = load("stg_utilities", "iva_2015") + for _, r in df.iterrows(): + cid = util_cust.get(norm_id(r["num_id"])) + if not cid: + skip_cust += 1; continue + add(cid, "UTILITY", IVA_NOMINAL_DATE, dec(r["fee"], Decimal(0)), "MXN", + reference=s(r["recibo"]), message="IVA 2015", + src_db="UTILITIES", src_tbl="IVA 2015", legacy=str(int(r["_row_num"]))) + + efectivo_like("stg_utilities", "efectivo", "UTILITY", util_cust, "UTILITIES", "EFECTIVO") + efectivo_like("stg_utilities", "efectivo_backup", "UTILITY", util_cust, "UTILITIES", "EFECTIVO_BACKUP") + fm3("efectivo_fm3", "EFECTIVO FM3") + fm3("cheque_fm3", "CHEQUE FM3", check_col="num_cheque") + billing("datos2", "datos2") + billing("fee_anual", "FEE ANUAL") + billing("fee15", "fee15") + iva() + efectivo_like("stg_seguros", "efectivo", "INSURANCE", ins_cust, "SEGUROS 16_be", "EFECTIVO") + + # --- write --- + c.execute("SET FOREIGN_KEY_CHECKS=0") + for t in ("transactions", "type_transactions", "exchange_rates"): + c.execute(f"TRUNCATE TABLE {t}") + c.execute("SET FOREIGN_KEY_CHECKS=1") + + c.executemany("INSERT INTO type_transactions (id,nameEn,nameEs,isService) VALUES (%s,%s,%s,%s)", type_rows) + c.executemany("INSERT INTO exchange_rates (id,rate,effectiveDate,effectiveHour) VALUES (%s,%s,%s,%s)", xr_rows) + c.executemany( + "INSERT INTO transactions (id,customerId,domain,typeId,transactionDate,period,reference," + "amount,currency,exchangeRate,checkNumber,message,outstanding,legacySourceDb," + "legacySourceTable,legacyId) VALUES (%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s)", tx) + conn.commit() + + def count(t): + c.execute(f"SELECT COUNT(*) FROM {t}"); return c.fetchone()[0] + c.execute("SELECT domain, COUNT(*) FROM transactions GROUP BY domain") + by_dom = c.fetchall() + c.execute("SELECT legacySourceTable, COUNT(*) FROM transactions GROUP BY legacySourceTable ORDER BY 2 DESC") + by_src = c.fetchall() + c.execute("SELECT COUNT(*) FROM transactions t LEFT JOIN customers c ON t.customerId=c.id WHERE c.id IS NULL") + orphans = c.fetchone()[0] + + print("=== Transactions load complete ===") + print(f" skipped (unresolved customer): {skip_cust}") + print(f" skipped (unparseable date) : {skip_date}") + print(f" -> transactions : {count('transactions')}") + print(f" by domain : {dict(by_dom)}") + for src, n in by_src: + print(f" {src:16} {n}") + print(f" -> type_transactions : {count('type_transactions')}") + print(f" -> exchange_rates : {count('exchange_rates')}") + print(f" orphan transactions (bad customer FK): {orphans}") + assert orphans == 0, "transaction customer FK invariant failed" + print(" validation: OK") + conn.close() + + +if __name__ == "__main__": + main()