rmancinasandClaude Opus 4.8
1b79b43a54
The --sync path had never been run and was broken in several ways. Fixed and verified against the dev DB (two consecutive syncs, both exit 0, 32/32 assertions: stable PKs, manual-row preservation, changed-row updates, legacy-delete, no child duplication, zero FK orphans; idempotent). - policies/properties: reuse each legacy row's existing id (by provenance) BEFORE building child rows, so children no longer point at a discarded fresh uuid; rebuild legacy-owned children via scoped delete + reinsert. - customers: replace zip(customers, refs) (mispaired almost every row) with a ref-grouped id remap; names now restore and no spurious customers appear. - drop the invalid Vehicle @@unique(legacySourceTable, legacyId) — one legacy policy row carries up to 3 vehicles sharing a legacyId; handle via delete+reinsert. - upsert lookup tables (policy_types, insurance_providers, type_transactions, adjusters) by natural name and remap child FKs instead of inserting fresh uuids that nothing points at. - transactions: drop updatedAt=NOW() (no such column); guard report formatting on NULL legacySourceTable (manual rows). Same report guard in bank. - add manual-safe prune (prune_empty_customers.py --sync, in SYNC_STEPS): prune only legacy-owned empties, never manually-added customers. web: customer-detail mini tx list now strikes voided rows with an "(anulado)" tag (was the last void-UI rendering gap; /estado-cuenta already handled it). docs: RESUME.md updated — Phase B sync marked verified end-to-end, void-UI browser pass recorded. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
445 lines
24 KiB
Python
445 lines
24 KiB
Python
"""
|
|
Migration plan step 3 (policies): consolidate every insurance line into one
|
|
`policies` table + child tables, resolving each policy's customer FK through
|
|
the insurance customer_legacy_refs (run AFTER transform_customers.py).
|
|
|
|
One Access table per line of business becomes rows in `policies` tagged by a
|
|
policy_types discriminator, plus:
|
|
- policy_payment_installments : the 4 hardcoded payment slots, unpivoted
|
|
- vehicles : the up-to-3 hardcoded vehicle slots
|
|
- insured_drivers : the up-to-3 hardcoded named-insured slots
|
|
- policy_beneficiaries : from BENEF (by policy number)
|
|
- claims : from DATOS (siniestros)
|
|
- adjusters : from AJUSTADORES / AJUSTADORESATLAS
|
|
|
|
Design (validated against staged data):
|
|
- Customer link column varies (num_id / numer_id); ~10 policy customer ids
|
|
across all tables don't resolve to a loaded insurance customer -> those
|
|
policies are skipped and counted (required FK).
|
|
- Payment slots: c_1er_pago is the first amount, pago_subsec the recurring
|
|
amount for slots 2-4; efectivo is a cash flag, no_cheque the check ref.
|
|
- Any source column not explicitly modeled (coverage amounts: edificio,
|
|
contenidos, robo, cristales, ...) is preserved verbatim in coveragesJson,
|
|
so nothing is lost in consolidation.
|
|
- docs_1/docs_2/foto1 (LONGBINARY) are skipped — documents are step 4.
|
|
|
|
Run: ./.venv/bin/python transform_policies.py --env dev
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import uuid
|
|
from datetime import datetime, timezone
|
|
from decimal import Decimal, InvalidOperation
|
|
from pathlib import Path
|
|
|
|
import pandas as pd
|
|
|
|
from dbenv import connect, env_arg
|
|
from sync import parse_mode, existing_ids, delete_missing
|
|
|
|
STG = Path(__file__).parent / "output" / "stg_seguros"
|
|
LEGACY_DB = "SEGUROS 16_be"
|
|
NULL = "∅"
|
|
NOW = datetime.now(timezone.utc).replace(tzinfo=None)
|
|
_ADDR = {"calle_y_no", "col", "edo", "tel", "cp", "_row_num", "nombre_aseg", "observaciones"}
|
|
|
|
|
|
def s(v):
|
|
if v is None or pd.isna(v):
|
|
return None
|
|
v = str(v).strip()
|
|
return None if v in ("", NULL, "0000-00-00") else v
|
|
|
|
|
|
def norm_id(v):
|
|
v = s(v)
|
|
if v is None:
|
|
return None
|
|
if v.endswith(".0"):
|
|
v = v[:-2]
|
|
return v
|
|
|
|
|
|
def dec(v):
|
|
v = s(v)
|
|
if v is None:
|
|
return None
|
|
try:
|
|
return Decimal(v.replace(",", ""))
|
|
except (InvalidOperation, ValueError):
|
|
return None
|
|
|
|
|
|
def dt(v):
|
|
v = s(v)
|
|
if v is None:
|
|
return None
|
|
d = pd.to_datetime(v, errors="coerce")
|
|
return None if pd.isna(d) else d.to_pydatetime()
|
|
|
|
|
|
def cur(v):
|
|
v = (s(v) or "").upper()
|
|
if v.startswith("DOL") or v.startswith("USD") or "DOLLAR" in v:
|
|
return "USD"
|
|
if v.startswith("PES") or v.startswith("MXN") or v.startswith("MN"):
|
|
return "MXN"
|
|
return "MXN"
|
|
|
|
|
|
def truthy(v):
|
|
return (s(v) or "0").lower() in {"1", "-1", "true", "si", "sí", "yes", "x"}
|
|
|
|
|
|
# --- per-table config ------------------------------------------------------- #
|
|
# fields: policy column -> source column. installments: list of slot dicts.
|
|
# vehicles: 'trip_underscore' | 'single' | 'mca2' | None. drivers: 'mca2' |
|
|
# 'licencias' | None.
|
|
HOME_INST = [
|
|
dict(seq=1, amt="c_1er_pago", cu="moned", d="fecha_pago", ck="no_cheque", cash="efectivo"),
|
|
dict(seq=2, amt="pago_subsec", cu="moned_2", d="fecha_pago_2", ck="no_cheque_2", cash="efectivo_2"),
|
|
dict(seq=3, amt="pago_subsec", cu="moned_3", d="fecha_pago_3", ck="no_cheque_3", cash="efectivo_3"),
|
|
dict(seq=4, amt="pago_subsec", cu="moned_4", d="fecha_pago_4", ck="no_cheque_4", cash="efectivo_4"),
|
|
]
|
|
HOME_FIELDS = dict(polno="no_poliza", agent="agent", comp="comp", desde="desde", hasta="hasta",
|
|
forma="forma_pago", curcol="moned", pneta="p_neta", dpol="d_pol", com="com",
|
|
liquidada="liquidada", numliq="num_liquidacion", fliq="f_liquida1", renov="renovacion")
|
|
AUTO_SINGLE_INST = [dict(seq=1, amt="total", cu="moneda", d="fecha_pago", ck="no_cheque", cash="efectivo")]
|
|
|
|
CONFIGS = {
|
|
"incendio": dict(ptype="INCENDIO", idcol="num_id", fields={**HOME_FIELDS, "curcol": "moneda"},
|
|
inst=HOME_INST, veh=None, drv=None),
|
|
"mult": dict(ptype="MULT", idcol="num_id", fields=HOME_FIELDS, inst=HOME_INST, veh=None, drv=None),
|
|
"m_empr": dict(ptype="M_EMPR", idcol="num_id", fields=HOME_FIELDS, inst=HOME_INST, veh=None, drv=None),
|
|
"tabla_autos": dict(ptype="AUTO", idcol="numer_id",
|
|
fields=dict(polno="no_poliza", agent="agent", comp="comp", desde="desde",
|
|
hasta="hasta", forma="forma_pago", curcol="moneda", pneta="prima_neta",
|
|
dpol="d_poliza", total="total", liquidada="liquidada",
|
|
numliq="num_liquidacion", fliq="f_liquida1", renov="renovacion"),
|
|
inst=AUTO_SINGLE_INST, veh="trip_underscore", drv=None),
|
|
"tabla_autos_ampl": dict(ptype="AUTO", idcol="numer_id",
|
|
fields=dict(polno="no_poliza", agent="agent", comp="comp", desde="desde",
|
|
hasta="hasta", forma="forma_pago", curcol="moneda", pneta="prima_neta",
|
|
dpol="d_poliza", total="total", liquidada="liquidada",
|
|
numliq="num_liquidacion", fliq="f_liquida1", renov="renovacion"),
|
|
inst=AUTO_SINGLE_INST, veh="single", drv=None),
|
|
"tabla_autos_limit": dict(ptype="AUTO", idcol="numer_id",
|
|
fields=dict(polno="no_poliza", agent="agent", comp="comp", desde="desde",
|
|
hasta="hasta", forma="forma_pago", curcol="moneda", pneta="prima_neta",
|
|
dpol="d_poliza", total="total", liquidada="liquidada", renov="renovacion"),
|
|
inst=AUTO_SINGLE_INST, veh="single", drv=None),
|
|
"tabla_autos_ampl_r": dict(ptype="AUTO", idcol="numer_id",
|
|
fields=dict(polno="no_poliza", agent="agent", comp="comp", desde="desde",
|
|
hasta="hasta", forma="forma_pago", curcol="moneda", pneta="prima_neta",
|
|
dpol="d_poliza", com="com", liquidada="liquidada",
|
|
numliq="num_liquidacion", fliq="f_liquida1", renov="renovacion"),
|
|
inst=[dict(seq=1, amt="c_1er_pago", cu="moneda", d="fecha_pago", ck="no_cheque", cash="efectivo"),
|
|
dict(seq=2, amt="pago_subsec", cu="moned_2", d="fecha_pago_2", ck="no_cheque_2", cash="efectivo_2")],
|
|
veh="single", drv=None),
|
|
"tabla_autos_rc_r": dict(ptype="AUTO", idcol="numer_id",
|
|
fields=dict(polno="no_poliza", agent="agent", comp="comp", desde="desde",
|
|
hasta="hasta", forma="forma_pago", curcol="moneda", pneta="prima_neta",
|
|
dpol="d_poliza", com="com", liquidada="liquidada",
|
|
numliq="num_liquidacion", fliq="f_liquida1", renov="renovacion"),
|
|
inst=[dict(seq=1, amt="c_1er_pago", cu="moneda", d="fecha_pago", ck="no_cheque", cash="efectivo"),
|
|
dict(seq=2, amt="pago_subsec", cu="moned_2", d="fecha_pago_2", ck="no_cheque_2", cash="efectivo_2")],
|
|
veh="single", drv=None),
|
|
"mca2": dict(ptype="AUTO", idcol="numer_id",
|
|
fields=dict(polno="poliza", agent="agent", comp="comp", desde="desde", hasta="hasta",
|
|
forma="forma_pago", curcol="moned", pneta="prima1", dpol="d_poliza1",
|
|
liquidada="liquidada", numliq="nuliquida", fliq="f_liquida1", renov="renovacion"),
|
|
inst=[dict(seq=1, amt="c_1er_pago", cu="moned", d="fecha_pago", ck="no_cheque", cash="efectivo")],
|
|
veh="mca2", drv="mca2"),
|
|
"licencias": dict(ptype="LICENCIAS", idcol="numer_id",
|
|
fields=dict(polno="no_poliza", agent="agent", comp="comp", desde="desde", hasta="hasta",
|
|
forma="forma_pago", curcol="moneda", pneta="prima_neta", dpol="d_poliza",
|
|
total="total", liquidada="liquidada", numliq="num_liquidacion",
|
|
fliq="f_liquida1", renov="renovacion"),
|
|
inst=[dict(seq=1, amt="total", cu="moneda", d="fecha_pago", ck="no_cheque", cash="efectivo")],
|
|
veh=None, drv="licencias"),
|
|
}
|
|
|
|
|
|
def load(name):
|
|
df = pd.read_parquet(STG / f"{name}.parquet").sort_values("_row_num").reset_index(drop=True)
|
|
df = df[[c for c in df.columns if c != "_legacy_source_table"]].copy()
|
|
for c in df.columns:
|
|
if c != "_row_num":
|
|
df[c] = df[c].astype("string").str.strip()
|
|
return df
|
|
|
|
|
|
def main():
|
|
env, sync_mode = parse_mode()
|
|
conn = connect(env)
|
|
print(f"[policies] target env: {env}")
|
|
c = conn.cursor()
|
|
|
|
c.execute("SELECT legacyId, customerId FROM customer_legacy_refs WHERE sourceSystem='insurance'")
|
|
cust = {r[0]: r[1] for r in c.fetchall()}
|
|
|
|
# Sync mode reuses each legacy policy's existing id (keyed by provenance) so
|
|
# its PK is stable and every child row built below points at the right
|
|
# parent. New legacy policies fall through to a fresh uuid.
|
|
existing_pol = existing_ids(
|
|
c, "policies", ("legacySourceDb", "legacySourceTable", "legacyId"),
|
|
"WHERE legacyId IS NOT NULL") if sync_mode else {}
|
|
pol_keys: set = set()
|
|
|
|
policies, insts, vehicles, drivers = [], [], [], []
|
|
polno_to_id = {} # policy number -> a policyId (for BENEF/DATOS linking)
|
|
providers, ptypes = set(), set()
|
|
skipped = 0
|
|
|
|
for table, cfg in CONFIGS.items():
|
|
df = load(table)
|
|
F = cfg["fields"]
|
|
ptypes.add(cfg["ptype"])
|
|
consumed = {cfg["idcol"], *F.values()}
|
|
for slot in cfg["inst"]:
|
|
consumed |= {slot["amt"], slot["cu"], slot["d"], slot["ck"], slot["cash"]}
|
|
|
|
for _, row in df.iterrows():
|
|
cid = cust.get(norm_id(row[cfg["idcol"]]))
|
|
if not cid:
|
|
skipped += 1
|
|
continue
|
|
legacy_pid = str(int(row["_row_num"]))
|
|
pkey = (LEGACY_DB, table, legacy_pid)
|
|
pol_keys.add(pkey)
|
|
pid = existing_pol.get(pkey) or str(uuid.uuid4())
|
|
comp = s(row.get(F.get("comp", ""), None)) if F.get("comp") else None
|
|
if comp:
|
|
providers.add(comp)
|
|
polno = s(row.get(F["polno"])) if F.get("polno") else None
|
|
if polno:
|
|
polno_to_id[polno] = pid
|
|
|
|
# coverages = everything not explicitly consumed / address / doc-ish
|
|
cov = {}
|
|
for col in df.columns:
|
|
if col in consumed or col in _ADDR:
|
|
continue
|
|
if col.startswith(("marca", "modelo", "carroceria", "motor", "placa", "clase",
|
|
"auto_id", "a_o", "edo", "name", "birth", "edad", "lic",
|
|
"ocup", "sex", "nombre_aseg_", "docs_", "foto")):
|
|
continue
|
|
val = s(row[col])
|
|
if val is not None:
|
|
cov[col] = val
|
|
|
|
policies.append((
|
|
pid, polno or "(SIN POLIZA)", cid, cfg["ptype"], comp,
|
|
s(row.get(F.get("agent", ""))) if F.get("agent") else None,
|
|
dt(row.get(F.get("desde", ""))) if F.get("desde") else None,
|
|
dt(row.get(F.get("desde", ""))) if F.get("desde") else None,
|
|
dt(row.get(F.get("hasta", ""))) if F.get("hasta") else None,
|
|
dec(row.get(F.get("pneta", ""))) if F.get("pneta") else None,
|
|
dec(row.get(F.get("dpol", ""))) if F.get("dpol") else None,
|
|
dec(row.get(F.get("com", ""))) if F.get("com") else None,
|
|
dec(row.get(F.get("total", ""))) if F.get("total") else None,
|
|
cur(row.get(F.get("curcol", ""))) if F.get("curcol") else "MXN",
|
|
s(row.get("observaciones")),
|
|
json.dumps(cov, ensure_ascii=False) if cov else None,
|
|
1 if truthy(row.get(F.get("liquidada", ""))) else 0,
|
|
s(row.get(F.get("numliq", ""))) if F.get("numliq") else None,
|
|
dt(row.get(F.get("fliq", ""))) if F.get("fliq") else None,
|
|
LEGACY_DB, table, str(int(row["_row_num"])), NOW,
|
|
))
|
|
|
|
# installments
|
|
for slot in cfg["inst"]:
|
|
amt = dec(row.get(slot["amt"]))
|
|
pdate = dt(row.get(slot["d"]))
|
|
if amt is None and pdate is None:
|
|
continue
|
|
insts.append((str(uuid.uuid4()), pid, slot["seq"], amt,
|
|
cur(row.get(slot["cu"])), pdate, s(row.get(slot["ck"])),
|
|
1 if truthy(row.get(slot["cash"])) else 0))
|
|
|
|
# vehicles
|
|
def add_vehicle(make, model, body, engine, plate, year=None, state=None):
|
|
if not any([make, model, plate, engine]):
|
|
return
|
|
vehicles.append((str(uuid.uuid4()), cid, pid, make, model, year, body,
|
|
engine, plate, state, table, str(int(row["_row_num"]))))
|
|
if cfg["veh"] == "trip_underscore":
|
|
for i in ("1", "2", "3"):
|
|
add_vehicle(s(row.get(f"marca_{i}")), s(row.get(f"modelo_{i}")),
|
|
s(row.get(f"carroceria_{i}")), s(row.get(f"motor_{i}")),
|
|
s(row.get(f"placa_{i}")))
|
|
elif cfg["veh"] == "single":
|
|
add_vehicle(s(row.get("marca")), s(row.get("modelo")), s(row.get("carroceria")),
|
|
s(row.get("motor")), s(row.get("placa")))
|
|
elif cfg["veh"] == "mca2":
|
|
for i in ("1", "2", "3"):
|
|
add_vehicle(s(row.get(f"marca{i}")), s(row.get(f"modelo{i}")),
|
|
s(row.get(f"clase{i}")), s(row.get(f"auto_id{i}")),
|
|
s(row.get(f"placa{i}")), year=s(row.get(f"a_o{i}")),
|
|
state=s(row.get(f"edo{i}")))
|
|
|
|
# drivers
|
|
def add_driver(name, birth=None, sex=None, occ=None, lic=None, lstate=None):
|
|
if not name and not lic:
|
|
return
|
|
drivers.append((str(uuid.uuid4()), pid, name, dt(birth), sex, occ, lic, lstate))
|
|
if cfg["drv"] == "mca2":
|
|
for i in ("1", "2", "3"):
|
|
add_driver(s(row.get(f"name{i}")), s(row.get(f"birth{i}")),
|
|
s(row.get(f"sex{i}")), s(row.get(f"ocup{i}")),
|
|
s(row.get(f"lic{i}")), s(row.get(f"licedo{i}")))
|
|
elif cfg["drv"] == "licencias":
|
|
for i in ("1", "2", "3"):
|
|
add_driver(s(row.get(f"nombre_aseg_{i}")), lic=s(row.get(f"lic_{i}")),
|
|
lstate=s(row.get(f"edo_{i}")))
|
|
|
|
# beneficiaries (BENEF -> by policy number)
|
|
benef = load("benef")
|
|
benes = []
|
|
for _, r in benef.iterrows():
|
|
pid = polno_to_id.get(s(r["bpoliza"]))
|
|
if pid:
|
|
benes.append((str(uuid.uuid4()), pid, s(r["bnombre"]), s(r["bdireccion"]),
|
|
s(r["btel"]), s(r["bemail"])))
|
|
|
|
# adjusters (AJUSTADORES + ATLAS)
|
|
adj_rows, adj_by_name = [], {}
|
|
for t in ("ajustadores", "ajustadoresatlas"):
|
|
for _, r in load(t).iterrows():
|
|
aid = str(uuid.uuid4())
|
|
name = s(r["nombre"])
|
|
adj_rows.append((aid, s(r["comp"]), s(r["ciudad"]), name, s(r["tel"]), s(r["beeper"])))
|
|
if name:
|
|
adj_by_name.setdefault(name.upper(), aid)
|
|
|
|
# claims (DATOS)
|
|
claims = []
|
|
for _, r in load("datos").iterrows():
|
|
pid = polno_to_id.get(s(r["poliza"]))
|
|
if not pid:
|
|
continue
|
|
adjid = adj_by_name.get((s(r["ajustador"]) or "").upper())
|
|
claims.append((str(uuid.uuid4()), pid, s(r["tipo"]), dt(r["fecha_siniestro"]),
|
|
dt(r["fecha_reportado"]), s(r["descripcion"]), adjid,
|
|
dec(r["cantidad_reclamada"]), dec(r["cantidad_pactada"]),
|
|
dt(r["fecha_cheque"]), s(r["num_cheque"]),
|
|
1 if truthy(r["concluido"]) else 0, s(r["resolucion"])))
|
|
|
|
pol_cols = ("id,policyNumber,customerId,policyTypeId,insuranceProviderId,agentName,policyDate,"
|
|
"policyFrom,policyTo,netPremium,policyFee,commission,total,currency,observations,"
|
|
"coveragesJson,liquidated,liquidationNumber,liquidationDate,legacySourceDb,"
|
|
"legacySourceTable,legacyId,updatedAt")
|
|
ph = ",".join(["%s"] * 23)
|
|
pol_upsert = (
|
|
f"INSERT INTO policies ({pol_cols}) VALUES ({ph}) ON DUPLICATE KEY UPDATE "
|
|
"customerId=VALUES(customerId),policyNumber=VALUES(policyNumber),policyTypeId=VALUES(policyTypeId),"
|
|
"insuranceProviderId=VALUES(insuranceProviderId),agentName=VALUES(agentName),policyDate=VALUES(policyDate),"
|
|
"policyFrom=VALUES(policyFrom),policyTo=VALUES(policyTo),netPremium=VALUES(netPremium),policyFee=VALUES(policyFee),"
|
|
"commission=VALUES(commission),total=VALUES(total),currency=VALUES(currency),observations=VALUES(observations),"
|
|
"coveragesJson=VALUES(coveragesJson),liquidated=VALUES(liquidated),liquidationNumber=VALUES(liquidationNumber),"
|
|
"liquidationDate=VALUES(liquidationDate),updatedAt=VALUES(updatedAt),archivedAt=NULL")
|
|
|
|
if sync_mode:
|
|
# policy_types / providers: upsert by their name unique, keep ids stable.
|
|
c.execute("SELECT name,id FROM policy_types")
|
|
ptype_ids = dict(c.fetchall())
|
|
for n in ptypes:
|
|
ptype_ids.setdefault(n, str(uuid.uuid4()))
|
|
c.executemany("INSERT INTO policy_types (id,name) VALUES (%s,%s) ON DUPLICATE KEY UPDATE name=VALUES(name)", [(i, n) for n, i in ptype_ids.items()])
|
|
c.execute("SELECT name,id FROM insurance_providers")
|
|
prov_ids = dict(c.fetchall())
|
|
for n in providers:
|
|
prov_ids.setdefault(n, str(uuid.uuid4()))
|
|
c.executemany("INSERT INTO insurance_providers (id,name) VALUES (%s,%s) ON DUPLICATE KEY UPDATE name=VALUES(name)", [(i, n) for n, i in prov_ids.items()])
|
|
|
|
# Adjusters carry no provenance key, so they can't be upserted by one.
|
|
# Resolve claims against the adjusters already in the DB (manual + prior
|
|
# loads), inserting only names not present yet — keeps manual adjusters
|
|
# and every claim's adjusterId FK valid.
|
|
c.execute("SELECT id,name FROM adjusters")
|
|
db_adj = {(nm or "").upper(): i for i, nm in c.fetchall()}
|
|
adj_id_name = {aid: (nm or "").upper() for aid, comp, city, nm, tel, bp in adj_rows}
|
|
new_adj = []
|
|
for aid, comp, city, nm, tel, bp in adj_rows:
|
|
if nm and nm.upper() not in db_adj:
|
|
db_adj[nm.upper()] = aid
|
|
new_adj.append((aid, comp, city, nm, tel, bp))
|
|
if new_adj:
|
|
c.executemany("INSERT INTO adjusters (id,company,city,name,phone,beeper) VALUES (%s,%s,%s,%s,%s,%s)", new_adj)
|
|
claims = [(cl[0], cl[1], *cl[2:6], db_adj.get(adj_id_name.get(cl[6])) if cl[6] else None, *cl[7:]) for cl in claims]
|
|
|
|
# Rebuild every child of a legacy-owned policy before re-inserting the
|
|
# children below (manual rows survive: vehicles by their own null
|
|
# provenance, the rest by their parent policy's null provenance).
|
|
c.execute("DELETE FROM vehicles WHERE legacyId IS NOT NULL")
|
|
for tbl in ("policy_payment_installments", "insured_drivers", "policy_beneficiaries", "claims"):
|
|
c.execute(f"DELETE ch FROM {tbl} ch JOIN policies p ON p.id=ch.policyId WHERE p.legacyId IS NOT NULL")
|
|
|
|
pol_rows = [tuple([p[0], p[1], p[2], ptype_ids.get(p[3]), prov_ids.get(p[4]), *p[5:]]) for p in policies]
|
|
c.executemany(pol_upsert, pol_rows)
|
|
# Drop legacy policies that vanished from source (children already gone).
|
|
delete_missing(c, "policies", ("legacySourceDb", "legacySourceTable", "legacyId"), pol_keys, "WHERE legacyId IS NOT NULL")
|
|
else:
|
|
c.execute("SET FOREIGN_KEY_CHECKS=0")
|
|
for t in ("policy_payment_installments", "vehicles", "insured_drivers",
|
|
"policy_beneficiaries", "claims", "adjusters",
|
|
"policies", "policy_types", "insurance_providers"):
|
|
c.execute(f"TRUNCATE TABLE {t}")
|
|
c.execute("SET FOREIGN_KEY_CHECKS=1")
|
|
ptype_ids = {n: str(uuid.uuid4()) for n in ptypes}
|
|
c.executemany("INSERT INTO policy_types (id,name) VALUES (%s,%s)", [(i, n) for n, i in ptype_ids.items()])
|
|
prov_ids = {n: str(uuid.uuid4()) for n in providers}
|
|
c.executemany("INSERT INTO insurance_providers (id,name) VALUES (%s,%s)", [(i, n) for n, i in prov_ids.items()])
|
|
c.executemany("INSERT INTO adjusters (id,company,city,name,phone,beeper) VALUES (%s,%s,%s,%s,%s,%s)", adj_rows)
|
|
pol_rows = [tuple([p[0], p[1], p[2], ptype_ids.get(p[3]), prov_ids.get(p[4]), *p[5:]]) for p in policies]
|
|
c.executemany(f"INSERT INTO policies ({pol_cols}) VALUES ({ph})", pol_rows)
|
|
|
|
|
|
|
|
c.executemany("INSERT INTO policy_payment_installments "
|
|
"(id,policyId,sequence,amount,currency,paidDate,checkNumber,isCash) "
|
|
"VALUES (%s,%s,%s,%s,%s,%s,%s,%s)", insts)
|
|
c.executemany("INSERT INTO vehicles (id,customerId,policyId,make,model,modelYear,bodyType,"
|
|
"engineNumber,licensePlate,stateCode,legacySourceTable,legacyId) "
|
|
"VALUES (%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s)", vehicles)
|
|
c.executemany("INSERT INTO insured_drivers (id,policyId,fullName,birthDate,sex,occupation,"
|
|
"licenseNumber,licenseState) VALUES (%s,%s,%s,%s,%s,%s,%s,%s)", drivers)
|
|
c.executemany("INSERT INTO policy_beneficiaries (id,policyId,name,address,phone,email) "
|
|
"VALUES (%s,%s,%s,%s,%s,%s)", benes)
|
|
c.executemany("INSERT INTO claims (id,policyId,claimType,incidentDate,reportedDate,description,"
|
|
"adjusterId,claimedAmount,settledAmount,settlementDate,checkNumber,resolved,"
|
|
"resolutionNotes) VALUES (%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s)", claims)
|
|
conn.commit()
|
|
|
|
# --- report ---
|
|
def count(t):
|
|
c.execute(f"SELECT COUNT(*) FROM {t}"); return c.fetchone()[0]
|
|
c.execute("SELECT pt.name, COUNT(*) FROM policies p JOIN policy_types pt ON p.policyTypeId=pt.id "
|
|
"GROUP BY pt.name ORDER BY 2 DESC")
|
|
by_type = c.fetchall()
|
|
c.execute("SELECT COUNT(*) FROM policies p LEFT JOIN customers c ON p.customerId=c.id WHERE c.id IS NULL")
|
|
orphans = c.fetchone()[0]
|
|
print("=== Policies load complete ===")
|
|
print(f" skipped (unresolved customer): {skipped}")
|
|
print(f" -> policies : {count('policies')}")
|
|
for n, k in by_type:
|
|
print(f" {n:12} {k}")
|
|
print(f" -> payment_installments : {count('policy_payment_installments')}")
|
|
print(f" -> vehicles : {count('vehicles')}")
|
|
print(f" -> insured_drivers : {count('insured_drivers')}")
|
|
print(f" -> policy_beneficiaries : {count('policy_beneficiaries')}")
|
|
print(f" -> claims : {count('claims')}")
|
|
print(f" -> policy_types : {count('policy_types')}")
|
|
print(f" -> insurance_providers : {count('insurance_providers')}")
|
|
print(f" -> adjusters : {count('adjusters')}")
|
|
print(f" orphan policies (bad customer FK): {orphans}")
|
|
assert orphans == 0, "policy customer FK invariant failed"
|
|
print(" validation: OK")
|
|
conn.close()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|