script.sh

gistfile1.txt

"""
Backfill loan_expiry_date on immovable mtg_collateral_application rows.

Legacy NOTICE.TERMINATION_DATE uses DD-MON-YY (e.g. 10-DEC-12). Expiry is
maturity + 2 years. Run after core migration if loan_expiry_date was null due
to the wrong date format in mtg_collateral_application.py.

Examples:
  python -m modules.mortgage_management.temp_data.patch_expiry_date --dry-run
  python -m modules.mortgage_management.temp_data.patch_expiry_date --execute
  python -m modules.mortgage_management.temp_data.patch_expiry_date --execute --mortgage-id 242618
  python -m modules.mortgage_management.temp_data.patch_expiry_date --execute --force
  python -m modules.mortgage_management.temp_data.patch_expiry_date --dry-run --verbose
"""

from __future__ import annotations

import argparse
import logging
import os
import sys
import time
from dataclasses import dataclass, field
from typing import Any, Dict, List, Optional, Set, Tuple

import pandas as pd
from sqlalchemy import text
from sqlalchemy.engine import Connection
from sqlalchemy.exc import SQLAlchemyError

from modules.common.utils import convert_date_format, chunk_size, engine, schema_name

_BASE = os.path.abspath(os.path.join(os.path.dirname(__file__), ".."))
NOTICE_CSV = os.path.join(_BASE, "files", "NOTICE.csv")

BLUE = "\033[94m"
RESET = "\033[0m"

logger = logging.getLogger(__name__)

SAMPLE_LIMIT = 15


@dataclass
class PatchStats:
    total_apps: int = 0
    skipped_already_set: int = 0
    skipped_unchanged: int = 0
    skipped_no_maturity: int = 0
    skipped_unparseable: int = 0
    maturity_from_notice_csv: int = 0
    maturity_from_db_column: int = 0
    notice_ids_requested: int = 0
    notice_ids_resolved: int = 0
    notice_csv_chunks_scanned: int = 0
    notice_csv_rows_matched: int = 0
    updates_planned: int = 0
    updates_applied: int = 0
    sample_already_set: List[str] = field(default_factory=list)
    sample_unchanged: List[str] = field(default_factory=list)
    sample_no_maturity: List[str] = field(default_factory=list)
    sample_unparseable: List[str] = field(default_factory=list)
    sample_updates: List[str] = field(default_factory=list)


def _format_duration(seconds: float) -> str:
    if seconds < 0:
        seconds = 0
    total = int(seconds)
    h = total // 3600
    m = (total % 3600) // 60
    s = total % 60
    if h > 0:
        return f"{h}h {m}m {s}s"
    if m > 0:
        return f"{m}m {s}s"
    return f"{s}s"


def print_progress_bar(
    current: int,
    total: int,
    prefix: str = "",
    elapsed_seconds: Optional[float] = None,
    eta_seconds: Optional[float] = None,
    details: str = "",
) -> None:
    total = max(total, 1)
    clamped = min(max(current, 0), total)
    pct = (clamped / total) * 100
    width = 30
    filled = int((clamped / total) * width)
    bar = "#" * filled + "-" * (width - filled)
    label = f"{prefix} " if prefix else ""
    extra = ""
    if elapsed_seconds is not None:
        extra += f" | elapsed {_format_duration(elapsed_seconds)}"
    if eta_seconds is not None and clamped < total:
        extra += f" | eta {_format_duration(eta_seconds)}"
    if details:
        extra += f" | {details}"
    end = "\n" if clamped >= total else "\r"
    print(f"{label}[{bar}] {pct:6.2f}% ({clamped}/{total}){extra}", end=end, flush=True)
    if clamped >= total:
        print(flush=True)


def _append_sample(samples: List[str], message: str) -> None:
    if len(samples) < SAMPLE_LIMIT:
        samples.append(message)


def configure_logging(verbose: bool) -> None:
    level = logging.DEBUG if verbose else logging.INFO
    logging.basicConfig(
        level=level,
        format="%(asctime)s | %(levelname)s | %(message)s",
        datefmt="%Y-%m-%d %H:%M:%S",
        stream=sys.stdout,
        force=True,
    )


