""" One-off schema+data step for the multi-bank chequera (docs/RECEIPT_CAPTURE_SPEC.md ยง3). `bank_transactions.bankAccountId` is REQUIRED in the Prisma schema, so `prisma db push` cannot introduce it on a table that already holds 22k rows. This script does the ordered dance that push can't: 1. create `banks` / `bank_accounts` (same DDL Prisma generates) 2. seed the one account every existing row belongs to โ€” Scotiabank MXN, the office's Utilities chequera, which is all `SCOTHIA.mdb` ever was 3. add `bankAccountId` NULLable, backfill every row to that account, then promote it to NOT NULL and attach the FK + index On a database that predates the feature, run it BEFORE `prisma db push`; push then sees no drift. On a fresh environment push creates the tables itself and this only seeds the rows. Either way `transform_bank.py` needs the account to exist, so `run_all.py` runs it first. Idempotent โ€” safe to re-run, and re-running once a second account exists does NOT re-point rows (the backfill only touches NULLs). ./.venv/bin/python backfill_bank_accounts.py --env dev """ from __future__ import annotations import uuid from dbenv import connect from sync import parse_mode # The account every migrated SCOTHIA row belongs to. Its id is derived, not # random, so a re-run against a half-applied database finds the same row and # `transform_bank.py` can resolve it by label without a lookup table. SCOTIABANK = "Scotiabank" UTILITIES_ACCOUNT = "Utilities โ€” Scotiabank (MXN)" def table_exists(c, name: str) -> bool: c.execute( "SELECT COUNT(*) FROM information_schema.tables " "WHERE table_schema = DATABASE() AND table_name = %s", (name,), ) return c.fetchone()[0] > 0 def column_exists(c, table: str, column: str) -> bool: c.execute( "SELECT COUNT(*) FROM information_schema.columns " "WHERE table_schema = DATABASE() AND table_name = %s AND column_name = %s", (table, column), ) return c.fetchone()[0] > 0 def constraint_exists(c, table: str, name: str) -> bool: c.execute( "SELECT COUNT(*) FROM information_schema.table_constraints " "WHERE table_schema = DATABASE() AND table_name = %s AND constraint_name = %s", (table, name), ) return c.fetchone()[0] > 0 def index_exists(c, table: str, name: str) -> bool: c.execute( "SELECT COUNT(*) FROM information_schema.statistics " "WHERE table_schema = DATABASE() AND table_name = %s AND index_name = %s", (table, name), ) return c.fetchone()[0] > 0 def main(): # `--sync` is accepted and ignored: this step is idempotent by nature, so # it behaves identically in both modes and can sit in run_all's two lists. env, _sync_mode = parse_mode() conn = connect(env) c = conn.cursor() print(f"[bank-accounts] target env: {env}") # --- 1. tables ---------------------------------------------------------- if not table_exists(c, "banks"): c.execute( """ CREATE TABLE `banks` ( `id` VARCHAR(191) NOT NULL, `name` VARCHAR(191) NOT NULL, `country` VARCHAR(191) NULL, PRIMARY KEY (`id`), UNIQUE KEY `banks_name_key` (`name`) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci """ ) print(" created banks") if not table_exists(c, "bank_accounts"): c.execute( """ CREATE TABLE `bank_accounts` ( `id` VARCHAR(191) NOT NULL, `bankId` VARCHAR(191) NOT NULL, `label` VARCHAR(191) NOT NULL, `currency` ENUM('USD','MXN') NOT NULL, `businessLine` ENUM('UTILITY','INSURANCE','TRUST') NULL, `active` TINYINT(1) NOT NULL DEFAULT 1, PRIMARY KEY (`id`), KEY `bank_accounts_bankId_fkey` (`bankId`), CONSTRAINT `bank_accounts_bankId_fkey` FOREIGN KEY (`bankId`) REFERENCES `banks` (`id`) ON DELETE RESTRICT ON UPDATE CASCADE ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci """ ) print(" created bank_accounts") # --- 2. seed the Utilities/Scotiabank chequera -------------------------- c.execute("SELECT id FROM banks WHERE name = %s", (SCOTIABANK,)) row = c.fetchone() if row: bank_id = row[0] else: bank_id = str(uuid.uuid4()) c.execute( "INSERT INTO banks (id, name, country) VALUES (%s, %s, %s)", (bank_id, SCOTIABANK, "MX"), ) print(f" seeded bank {SCOTIABANK}") c.execute("SELECT id FROM bank_accounts WHERE label = %s", (UTILITIES_ACCOUNT,)) row = c.fetchone() if row: account_id = row[0] else: account_id = str(uuid.uuid4()) c.execute( "INSERT INTO bank_accounts (id, bankId, label, currency, businessLine, active) " "VALUES (%s, %s, %s, 'MXN', 'UTILITY', 1)", (account_id, bank_id, UTILITIES_ACCOUNT), ) print(f" seeded account {UTILITIES_ACCOUNT}") print(f" account id: {account_id}") # --- 3. column, backfill, promote to NOT NULL --------------------------- if not column_exists(c, "bank_transactions", "bankAccountId"): c.execute("ALTER TABLE `bank_transactions` ADD COLUMN `bankAccountId` VARCHAR(191) NULL") print(" added bank_transactions.bankAccountId (nullable)") c.execute( "UPDATE bank_transactions SET bankAccountId = %s WHERE bankAccountId IS NULL", (account_id,), ) print(f" backfilled {c.rowcount} movement(s) to {UTILITIES_ACCOUNT}") c.execute("SELECT COUNT(*) FROM bank_transactions WHERE bankAccountId IS NULL") orphans = c.fetchone()[0] if orphans: raise SystemExit(f"abort: {orphans} bank_transactions still have no account") c.execute("ALTER TABLE `bank_transactions` MODIFY `bankAccountId` VARCHAR(191) NOT NULL") if not index_exists(c, "bank_transactions", "bank_transactions_bankAccountId_transactionDate_idx"): c.execute( "CREATE INDEX `bank_transactions_bankAccountId_transactionDate_idx` " "ON `bank_transactions` (`bankAccountId`, `transactionDate`)" ) print(" created (bankAccountId, transactionDate) index") if not constraint_exists(c, "bank_transactions", "bank_transactions_bankAccountId_fkey"): c.execute( "ALTER TABLE `bank_transactions` " "ADD CONSTRAINT `bank_transactions_bankAccountId_fkey` FOREIGN KEY (`bankAccountId`) " "REFERENCES `bank_accounts` (`id`) ON DELETE RESTRICT ON UPDATE CASCADE" ) print(" attached bankAccountId FK") conn.commit() c.execute( "SELECT a.label, a.currency, COUNT(t.id), COALESCE(SUM(t.amount), 0) " "FROM bank_accounts a LEFT JOIN bank_transactions t ON t.bankAccountId = a.id " "GROUP BY a.id, a.label, a.currency ORDER BY a.label" ) print("=== Multi-bank chequera ready ===") for label, currency, n, total in c.fetchall(): print(f" {label:36} {currency} {n:6} movimientos neto {total}") print(" validation: OK") conn.close() if __name__ == "__main__": main()