jorgecuadros-platform/migration/backfill_bank_accounts.py

"""
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()