def log_phase_start(phase: str) -> float:
    logger.info("Starting phase: %s", phase)
    return time.time()


def log_phase_end(phase: str, started_at: float, detail: str = "") -> None:
    elapsed = time.time() - started_at
    suffix = f" ({detail})" if detail else ""
    logger.info("Finished phase: %s in %s%s", phase, _format_duration(elapsed), suffix)


def clean_notice_id(value: Any) -> Optional[str]:
    if value is None:
        return None
    try:
        if pd.isna(value):
            return None
    except Exception:
        pass
    s = str(value).strip()
    if not s:
        return None
    try:
        return str(int(float(s)))
    except Exception:
        return s


def compute_loan_expiry_date(maturity_value: Any) -> Optional[str]:
    normalized = convert_date_format(maturity_value)
    if not normalized:
        return None
    maturity = pd.to_datetime(normalized, errors="coerce")
    if pd.isna(maturity):
        return None
    expiry = maturity + pd.DateOffset(years=2)
    return expiry.strftime("%Y-%m-%d %H:%M:%S")


def load_termination_dates_by_notice(
    notice_ids: Set[str], stats: PatchStats
) -> Dict[str, Any]:
    if not os.path.isfile(NOTICE_CSV):
        raise FileNotFoundError(f"NOTICE.csv not found: {NOTICE_CSV}")
    if not notice_ids:
        logger.warning("No notice IDs to resolve from NOTICE.csv")
        return {}

    file_size_mb = os.path.getsize(NOTICE_CSV) / (1024 * 1024)
    logger.info(
        "Scanning NOTICE.csv (%.1f MB) for %s notice ID(s)",
        file_size_mb,
        len(notice_ids),
    )

    out: Dict[str, Any] = {}
    started_at = time.time()
    chunk_count = 0
    for chunk in pd.read_csv(
        NOTICE_CSV,
        usecols=["ID", "TERMINATION_DATE"],
        dtype={"ID": str, "TERMINATION_DATE": str},
        chunksize=chunk_size,
        low_memory=False,
    ):
        chunk_count += 1
        stats.notice_csv_chunks_scanned = chunk_count
        chunk["_notice_id"] = chunk["ID"].apply(clean_notice_id)
        sub = chunk[chunk["_notice_id"].isin(notice_ids)]
        if not sub.empty:
            stats.notice_csv_rows_matched += len(sub)
            for _, row in sub.iterrows():
                notice_id = row["_notice_id"]
                if notice_id:
                    out[notice_id] = row.get("TERMINATION_DATE")
                    logger.debug(
                        "NOTICE match notice_id=%s termination_date=%r",
                        notice_id,
                        row.get("TERMINATION_DATE"),
                    )

        if chunk_count % 20 == 0:
            elapsed = time.time() - started_at
            resolved = len(out)
            logger.info(
                "NOTICE.csv scan progress: chunk=%s resolved=%s/%s elapsed=%s",
                chunk_count,
                resolved,
                len(notice_ids),
                _format_duration(elapsed),
            )

    stats.notice_ids_resolved = len(out)
    missing = notice_ids - set(out.keys())
    logger.info(
        "NOTICE.csv scan complete: chunks=%s rows_matched=%s resolved=%s missing=%s",
        chunk_count,
        stats.notice_csv_rows_matched,
        len(out),
        len(missing),
    )
    if missing and logger.isEnabledFor(logging.DEBUG):
        preview = sorted(missing)[:SAMPLE_LIMIT]
        logger.debug(
            "Sample notice IDs not found in NOTICE.csv (%s total): %s",
            len(missing),
            ", ".join(preview),
        )
    return out


