"""
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()
