rmancinasandClaude Opus 5
29ae9fa5bc
Legacy ran a year-end corte: it summed the closing year, wrote that total back
as each customer's Jan-1 BALANCE FORWARD, and started the next year clean. The
platform inherited those opening rows but never the years behind them — only
the current year's charges were ever staged, so every prior year held receipts
and no bills. Rendering one would have shown a customer credits with nothing
owed against them, which is worse than showing nothing.
The archives are whole-database Access snapshots named for the period they
hold, so `2025.accdb` is discovered by filename, staged, and loaded through the
existing DATOS2 branch — a snapshot's `datos2` is the identical shape, one year
older. Only the ledger and DATGRAL come out of a snapshot; everything else in
it is a year-stale copy of a live table. The cash side is deliberately left
behind: EFECTIVO is a lifetime journal, so the snapshot's copy is a subset of
the live one and importing it would double-book every prior-year receipt.
Period travels with the row, in legacySourceTable as `datos2@2025`. That tag,
not the date, is what a year view should filter on: the archives are not
cleanly bounded (2025 carries ten undated rows and two dated into 2026) and
legacy never filtered by date either — its reader is `SELECT ... FROM \`2025\``.
The tag also keeps legacyId safe, since it is a positional ordinal that
restarts at 0 in every archive and would otherwise collide row-for-row.
Two guards, because attaching a prior year by NUMid is the one thing here that
can go quietly wrong:
- Reissued numbers are skipped, not imported. Comparing each archive's
DATGRAL against the live one, 13 names moved since 2025 and 40 since 2024;
most are the same customer re-described, but a few are a different
household holding a recycled number, and filing their ledger under the new
owner would show a stranger's charges. Sharing any word of three or more
characters separates a rename from a reissue. Names are compared
legacy-to-legacy: `customers.name` has been through blank-name recovery,
and comparing to it reported 121 drifts where there are 13.
- Every period is checked against the corte identity it must satisfy —
SUM(year N) == BALANCE FORWARD(N+1) — and the result is reported per year.
A truncated export, a file dropped under the wrong year, or a botched
customer match all fail loudly here. 2025 reconciles 1,159/1,167 (99.3%)
and 2024 1,144/1,156 (99.0%); the recycle guard raised 2024 from 98.2%.
Balances are untouched: BALANCE_FLOOR_JOIN floors on the newest BALANCE FORWARD
per customer, so rows behind it are already excluded from every balance read.
Uploads go through the existing ingest endpoint, allowlisted by an anchored
`AAAA.accdb` pattern that also keeps a caller-supplied name inside the ingest
directory. The Operaciones page grows an entry point for an archive that has no
row yet, reading the period off the chosen file's own name.
Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
498 lines
22 KiB
Python
498 lines
22 KiB
Python
"""
|
|
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 -> EFECTIVO in full, plus only the BACKUP rows
|
|
whose business key is absent from EFECTIVO
|
|
(BACKUP is a stale copy: 12386 of its 12387
|
|
rows are verbatim duplicates). De-dup on the
|
|
business key, never on folio (folio collides
|
|
across the two tables).
|
|
- 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
|
|
from sync import parse_mode
|
|
|
|
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 verify_corte(c, year: int) -> None:
|
|
"""Assert legacy's corte identity: SUM(period Y) == BALANCE FORWARD(Y+1).
|
|
|
|
This is the whole reason a year can be shown on its own. Legacy closed each
|
|
year by summing it and writing that total back as every customer's Jan-1
|
|
opening row for the next one, so if an archive is the right file, complete,
|
|
and attached to the right customers, its per-customer total lands exactly on
|
|
the next year's BALANCE FORWARD. A truncated export, a file dropped under
|
|
the wrong year, or a botched customer match all break the identity loudly
|
|
here instead of quietly six months from now.
|
|
|
|
Reported, never fatal. Legacy publishes on its own schedule, so a handful of
|
|
customers legitimately drift between the snapshot and the cut — the run that
|
|
established this reconciled 1,160 of 1,170.
|
|
"""
|
|
c.execute(
|
|
"""
|
|
SELECT sums.customerId, sums.total, bf.amount
|
|
FROM (
|
|
SELECT customerId, ROUND(SUM(amount), 2) AS total
|
|
FROM transactions
|
|
WHERE legacySourceTable = %s AND voidedAt IS NULL
|
|
GROUP BY customerId
|
|
) sums
|
|
LEFT JOIN (
|
|
SELECT t.customerId, ROUND(SUM(t.amount), 2) AS amount
|
|
FROM transactions t
|
|
JOIN type_transactions tt ON tt.id = t.typeId
|
|
WHERE tt.nameEn = 'BALANCE FORWARD' AND t.voidedAt IS NULL
|
|
AND t.transactionDate >= %s AND t.transactionDate < %s
|
|
GROUP BY t.customerId
|
|
) bf ON bf.customerId = sums.customerId
|
|
""",
|
|
(f"datos2@{year}", f"{year + 1}-01-01", f"{year + 1}-01-02"),
|
|
)
|
|
rows = c.fetchall()
|
|
matched = mismatched = 0
|
|
missing = 0
|
|
drift = Decimal(0)
|
|
for _cid, total, amount in rows:
|
|
if amount is None:
|
|
missing += 1
|
|
continue
|
|
if abs(Decimal(str(total)) - Decimal(str(amount))) < Decimal("0.02"):
|
|
matched += 1
|
|
else:
|
|
mismatched += 1
|
|
drift += abs(Decimal(str(total)) - Decimal(str(amount)))
|
|
checked = matched + mismatched
|
|
pct = (100 * matched / checked) if checked else 0
|
|
print(
|
|
f" corte {year} -> BF {year + 1}: {matched}/{checked} match ({pct:.1f}%)"
|
|
f", {mismatched} off by {drift:,.2f}, {missing} with no BF row"
|
|
)
|
|
|
|
|
|
def main():
|
|
env, sync_mode = parse_mode()
|
|
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
|
|
|
|
def type_id_for(raw) -> str | None:
|
|
"""Resolve a transaction type, minting one when the lookup lacks it.
|
|
|
|
The Access `TYPE OF TRX` table is a stale pick-list, not a constraint —
|
|
staff free-text straight into DATOS2, so 78 values covering 3,939 rows
|
|
(BALANCE FORWARD 1,188, ANNUAL FEE 1,116, IZZI 367, ...) appear in the
|
|
ledger but not the lookup. Leaving those unmapped stored typeId NULL and
|
|
lost the label outright: nothing else on `transactions` carries the type
|
|
text, so the row rendered blank and was unrecoverable after migration.
|
|
Minting from the literal keeps the display string; nameEs stays NULL
|
|
because only the lookup has translations.
|
|
"""
|
|
en = s(raw)
|
|
if not en:
|
|
return None
|
|
key = en.upper()
|
|
tid = type_map.get(key)
|
|
if tid is None:
|
|
tid = str(uuid.uuid4())
|
|
type_rows.append((tid, en, None, 0))
|
|
type_map[key] = tid
|
|
return 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 = skip_dupe = 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,
|
|
outstanding=0):
|
|
tx.append((str(uuid.uuid4()), cid, domain, typeid, tdate, period, reference,
|
|
amount if amount is not None else Decimal(0), currency, None, check,
|
|
message, outstanding, src_db, src_tbl, legacy))
|
|
|
|
# Business key of a real cash payment. `folio` is deliberately excluded: it
|
|
# is a per-table sequential number that collides between EFECTIVO and
|
|
# EFECTIVO_BACKUP (12363 shared numbers, 12204 of them on different
|
|
# payments), so it identifies nothing across tables.
|
|
def biz_key(r):
|
|
return (
|
|
norm_id(r["cl"]),
|
|
s(r["fecha"]),
|
|
dec(r["monto"], Decimal(0)),
|
|
s(r["conepto"]),
|
|
)
|
|
|
|
def efectivo_like(src, name, domain, custmap, src_db, legacy_tbl, *, seen=None,
|
|
type_label=None):
|
|
"""Load an EFECTIVO-shaped cash ledger.
|
|
|
|
`seen` (a set) makes the load de-duplicating: keys are added to it as
|
|
rows load, and a row whose key is already present is skipped. That is
|
|
how EFECTIVO_BACKUP contributes only its genuinely-new rows.
|
|
|
|
`type_label` names the transaction type for every row. These tables have
|
|
no type column at all — in Access the type is implied by which table the
|
|
row lives in — so unlike DATOS2 there is no string to map and typeId came
|
|
out NULL for all of them.
|
|
|
|
That is not merely a blank label. handleGetAccountDetails in
|
|
my.jorgecuadros.com identifies payments by matching TYPEOFTRX against
|
|
('PAYMENT THANK YOU', 'PAYPAL', 'CASH DEPOSIT', 'CHECK DEPOSIT') to reset
|
|
the running balance in mode=current; an unlabelled payment is not
|
|
recognised and the balance silently diverges from legacy.
|
|
"""
|
|
nonlocal skip_cust, skip_date, skip_dupe
|
|
df = load(src, name)
|
|
for _, r in df.iterrows():
|
|
if seen is not None:
|
|
key = biz_key(r)
|
|
if key in seen:
|
|
skip_dupe += 1; continue
|
|
seen.add(key)
|
|
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"]),
|
|
typeid=type_id_for(type_label),
|
|
src_db=src_db, src_tbl=legacy_tbl, legacy=str(int(r["_row_num"])))
|
|
|
|
def fm3(name, legacy_tbl, check_col=None):
|
|
"""FM3 fee streams. Deliberately left unlabelled, unlike EFECTIVO.
|
|
|
|
These rows (EFECTIVO FM3 627, CHEQUE FM3 157) also have no type column,
|
|
but every one of them predates the two periods the site exposes — it
|
|
allowlists only the current year and the prior year — so none can be
|
|
matched against a legacy label, and none can reach a customer. Inventing
|
|
a plausible name like "CHECK DEPOSIT" would feed the payment-detection
|
|
list in handleGetAccountDetails on nothing but a guess. Leave them NULL
|
|
until a real mapping is available.
|
|
"""
|
|
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, *, src="stg_utilities", skip_numids=None):
|
|
"""Load a DATOS2-shaped billing ledger.
|
|
|
|
`src` names the staging schema, so a prior-period archive
|
|
(stg_period_2025) loads through this same path: the snapshot's `datos2`
|
|
is the identical eleven-column shape, one year older.
|
|
|
|
NOPAGO is the legacy "still owed" flag. The website reads it directly —
|
|
`account.statement.php` splits the statement on `NOPAGO = 0` vs
|
|
`NOPAGO = 1` and renders the latter as the "Outstanding Bills Requiring
|
|
Attention" table — so dropping it does not merely lose a column, it
|
|
silently empties that whole section for anyone served off the platform.
|
|
Only these three tables carry it (76 rows set in DATOS2 today); the
|
|
EFECTIVO/FM3 cash streams have no such column and stay 0.
|
|
"""
|
|
nonlocal skip_cust, skip_date, skip_recycled
|
|
df = load(src, name)
|
|
for _, r in df.iterrows():
|
|
numid = norm_id(r["numid"])
|
|
if skip_numids and numid in skip_numids:
|
|
skip_recycled += 1; continue
|
|
cid = util_cust.get(numid)
|
|
if not cid:
|
|
skip_cust += 1; continue
|
|
td = dt(r["date"])
|
|
if td is None:
|
|
skip_date += 1; continue
|
|
tid = type_id_for(r["type_of_trx"])
|
|
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"])),
|
|
outstanding=1 if s(r["nopago"]) == "1" else 0)
|
|
|
|
skip_recycled = 0
|
|
recycle_report: list[tuple[int, str, str, str]] = []
|
|
|
|
def period_numid_guard(year: int) -> set[str]:
|
|
"""NUMids whose prior-period owner is not today's customer.
|
|
|
|
Prior-period rows attach by NUMid and nothing else, so a number the
|
|
office retired and reissued would file one customer's ledger under
|
|
another's name — the one error this feature must never make, because it
|
|
shows a stranger's charges to whoever holds the number now.
|
|
|
|
Reuse is real but rare: comparing each archive's DATGRAL against the
|
|
live one, 13 names moved since 2025 and 40 since 2024. Most are the same
|
|
customer re-described — a typo fixed (VIKIE -> VICKIE), a spouse added
|
|
or dropped (STEWART, ALAN R. -> STEWART, ALAN & JENNIFER). A few are
|
|
genuinely a different household (STRONKS, BOB -> SWEET, DONALD E.).
|
|
|
|
Sharing any word of three or more characters separates the two cleanly:
|
|
a rename keeps the surname, a reissue keeps nothing. Names are compared
|
|
legacy-to-legacy, archive DATGRAL against live DATGRAL, deliberately not
|
|
against `customers.name` — that column has been through the blank-name
|
|
recovery pass, and comparing to it reported 121 drifts where there are
|
|
13, every extra one a false positive that would have discarded good
|
|
history.
|
|
"""
|
|
try:
|
|
arch = load(f"stg_period_{year}", "datgral")
|
|
live = load("stg_utilities", "datgral")
|
|
except (FileNotFoundError, OSError):
|
|
return set()
|
|
|
|
def toks(v) -> set[str]:
|
|
return {w for w in "".join(ch if ch.isalnum() else " " for ch in (s(v) or "").upper()).split() if len(w) >= 3}
|
|
|
|
live_names = {norm_id(r["num_id"]): s(r["nombre"]) for _, r in live.iterrows()}
|
|
blocked: set[str] = set()
|
|
for _, r in arch.iterrows():
|
|
numid = norm_id(r["num_id"])
|
|
was, now = s(r["nombre"]), live_names.get(numid)
|
|
if not numid or not was or not now:
|
|
continue
|
|
if toks(was) & toks(now):
|
|
continue
|
|
blocked.add(numid)
|
|
recycle_report.append((year, numid, was, now))
|
|
return blocked
|
|
|
|
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"])))
|
|
|
|
# Shared across both calls so BACKUP is de-duplicated against EFECTIVO —
|
|
# order matters: EFECTIVO is the live table and loads first, so a collision
|
|
# always resolves in its favour.
|
|
cash_seen: set = set()
|
|
# "CASH DEPOSIT" is not a guess: matching these rows to the live site on
|
|
# (NUMid, date, amount) resolves to that label unanimously — 66/66 in the
|
|
# current-year `datosfreak` and 100/100 in the prior-year `2025` table,
|
|
# which are the only two periods the site exposes.
|
|
efectivo_like("stg_utilities", "efectivo", "UTILITY", util_cust, "UTILITIES",
|
|
"EFECTIVO", seen=cash_seen, type_label="CASH DEPOSIT")
|
|
efectivo_like("stg_utilities", "efectivo_backup", "UTILITY", util_cust, "UTILITIES",
|
|
"EFECTIVO_BACKUP", seen=cash_seen, type_label="CASH DEPOSIT")
|
|
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()
|
|
|
|
# --- prior periods -----------------------------------------------------
|
|
#
|
|
# Legacy kept each closed year in its own table and opened the next one with
|
|
# a Jan-1 BALANCE FORWARD carrying the closing total. The platform has one
|
|
# `transactions` table, so the period a row belongs to has to travel with
|
|
# the row: it rides in legacySourceTable as `datos2@2025`.
|
|
#
|
|
# That tag, not the date, is what a year view should filter on. The archives
|
|
# are not cleanly bounded — 2025's ledger carries ten undated rows and two
|
|
# dated into 2026 — and legacy itself never filtered by date either: its
|
|
# reader is `SELECT ... FROM \`2025\``. Keying on provenance reproduces the
|
|
# legacy period exactly and strands nothing.
|
|
#
|
|
# The tag also keeps the unique key safe. legacyId is a positional row
|
|
# ordinal, so every archive restarts it at 0 and would collide with the live
|
|
# `datos2` row-for-row if they shared a source-table name.
|
|
periods = sorted(
|
|
int(d.name.rsplit("_", 1)[1])
|
|
for d in STG.glob("stg_period_*")
|
|
if d.is_dir() and d.name.rsplit("_", 1)[1].isdigit()
|
|
)
|
|
for year in periods:
|
|
billing("datos2", f"datos2@{year}", src=f"stg_period_{year}",
|
|
skip_numids=period_numid_guard(year))
|
|
# Same record shape in the seguros DB. Labelled for consistency in the
|
|
# platform's own UI; unverifiable against the site, which only ever reads
|
|
# domain='UTILITY', so no customer-facing behaviour depends on it.
|
|
efectivo_like("stg_seguros", "efectivo", "INSURANCE", ins_cust, "SEGUROS 16_be",
|
|
"EFECTIVO", type_label="CASH DEPOSIT")
|
|
|
|
if sync_mode:
|
|
# Transaction types are rebuilt with fresh uuids each run; resolve them
|
|
# against the rows already in the DB by English name (inserting any that
|
|
# are new) and remap each tx's typeId onto the persisted id so the FK to
|
|
# type_transactions holds. exchange_rates isn't referenced by tx, so it
|
|
# is left untouched in sync.
|
|
c.execute("SELECT id,nameEn FROM type_transactions")
|
|
db_types = {(nm or "").upper(): i for i, nm in c.fetchall()}
|
|
fresh_name = {tid: (en or "").upper() for tid, en, es, active in type_rows}
|
|
new_types = []
|
|
for tid, en, es, active in type_rows:
|
|
if (en or "").upper() not in db_types:
|
|
db_types[(en or "").upper()] = tid
|
|
new_types.append((tid, en, es, active))
|
|
if new_types:
|
|
c.executemany("INSERT INTO type_transactions (id,nameEn,nameEs,isService) VALUES (%s,%s,%s,%s)", new_types)
|
|
tx = [(t[0], t[1], t[2], (db_types.get(fresh_name.get(t[3])) if t[3] else None), *t[4:]) for t in tx]
|
|
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) ON DUPLICATE KEY UPDATE customerId=VALUES(customerId),domain=VALUES(domain),typeId=VALUES(typeId),transactionDate=VALUES(transactionDate),period=VALUES(period),reference=VALUES(reference),amount=VALUES(amount),currency=VALUES(currency),checkNumber=VALUES(checkNumber),message=VALUES(message),outstanding=VALUES(outstanding),voidedAt=NULL", tx)
|
|
else:
|
|
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" skipped (EFECTIVO_BACKUP dup): {skip_dupe}")
|
|
print(f" skipped (reissued NUMid) : {skip_recycled}")
|
|
print(f" -> transactions : {count('transactions')}")
|
|
print(f" by domain : {dict(by_dom)}")
|
|
for src, n in by_src:
|
|
print(f" {(src or '(manual)'):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"
|
|
|
|
if recycle_report:
|
|
print(f" ! reissued NUMids, prior-period rows NOT imported: {len(recycle_report)}")
|
|
for year, numid, was, now in recycle_report[:8]:
|
|
print(f" {year} NUMid {numid}: '{was}' -> '{now}'")
|
|
if len(recycle_report) > 8:
|
|
print(f" ... {len(recycle_report) - 8} more")
|
|
|
|
for year in periods:
|
|
verify_corte(c, year)
|
|
|
|
print(" validation: OK")
|
|
conn.close()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|