def fetch_applications(
    connection: Connection, mortgage_id_filter: Optional[str]
) -> pd.DataFrame:
    q = f"""
        SELECT id, previous_pk, previous_notice_id, loan_maturity_date, loan_expiry_date
        FROM {schema_name}.mtg_collateral_application
        WHERE collateral_type = 'IMMOVABLE'
          AND previous_notice_id IS NOT NULL
    """
    params: Dict[str, Any] = {}
    if mortgage_id_filter:
        q += " AND CAST(previous_pk AS TEXT) = :mid"
        params["mid"] = str(mortgage_id_filter).strip()
        logger.info("Filtering applications to mortgage previous_pk=%s", params["mid"])
    else:
        logger.info("Fetching all immovable collateral applications with previous_notice_id")

    df = pd.read_sql(text(q), connection, params=params)
    null_expiry = int(df["loan_expiry_date"].isna().sum()) if not df.empty else 0
    logger.info(
        "Fetched %s application(s) from %s.mtg_collateral_application (loan_expiry_date null=%s)",
        len(df),
        schema_name,
        null_expiry,
    )
    return df


def build_updates(
    apps: pd.DataFrame,
    termination_by_notice: Dict[str, Any],
    only_null: bool,
    stats: PatchStats,
) -> List[Tuple[str, str]]:
    updates: List[Tuple[str, str]] = []
    stats.total_apps = len(apps)
    total = len(apps.index)
    started_at = time.time()

    for idx, (_, row) in enumerate(apps.iterrows(), start=1):
        app_id = str(row["id"])
        mortgage_id = row.get("previous_pk")
        notice_id = clean_notice_id(row.get("previous_notice_id"))

        if only_null and pd.notna(row.get("loan_expiry_date")):
            stats.skipped_already_set += 1
            _append_sample(
                stats.sample_already_set,
                f"app_id={app_id} mortgage={mortgage_id} notice={notice_id} "
                f"existing_expiry={row.get('loan_expiry_date')}",
            )
            continue

        maturity_source = None
        source_label = None
        if notice_id and notice_id in termination_by_notice:
            notice_maturity = termination_by_notice[notice_id]
            if notice_maturity is not None and not (
                isinstance(notice_maturity, float) and pd.isna(notice_maturity)
            ):
                maturity_source = notice_maturity
                source_label = "NOTICE.csv"
                stats.maturity_from_notice_csv += 1

        if maturity_source is None or (
            isinstance(maturity_source, float) and pd.isna(maturity_source)
        ):
            db_maturity = row.get("loan_maturity_date")
            if db_maturity is not None and not (
                isinstance(db_maturity, float) and pd.isna(db_maturity)
            ):
                maturity_source = db_maturity
                source_label = "loan_maturity_date"
                stats.maturity_from_db_column += 1

        if maturity_source is None or (
            isinstance(maturity_source, float) and pd.isna(maturity_source)
        ):
            stats.skipped_no_maturity += 1
            _append_sample(
                stats.sample_no_maturity,
                f"app_id={app_id} mortgage={mortgage_id} notice={notice_id}",
            )
            logger.debug(
                "Skip no maturity source app_id=%s mortgage=%s notice=%s",
                app_id,
                mortgage_id,
                notice_id,
            )
            continue

        expiry = compute_loan_expiry_date(maturity_source)
        if not expiry:
            stats.skipped_unparseable += 1
            _append_sample(
                stats.sample_unparseable,
                f"app_id={app_id} mortgage={mortgage_id} notice={notice_id} "
                f"source={source_label} raw={maturity_source!r}",
            )
            logger.debug(
                "Skip unparseable maturity app_id=%s mortgage=%s notice=%s source=%s raw=%r",
                app_id,
                mortgage_id,
                notice_id,
                source_label,
                maturity_source,
            )
            continue

        current = row.get("loan_expiry_date")
        if pd.notna(current):
            current_str = pd.to_datetime(current).strftime("%Y-%m-%d %H:%M:%S")
            if current_str == expiry:
                stats.skipped_unchanged += 1
                _append_sample(
                    stats.sample_unchanged,
                    f"app_id={app_id} mortgage={mortgage_id} expiry={expiry}",
                )
                continue

        updates.append((app_id, expiry))
        _append_sample(
            stats.sample_updates,
            f"app_id={app_id} mortgage={mortgage_id} notice={notice_id} "
            f"source={source_label} maturity={maturity_source!r} expiry={expiry}",
        )
        logger.debug(
            "Plan update app_id=%s mortgage=%s notice=%s source=%s maturity=%r expiry=%s",
            app_id,
            mortgage_id,
            notice_id,
            source_label,
            maturity_source,
            expiry,
        )

        if idx % 5000 == 0 or idx == total:
            elapsed = time.time() - started_at
            eta = None
            if idx > 0 and idx < total:
                rate = idx / max(elapsed, 1e-6)
                eta = (total - idx) / max(rate, 1e-6)
            print_progress_bar(
                idx,
                total,
                prefix="Build updates",
                elapsed_seconds=elapsed,
                eta_seconds=eta,
                details=f"planned={len(updates)}",
            )

    stats.updates_planned = len(updates)
    logger.info(
        "Update plan complete: planned=%s already_set=%s unchanged=%s no_maturity=%s unparseable=%s",
        stats.updates_planned,
        stats.skipped_already_set,
        stats.skipped_unchanged,
        stats.skipped_no_maturity,
        stats.skipped_unparseable,
    )
    return updates


