jorgecuadros-platform/migration/backfill_statement_match_fields.py

"""
Closes the three data gaps the statement-OCR matcher depends on
(docs/RECEIPT_CAPTURE_SPEC.md §2, "Matching logic").

OCR matching is only as good as the field it matches against, and a
field-by-field check of real scanned statements against what
`transform_properties.py` actually loaded turned up three mismatches. This
script fixes them on an existing database; `transform_properties.py` has been
updated in the same commit so a full re-migration produces them directly.

  1. CLAVE CATASTRAL — printed on both the CESPT water bill ("Cve. Cat.:
     KB078025") and the predial statement, and held in `DATMEX.clave` (934
     rows, format `[A-Z]{2}[0-9]{6}`). It was never migrated. What
     `PROPERTY_TAX.accountNumber` carries instead is `DATMEX.predial`, a
     different, purely numeric column that is *not* unique — 663 distinct
     values across 1135 filled rows — and appears on no statement. So predial
     is left exactly where it is, and the clave lands on `Property` (it is a
     property-level key, which is why two different services both print it).

  2. GAS — `GAS.meterNumber` is empty for all 334 rows because the transform
     put `DATMEX.gas` into `notes`. That column is mixed: 160 rows hold a real
     numeric account/meter number, the remaining 174 hold a tank descriptor
     ("ESTACIONARIO", "CILINDRO"). The numeric ones are recoverable now; the
     descriptors legitimately have no number, so those statements still start
     cold and get their number from the first human confirmation.

  3. TELEPHONE — no such `ServiceKind` existed, so a Telnor bill had nothing to
     match against. One service row is created per property with a `phone1`.
     Only phone1: 534 properties have one, 18 have a phone2 and exactly 1 has a
     phone3, so the secondaries are alternate contacts rather than separately
     billed lines.

Idempotent — re-running updates nothing it has already done, and it never
overwrites a value a human has since corrected.

    ./.venv/bin/python backfill_statement_match_fields.py --env dev
"""

from __future__ import annotations

import uuid
from pathlib import Path

import pandas as pd

from dbenv import connect
from sync import parse_mode

STG = Path(__file__).parent / "output" / "stg_utilities"
NULL = "∅"


def s(v):
    if v is None or pd.isna(v):
        return None
    v = str(v).strip()
    return None if v in ("", NULL, "0") else v


def main():
    # `--sync` is accepted and ignored — the script is idempotent either way.
    env, _sync_mode = parse_mode()
    conn = connect(env)
    c = conn.cursor()
    print(f"[statement-match-fields] target env: {env}")

    # --- 1. clave catastral -> properties.cadastralKey ----------------------
    # Joined on provenance, the same key transform_properties.py writes, so a
    # property that was re-created by a later sync still lines up.
    dm = pd.read_parquet(STG / "datmex.parquet")
    claves = []
    for _, row in dm.iterrows():
        clave = s(row["clave"])
        if clave:
            claves.append((clave, str(int(row["_row_num"]))))

    updated = 0
    for clave, legacy_id in claves:
        c.execute(
            "UPDATE properties SET cadastralKey = %s "
            "WHERE legacySourceTable = 'DATMEX' AND legacyId = %s AND cadastralKey IS NULL",
            (clave, legacy_id),
        )
        updated += c.rowcount
    print(f"  cadastralKey: set on {updated} propert(ies) ({len(claves)} in source)")

    # --- 2. gas account numbers out of notes -> GAS.meterNumber -------------
    # REGEXP rather than a Python loop: the value is already sitting in the
    # notes column, so this is one pass over 334 rows inside the database.
    c.execute(
        "UPDATE property_services SET meterNumber = notes "
        "WHERE kind = 'GAS' AND meterNumber IS NULL "
        "AND notes REGEXP '^[0-9]{5,}$'"
    )
    print(f"  GAS.meterNumber: recovered {c.rowcount} account number(s) from notes")

    # --- 3. TELEPHONE service rows ------------------------------------------
    # Digits only, matching how the transform now writes them: a scanned Telnor
    # bill prints "664 609 3444" and reduces to the stored local 6093444 once
    # the LADA is stripped, which is the matcher's job, not this script's.
    c.execute(
        "SELECT p.id, p.phone1 FROM properties p "
        "WHERE p.phone1 IS NOT NULL AND p.phone1 <> '' "
        "AND NOT EXISTS (SELECT 1 FROM property_services ps "
        "                WHERE ps.propertyId = p.id AND ps.kind = 'TELEPHONE')"
    )
    rows = c.fetchall()
    made = []
    for pid, phone in rows:
        digits = "".join(ch for ch in str(phone) if ch.isdigit())
        if digits:
            made.append((str(uuid.uuid4()), pid, digits))
    if made:
        c.executemany(
            "INSERT INTO property_services "
            "(id, propertyId, kind, accountNumber, active, notes) "
            "VALUES (%s, %s, 'TELEPHONE', %s, 1, 'from DATMEX.telefono')",
            made,
        )
    print(f"  TELEPHONE: created {len(made)} service row(s)")

    conn.commit()

    # --- validation ---------------------------------------------------------
    c.execute("SELECT COUNT(*) FROM properties WHERE cadastralKey IS NOT NULL")
    n_clave = c.fetchone()[0]
    c.execute(
        "SELECT COUNT(*) FROM property_services WHERE kind='GAS' AND meterNumber IS NOT NULL"
    )
    n_gas = c.fetchone()[0]
    c.execute("SELECT COUNT(*) FROM property_services WHERE kind='TELEPHONE'")
    n_tel = c.fetchone()[0]

    # A clave that is not unique would silently make the secondary match key
    # ambiguous, which is worse than not having one — surface it rather than
    # letting the matcher discover it a statement at a time.
    c.execute(
        "SELECT COUNT(*) FROM (SELECT cadastralKey FROM properties "
        "WHERE cadastralKey IS NOT NULL GROUP BY cadastralKey HAVING COUNT(*) > 1) d"
    )
    dupe_claves = c.fetchone()[0]

    print("=== Statement match fields ready ===")
    print(f"  properties with cadastralKey : {n_clave}")
    print(f"  GAS services with meterNumber: {n_gas}")
    print(f"  TELEPHONE services           : {n_tel}")
    print(f"  duplicated cadastralKey values: {dupe_claves}"
          + ("  (matcher treats these as ambiguous)" if dupe_claves else ""))
    assert n_clave > 0 and n_tel > 0, "backfill produced nothing — check staging output"
    print("  validation: OK")
    conn.close()


if __name__ == "__main__":
    main()