rmancinasandClaude Opus 5
4d5008b545
Staff key 300+ utility statements per company per month by hand. This adds the ingest -> split -> OCR -> match -> review pipeline that proposes customer and amount per page instead (RECEIPT_CAPTURE_SPEC §2), posting through the existing BillingService.createBatch seam with source=OCR and a per-document captureRef so machine and hand capture share one write path and audit trail. Everything was designed against 10 real scanned statements (46 pages of CFE, CESPT and Telnor bills) rather than from the sample-free spec. The scans have no text layer at all — they are camera images — so OCR is mandatory, and they arrive bundled one customer per page. Measured on those pages the parser identifies the provider 46/46 and reads an account reference 43/46; against the dev database that is 39/46 (85%) exact auto-match, 40/46 identified, with the rest genuine review cases. That closes the OCR-provider question in favour of self-hosted Tesseract: it clears the bar for a queue where a human confirms every row, and OcrProvider keeps a managed API a one-line swap. The samples corrected three things the spec had wrong or unknown: - Clave catastral is NOT predial. DATMEX.clave (934 rows) is what CESPT and predial bills print; DATMEX.predial, which PROPERTY_TAX.accountNumber holds, has 663 distinct values across 1135 rows and appears on no statement. The clave now lives on Property.cadastralKey as the matcher's secondary key; predial is left untouched. This had been blocking predial matching. - Gas was recoverable: 160 of 334 DATMEX.gas values are real account numbers (the rest are ESTACIONARIO/CILINDRO descriptors), now in GAS.meterNumber. - Phone is one billed line per property (534/18/1 across phone1/2/3), so the new TELEPHONE ServiceKind backfills from phone1 only, not three rows. Matching is scoped to one column per service kind and never reads the customer name — a CESPT receipt prints ARNAIZ ROSAS ELSA AURORA for an account this office holds under CATT, RANDY, because the printed name is the registrant, not the current owner. Where a provider prints a payment barcode it beats the printed label (one CFE label OCR'd a digit too many while its barcode was correct) and the two cross-check, with disagreement forcing review. Confirming a document whose service had no reference writes it back, so gas and any other cold start is a one-time cost rather than a permanent queue. Verified end to end against the live dev API and MinIO: real scans uploaded over HTTP, matched, confirmed against a check, and the resulting rows checked in MySQL (negative amounts, captureSource=OCR, concept derived from the batch kind, captureRef linking back to each page). Re-confirming a posted batch is refused. Test data was removed afterwards. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
111 lines
4.3 KiB
Python
111 lines
4.3 KiB
Python
"""
|
|
Run the full data migration against one environment, in dependency order.
|
|
|
|
Every step is idempotent (truncate + rebuild), so this is safe to re-run. The
|
|
target DB is chosen with --env (reads deploy/.env.<env>); the same staged
|
|
Parquet feeds every environment.
|
|
|
|
Prerequisites (once per environment, NOT done here):
|
|
1. DB stack deployed (deploy/jorgecuadros-db.stack.yml) and deploy/.env.<env> written.
|
|
2. Prisma schema pushed to it:
|
|
DATABASE_URL="<that env's url>" \
|
|
npx prisma@5 db push --schema=packages/database/prisma/schema.prisma
|
|
|
|
Then:
|
|
./.venv/bin/python run_all.py --env dev # data only (staging already present)
|
|
./.venv/bin/python run_all.py --env prod --stage # re-extract from Access first, then load
|
|
|
|
Reproducing dev -> prod is exactly `--env prod` (plus --stage if the staged
|
|
Parquet isn't present on the machine running it).
|
|
|
|
Note: the blob_extract step re-reads the original Access files directly (the
|
|
blobs are not in the staged Parquet), so the machine running this needs
|
|
SOURCE_ROOT + mdbtools + MinIO credentials even without --stage.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import subprocess
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
HERE = Path(__file__).parent
|
|
PY = sys.executable # the venv python running this orchestrator
|
|
|
|
# Dependency order. Every step truncates what it owns, so anything downstream
|
|
# of a truncated table has to be rebuilt in the same pass — blob_extract is in
|
|
# this list because transform_properties and transform_policies truncate
|
|
# service_documents / policy_documents, which would otherwise leave the
|
|
# uploaded MinIO objects with no rows pointing at them.
|
|
STEPS = [
|
|
"transform_customers.py",
|
|
"transform_properties.py",
|
|
# Statement-OCR match fields. transform_properties.py now produces these
|
|
# directly, so on a full rebuild this is a no-op that re-asserts they are
|
|
# there; on a database predating the OCR module it is what fills them in.
|
|
# Must follow transform_properties.py, which truncates both tables it
|
|
# touches.
|
|
"backfill_statement_match_fields.py",
|
|
"transform_policies.py",
|
|
"transform_transactions.py",
|
|
"prune_empty_customers.py",
|
|
# Seeds the Scotiabank chequera that every SCOTHIA movement is booked into;
|
|
# transform_bank.py fails fast without it.
|
|
"backfill_bank_accounts.py",
|
|
"transform_bank.py",
|
|
"blob_extract.py",
|
|
]
|
|
|
|
SYNC_STEPS = [
|
|
"transform_customers.py",
|
|
"transform_properties.py",
|
|
# Statement-OCR match fields. transform_properties.py now produces these
|
|
# directly, so on a full rebuild this is a no-op that re-asserts they are
|
|
# there; on a database predating the OCR module it is what fills them in.
|
|
# Must follow transform_properties.py, which truncates both tables it
|
|
# touches.
|
|
"backfill_statement_match_fields.py",
|
|
"transform_policies.py",
|
|
"transform_transactions.py",
|
|
# Manual-safe prune: drops legacy-owned empties that the customer upsert
|
|
# re-creates from Parquet, but leaves manually-added customers alone.
|
|
"prune_empty_customers.py",
|
|
# Seeds the Scotiabank chequera that every SCOTHIA movement is booked into;
|
|
# transform_bank.py fails fast without it.
|
|
"backfill_bank_accounts.py",
|
|
"transform_bank.py",
|
|
]
|
|
|
|
|
|
def run(cmd: list[str]) -> None:
|
|
print("+ " + " ".join(cmd), flush=True)
|
|
r = subprocess.run(cmd)
|
|
if r.returncode:
|
|
sys.exit(r.returncode)
|
|
|
|
|
|
def main() -> None:
|
|
ap = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter)
|
|
ap.add_argument("--env", default="dev", help="target environment (reads deploy/.env.<env>)")
|
|
ap.add_argument("--stage", action="store_true",
|
|
help="re-run the raw staging load first (needs the Access files + mdbtools)")
|
|
ap.add_argument("--sync", action="store_true",
|
|
help="upsert legacy rows and archive removed legacy rows; preserve manual rows")
|
|
args = ap.parse_args()
|
|
|
|
if args.stage:
|
|
run([PY, str(HERE / "load_staging.py"), "--output-dir", str(HERE / "output")])
|
|
|
|
for step in SYNC_STEPS if args.sync else STEPS:
|
|
cmd = [PY, str(HERE / step), "--env", args.env]
|
|
if args.sync:
|
|
cmd.append("--sync")
|
|
run(cmd)
|
|
|
|
print(f"\n✓ migration complete for env={args.env}")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|