def apply_updates(
    connection: Connection, updates: List[Tuple[str, str]], stats: PatchStats
) -> int:
    if not updates:
        logger.info("No updates to apply")
        return 0

    stmt = text(
        f"""
        UPDATE {schema_name}.mtg_collateral_application
        SET loan_expiry_date = :expiry
        WHERE id = :app_id
        """
    )
    batch_size = 1000
    updated = 0
    total_batches = (len(updates) + batch_size - 1) // batch_size
    started_at = time.time()
    logger.info(
        "Applying %s update(s) in %s batch(es) of up to %s row(s)",
        len(updates),
        total_batches,
        batch_size,
    )

    for batch_num, i in enumerate(range(0, len(updates), batch_size), start=1):
        batch = updates[i : i + batch_size]
        connection.execute(
            stmt,
            [{"app_id": app_id, "expiry": expiry} for app_id, expiry in batch],
        )
        updated += len(batch)
        elapsed = time.time() - started_at
        eta = None
        if updated < len(updates):
            rate = updated / max(elapsed, 1e-6)
            eta = (len(updates) - updated) / max(rate, 1e-6)
        print_progress_bar(
            updated,
            len(updates),
            prefix="Apply updates",
            elapsed_seconds=elapsed,
            eta_seconds=eta,
            details=f"batch {batch_num}/{total_batches}",
        )
        logger.debug(
            "Applied batch %s/%s rows=%s total_applied=%s",
            batch_num,
            total_batches,
            len(batch),
            updated,
        )

    stats.updates_applied = updated
    logger.info("Applied %s update(s) to %s.mtg_collateral_application", updated, schema_name)
    return updated


def _log_sample_block(title: str, samples: List[str]) -> None:
    if not samples:
        return
    logger.info("%s (showing up to %s):", title, len(samples))
    for line in samples:
        logger.info("  %s", line)


def log_summary(stats: PatchStats, mode: str) -> None:
    logger.info("=== Patch summary (%s) ===", mode)
    logger.info("Applications examined: %s", stats.total_apps)
    logger.info("Notice IDs requested / resolved: %s / %s", stats.notice_ids_requested, stats.notice_ids_resolved)
    logger.info(
        "NOTICE.csv scan: chunks=%s matched_rows=%s",
        stats.notice_csv_chunks_scanned,
        stats.notice_csv_rows_matched,
    )
    logger.info("Maturity source: NOTICE.csv=%s loan_maturity_date column=%s", stats.maturity_from_notice_csv, stats.maturity_from_db_column)
    logger.info("Skipped (already set): %s", stats.skipped_already_set)
    logger.info("Skipped (unchanged expiry): %s", stats.skipped_unchanged)
    logger.info("Skipped (no maturity source): %s", stats.skipped_no_maturity)
    logger.info("Skipped (unparseable maturity): %s", stats.skipped_unparseable)
    logger.info("Updates planned: %s", stats.updates_planned)
    if mode == "execute":
        logger.info("Updates applied: %s", stats.updates_applied)

    _log_sample_block("Sample rows skipped (already set)", stats.sample_already_set)
    _log_sample_block("Sample rows skipped (unchanged)", stats.sample_unchanged)
    _log_sample_block("Sample rows skipped (no maturity)", stats.sample_no_maturity)
    _log_sample_block("Sample rows skipped (unparseable)", stats.sample_unparseable)
    _log_sample_block("Sample planned updates", stats.sample_updates)


