rmancinasandClaude Opus 5
48e01ddd21
The capture form only ever had prima neta, derecho de póliza and comisión.
The Access form it replaces has seven figures, and the four that were missing
are the ones that make a policy paid in installments add up.
Adds recargo, IVA, prima total and forma de pago to the policy header, the
same breakdown per installment, and a per-line-of-business IVA rate.
IVA and prima total are the only derived figures:
base = prima neta + recargo + derecho de póliza
IVA = round(base * tasa)
total = base + IVA
The recargo is inside the taxable base. That is not a guess — policy 7006785
prints IVA 52.03 on 610.86 + 8.55 + 31.00, and leaving the recargo out gives
51.35, which matches nothing on the page. Both of its money rows are asserted
in premium.spec.ts. The recargo itself is never derived: the carrier quotes it,
so staff key it in, and the field is disabled on ANNUAL/SINGLE. Both derived
figures are stored rather than recomputed on read, and stay editable, because
the printed policy is the record of truth and a later rate change must not
silently restate what was issued.
The rate lives on PolicyType (seeded to 0.08, editable in Catálogos), which is
the legacy one-row IMPUESTOS / IMPUESTOS_AUTOS tables made configurable. The
rate applied is stamped on the policy so an old one reads back at its original
rate.
Per-installment, not two fixed slots on the header: a policy split into several
exhibiciones prices each payment separately — that is why the Access form drew
the money row twice — and a trimestral policy needs four, which the Access
layout could not hold.
Also fixes two losses in the ETL, which is how these went missing:
- `forma_pago` was marked consumed by the coverage sweep and then never
written to any column, so FORMA PAGO existed nowhere in the platform.
- `recargo` and the whole second money row fell into `coveragesJson` as
loose strings, mislabeled as coverage amounts.
transform_policies.py now writes all of it directly;
backfill_policy_premium_breakdown.py recovers it on a database that must not be
re-imported, and strips the migrated keys back out of coveragesJson. Both are
COALESCE-only, so a figure a human has corrected in the app wins.
IVA and TOTAL are NOT backfilled: they were unbound calculated controls on the
Access form, never columns, so there is nothing to recover and every migrated
policy reads null until it is edited.
The backfill warns on 5 annual policies that carry a non-zero recargo — a
contradiction that predates this change and is left for a human, not silently
corrected.
Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
505 lines
27 KiB
Python
505 lines
27 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.
|
|
- Premium breakdown: a policy paid in more than one exhibicion prices EACH
|
|
payment separately, which is why the home tables carry the whole money row
|
|
twice (p_neta/recargo/d_pol/com and their _2 twins). The unsuffixed set is
|
|
the policy header, the suffixed one belongs to payment 2, and both are
|
|
written per installment as well. This used to be lost: `recargo` and the
|
|
_2 columns fell into coveragesJson as loose strings and forma_pago was
|
|
marked consumed but never written anywhere at all.
|
|
- IVA and the printed TOTAL are NOT in Access for the home tables. They were
|
|
unbound calculated controls on the form, so there is nothing to migrate;
|
|
the app computes them (apps/api/src/policies/premium.ts) from
|
|
(p_neta + recargo + d_pol) * rate.
|
|
- 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"}
|
|
|
|
|
|
# Access FORMA PAGO -> PaymentFrequency. The whole staged corpus holds exactly
|
|
# four spellings (ANNUAL 1851, SEMESTRAL 29, semestral 2, CONTADO 1); anything
|
|
# else is left null rather than guessed, because the value decides whether a
|
|
# recargo is legitimate on the row.
|
|
_FREQ = {
|
|
"ANNUAL": "ANNUAL",
|
|
"ANUAL": "ANNUAL",
|
|
"SEMESTRAL": "SEMIANNUAL",
|
|
"TRIMESTRAL": "QUARTERLY",
|
|
"MENSUAL": "MONTHLY",
|
|
"CONTADO": "SINGLE",
|
|
}
|
|
|
|
|
|
def freq(v):
|
|
return _FREQ.get((s(v) or "").upper())
|
|
|
|
|
|
def slot_dec(row, slot, key):
|
|
"""One figure of a payment's premium breakdown, or None when the source
|
|
table has no such column. Deliberately not zero: a zero d_pol on the second
|
|
payment is a real figure in the books and must stay distinguishable from
|
|
'this table never had that column'."""
|
|
col = slot.get(key)
|
|
return dec(row.get(col)) if col else None
|
|
|
|
|
|
# --- per-table config ------------------------------------------------------- #
|
|
# fields: policy column -> source column. installments: list of slot dicts.
|
|
# vehicles: 'trip_underscore' | 'single' | 'mca2' | None. drivers: 'mca2' |
|
|
# 'licencias' | None.
|
|
# `pneta`/`recarg`/`dpol`/`com` on a slot are that payment's own share of the
|
|
# premium. Only the first two payments have one in Access — the form only ever
|
|
# drew the money row twice — so slots 3 and 4 carry none and keep just the
|
|
# amount actually collected.
|
|
HOME_INST = [
|
|
dict(seq=1, amt="c_1er_pago", cu="moned", d="fecha_pago", ck="no_cheque", cash="efectivo",
|
|
pneta="p_neta", recarg="recargo", dpol="d_pol", com="com"),
|
|
dict(seq=2, amt="pago_subsec", cu="moned_2", d="fecha_pago_2", ck="no_cheque_2", cash="efectivo_2",
|
|
pneta="p_neta_2", recarg="recargo_2", dpol="d_pol_2", com="com_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", recarg="recargo",
|
|
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",
|
|
pneta="prima_neta", dpol="d_poliza")]
|
|
|
|
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", pneta="prima_neta", dpol="d_poliza")],
|
|
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"]}
|
|
# The per-payment premium columns are now modeled, so they must
|
|
# leave the coveragesJson sweep — otherwise every recargo would be
|
|
# written twice, once as a column and once as a fake coverage.
|
|
consumed |= {slot[k] for k in ("pneta", "recarg", "dpol", "com") if slot.get(k)}
|
|
|
|
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("recarg", ""))) if F.get("recarg") 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,
|
|
freq(row.get(F.get("forma", ""))) if F.get("forma") 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
|
|
# Slot breakdown, where the source table has one.
|
|
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,
|
|
slot_dec(row, slot, "pneta"), slot_dec(row, slot, "recarg"),
|
|
slot_dec(row, slot, "dpol"), slot_dec(row, slot, "com")))
|
|
|
|
# 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,surcharge,policyFee,commission,total,paymentFrequency,"
|
|
"currency,observations,"
|
|
"coveragesJson,liquidated,liquidationNumber,liquidationDate,legacySourceDb,"
|
|
"legacySourceTable,legacyId,updatedAt")
|
|
ph = ",".join(["%s"] * 25)
|
|
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),"
|
|
"surcharge=VALUES(surcharge),policyFee=VALUES(policyFee),"
|
|
"commission=VALUES(commission),total=VALUES(total),paymentFrequency=VALUES(paymentFrequency),"
|
|
"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,"
|
|
"netPremium,surcharge,policyFee,commission) "
|
|
"VALUES (%s,%s,%s,%s,%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()
|