diff --git a/deploy/jorgecuadros-minio.stack.yml b/deploy/jorgecuadros-minio.stack.yml new file mode 100644 index 0000000..a8682e8 --- /dev/null +++ b/deploy/jorgecuadros-minio.stack.yml @@ -0,0 +1,56 @@ +# S3-compatible object storage (MinIO) for the platform's document blobs. +# +# Holds the scanned utility bills / IDs / policy docs extracted from the +# Access LONGBINARY columns (migration step 4). MySQL keeps only the pointer +# (storageKey) + metadata; the bytes live here. +# +# Target: Portainer local endpoint on cubex (3-node Swarm). Same statefulness +# rules as the MySQL stack (deploy/jorgecuadros-db.stack.yml): named volume + +# pinned to one node so the data volume is stable. Reuses the same node label. +# +# DEV / PROD as two stacks from this one file: +# dev : stack jorgecuadros-dev-minio API 9100 / console 9101 +# prod: stack jorgecuadros-prod-minio API 9000 / console 9001 +# Swarm namespaces the volume per stack name -> isolated data per environment. +# +# Secrets (MINIO_ROOT_USER / MINIO_ROOT_PASSWORD) injected via Portainer stack +# env at deploy time, not committed. + +version: "3.8" + +services: + minio: + image: minio/minio:RELEASE.2024-10-13T13-34-11Z + command: server /data --console-address ":9001" + environment: + MINIO_ROOT_USER: ${MINIO_ROOT_USER:?MINIO_ROOT_USER must be set} + MINIO_ROOT_PASSWORD: ${MINIO_ROOT_PASSWORD:?MINIO_ROOT_PASSWORD must be set} + ports: + - target: 9000 + published: ${MINIO_API_PORT:-9000} + protocol: tcp + mode: ingress + - target: 9001 + published: ${MINIO_CONSOLE_PORT:-9001} + protocol: tcp + mode: ingress + volumes: + - minio_data:/data + deploy: + replicas: 1 + placement: + constraints: + - node.labels.jorgecuadros_db == true + restart_policy: + condition: any + update_config: + order: stop-first + healthcheck: + test: ["CMD-SHELL", "mc ready local || curl -f http://localhost:9000/minio/health/live || exit 1"] + interval: 10s + timeout: 5s + retries: 12 + start_period: 20s + +volumes: + minio_data: diff --git a/migration/blob_extract.py b/migration/blob_extract.py new file mode 100644 index 0000000..4464063 --- /dev/null +++ b/migration/blob_extract.py @@ -0,0 +1,205 @@ +""" +Migration plan step 4: extract LONGBINARY document blobs to object storage. + +The Access LONGBINARY columns hold scanned utility bills / IDs / policy docs +wrapped in an Access OLE Object container (a "\\x15\\x1c...Pres..." header + +optional DIB preview, then the real embedded file). Staging used +`mdb-export -b strip` (blobs dropped); this re-reads each table with +`-b hex`, carves the embedded file out of the OLE wrapper by locating its +magic bytes, uploads it to MinIO (S3), and writes a *_documents row pointing +at it (MySQL keeps only the pointer + metadata, per the plan). + +Row alignment: `mdb-export` order is deterministic and identical to the order +load_staging used, so a row's position == its staged `_row_num`. Policy docs +resolve to a policy by (legacySourceTable, legacyId=row position); property +docs resolve by DATMEX numer_id -> propertyId. + +Idempotent: truncates the *_documents tables for the selected models and +re-uploads under deterministic keys (overwrite). Use --limit N for a small +test pass, --tables to restrict. + +Run: ./.venv/bin/python blob_extract.py --env dev [--limit N] [--tables datmex,mult] +""" + +from __future__ import annotations + +import argparse +import csv +import subprocess +import sys +import uuid +from pathlib import Path + +import boto3 +from botocore.config import Config + +from dbenv import connect, load_env +from extract import sanitize_column_name as san + +csv.field_size_limit(300_000_000) + +SOURCE_ROOT = Path.home() / "Downloads" / "JorgeCuadros-Legacy" + +# (key, access_file, access_table, staged_table_name, blob_cols, model) +SOURCES = [ + dict(key="datmex", file="UTILITIES.accdb", table="DATMEX", staged="DATMEX", + blobs=["doc_1", "doc_2"], model="service"), + dict(key="mult", file="SEGUROS 16_be.mdb", table="MULT", staged="mult", + blobs=["foto1", "docs_1", "docs_2"], model="policy"), + dict(key="autos_ampl", file="SEGUROS 16_be.mdb", table="TABLA AUTOS AMPL", + staged="tabla_autos_ampl", blobs=["foto1", "docs_1", "docs_2"], model="policy"), +] + +# magic -> (ext, content-type). Order = priority when several appear. +MAGICS = [ + (b"\xff\xd8\xff", "jpg", "image/jpeg"), + (b"\x89PNG\r\n\x1a\n", "png", "image/png"), + (b"%PDF", "pdf", "application/pdf"), + (b"GIF8", "gif", "image/gif"), + (b"II*\x00", "tif", "image/tiff"), + (b"MM\x00*", "tif", "image/tiff"), +] + + +def carve(b: bytes): + """Locate the embedded file inside the OLE wrapper and return + (bytes, ext, content_type) or None if no known type is present.""" + best = None + for sig, ext, ct in MAGICS: + i = b.find(sig) + if i >= 0 and (best is None or i < best[0]): + best = (i, ext, ct) + if best is None: + return None + i, ext, ct = best + data = b[i:] + # trim trailing OLE junk after the real end marker where we know it + if ext == "jpg": + e = data.rfind(b"\xff\xd9") + if e >= 0: + data = data[: e + 2] + elif ext == "png": + e = data.rfind(b"IEND\xaeB`\x82") + if e >= 0: + data = data[: e + 8] + return data, ext, ct + + +def main(): + ap = argparse.ArgumentParser() + ap.add_argument("--env", default="dev") + ap.add_argument("--limit", type=int, default=0, help="max rows per table (0 = all); test with a small N") + ap.add_argument("--tables", default="", help="comma list of source keys to run (default all)") + args = ap.parse_args() + + only = set(x.strip() for x in args.tables.split(",") if x.strip()) + sources = [s for s in SOURCES if not only or s["key"] in only] + + env = load_env(args.env) + s3 = boto3.client( + "s3", endpoint_url=env["S3_ENDPOINT"], + aws_access_key_id=env["MINIO_ROOT_USER"], aws_secret_access_key=env["MINIO_ROOT_PASSWORD"], + config=Config(signature_version="s3v4"), region_name="us-east-1") + bucket = env["S3_BUCKET"] + + conn = connect(args.env) + cur = conn.cursor() + + # parent lookups + cur.execute("SELECT legacyId, id FROM properties WHERE legacySourceTable='DATMEX'") + prop_by_numer = {} + for lid, pid in cur.fetchall(): + prop_by_numer.setdefault(lid, pid) # first property per numer_id + cur.execute("SELECT legacySourceTable, legacyId, id FROM policies") + pol_by_row = {(t, l): i for t, l, i in cur.fetchall()} + + # fresh rebuild of the doc tables we're loading (unless a limited test pass) + if not args.limit: + cur.execute("SET FOREIGN_KEY_CHECKS=0") + if any(s["model"] == "service" for s in sources): + cur.execute("TRUNCATE TABLE service_documents") + if any(s["model"] == "policy" for s in sources): + cur.execute("TRUNCATE TABLE policy_documents") + cur.execute("SET FOREIGN_KEY_CHECKS=1") + conn.commit() + + svc_rows, pol_rows = [], [] + stats = {} + + for src in sources: + path = SOURCE_ROOT / src["file"] + want = {c: None for c in src["blobs"]} + uploaded = skipped_noparent = no_magic = empty = 0 + + p = subprocess.Popen(["mdb-export", "-b", "hex", str(path), src["table"]], + stdout=subprocess.PIPE, text=True, encoding="utf-8", + errors="replace", bufsize=1) + rdr = csv.reader(p.stdout) + hdr = next(rdr) + sh = [san(c) for c in hdr] + idx = {c: sh.index(c) for c in src["blobs"] if c in sh} + numer_idx = sh.index("numer_id") if "numer_id" in sh else None + + for ri, row in enumerate(rdr): + if args.limit and ri >= args.limit: + break + # resolve parent + if src["model"] == "service": + numer = (row[numer_idx].strip() if numer_idx is not None and numer_idx < len(row) else "") + if numer.endswith(".0"): + numer = numer[:-2] + parent = prop_by_numer.get(numer) + else: + parent = pol_by_row.get((src["staged"], str(ri))) + for col, ci in idx.items(): + h = row[ci].strip() if ci < len(row) else "" + if len(h) < 16: + empty += 1 + continue + if not parent: + skipped_noparent += 1 + continue + try: + raw = bytes.fromhex(h) + except ValueError: + continue + out = carve(raw) + if out is None: + no_magic += 1 + continue + data, ext, ct = out + prefix = "service" if src["model"] == "service" else "policy" + key = f"{prefix}/{parent}/{src['staged']}_{ri}_{col}.{ext}" + s3.put_object(Bucket=bucket, Key=key, Body=data, ContentType=ct) + dtype = col.upper() + if src["model"] == "service": + svc_rows.append((str(uuid.uuid4()), parent, dtype, key)) + else: + pol_rows.append((str(uuid.uuid4()), parent, dtype, key, col)) + uploaded += 1 + p.stdout.close(); p.wait() + stats[src["key"]] = dict(uploaded=uploaded, no_parent=skipped_noparent, + no_magic=no_magic, empty_cells=empty) + print(f" [{src['key']}] uploaded={uploaded} no_parent={skipped_noparent} " + f"no_magic={no_magic}") + + if svc_rows: + cur.executemany("INSERT INTO service_documents (id,propertyId,documentType,storageKey) " + "VALUES (%s,%s,%s,%s)", svc_rows) + if pol_rows: + cur.executemany("INSERT INTO policy_documents (id,policyId,documentType,storageKey,originalColumn) " + "VALUES (%s,%s,%s,%s,%s)", pol_rows) + conn.commit() + + cur.execute("SELECT COUNT(*) FROM service_documents"); ns = cur.fetchone()[0] + cur.execute("SELECT COUNT(*) FROM policy_documents"); npd = cur.fetchone()[0] + print("=== Blob extraction complete ===") + for k, v in stats.items(): + print(f" {k}: {v}") + print(f" -> service_documents rows total: {ns}") + print(f" -> policy_documents rows total : {npd}") + conn.close() + + +if __name__ == "__main__": + main() diff --git a/migration/requirements.txt b/migration/requirements.txt index ae84ecc..1c5b6b4 100644 --- a/migration/requirements.txt +++ b/migration/requirements.txt @@ -4,6 +4,7 @@ pandas>=2.2 pyarrow>=15.0 sqlalchemy>=2.0 pymysql>=1.1 +boto3>=1.34 # S3/MinIO client for blob_extract.py (migration step 4) # Windows-only, historical — the DAO/COM object catalog (catalog_objects.py) # was already run on Windows and its output is committed (objects.json). Not