def run_patch(
    conn: Connection,
    args: argparse.Namespace,
    stats: PatchStats,
    mode: str,
    overall_started: float,
) -> None:
    only_null = not args.force

    phase_started = log_phase_start("fetch applications")
    apps = fetch_applications(conn, args.mortgage_id)
    log_phase_end("fetch applications", phase_started, f"{len(apps)} row(s)")

    if apps.empty:
        logger.warning("No immovable collateral applications matched filters.")
        return

    notice_ids = {
        nid
        for nid in (clean_notice_id(x) for x in apps["previous_notice_id"].tolist())
        if nid
    }
    stats.notice_ids_requested = len(notice_ids)

    phase_started = log_phase_start("load NOTICE termination dates")
    termination_by_notice = load_termination_dates_by_notice(notice_ids, stats)
    log_phase_end(
        "load NOTICE termination dates",
        phase_started,
        f"resolved {stats.notice_ids_resolved}/{stats.notice_ids_requested}",
    )

    phase_started = log_phase_start("build update plan")
    updates = build_updates(apps, termination_by_notice, only_null=only_null, stats=stats)
    log_phase_end("build update plan", phase_started, f"{len(updates)} update(s)")

    if not updates:
        logger.info("Nothing to update.")
        log_summary(stats, mode)
        return

    if args.dry_run:
        logger.info("Dry run: no database writes will be performed.")
        for app_id, expiry in updates[:10]:
            logger.info("Would set id=%s loan_expiry_date=%s", app_id, expiry)
        if len(updates) > 10:
            logger.info("... and %s more update(s)", len(updates) - 10)
        log_summary(stats, mode)
        logger.info(
            "patch_expiry_date finished in %s",
            _format_duration(time.time() - overall_started),
        )
        return

    phase_started = log_phase_start("apply database updates")
    apply_updates(conn, updates, stats)
    log_phase_end("apply database updates", phase_started, f"{stats.updates_applied} row(s)")

    log_summary(stats, mode)
    logger.info(
        "patch_expiry_date finished in %s",
        _format_duration(time.time() - overall_started),
    )


def main() -> None:
    parser = argparse.ArgumentParser(
        description="Backfill loan_expiry_date from NOTICE TERMINATION_DATE (+2 years)."
    )
    parser.add_argument(
        "--dry-run",
        action="store_true",
        help="Report rows that would be updated; no DB writes.",
    )
    parser.add_argument(
        "--execute",
        action="store_true",
        help="Apply updates to the database.",
    )
    parser.add_argument(
        "--mortgage-id",
        type=str,
        default=None,
        help="Limit to one legacy mortgage (previous_pk).",
    )
    parser.add_argument(
        "--force",
        action="store_true",
        help="Update even when loan_expiry_date is already set.",
    )
    parser.add_argument(
        "--verbose",
        action="store_true",
        help="Enable debug logging (per-row decisions and NOTICE.csv matches).",
    )
    args = parser.parse_args()

    if args.dry_run == args.execute:
        parser.error("Specify exactly one of --dry-run or --execute.")

    configure_logging(args.verbose)
    only_null = not args.force
    mode = "dry-run" if args.dry_run else "execute"
    stats = PatchStats()
    overall_started = time.time()

    logger.info("patch_expiry_date started mode=%s schema=%s only_null=%s", mode, schema_name, only_null)
    logger.info("NOTICE.csv path: %s", NOTICE_CSV)

    try:
        if args.dry_run:
            with engine.connect() as conn:
                run_patch(conn, args, stats, mode, overall_started)
        else:
            with engine.begin() as conn:
                run_patch(conn, args, stats, mode, overall_started)
    except SQLAlchemyError as exc:
        logger.exception("Database error during patch_expiry_date: %s", exc)
        raise SystemExit(1) from exc
    except FileNotFoundError as exc:
        logger.error("%s", exc)
        raise SystemExit(1) from exc


if __name__ == "__main__":
    main()