Calidad de datos y catálogo

"""#17 Framework de calidad de datos y catálogo automático sobre el DWH del proyecto #16.

Reglas declarativas (completitud, unicidad, rango, catálogo, integridad referencial y
consistencia) ejecutadas como SQL, con semáforo, diccionario de datos generado y clasificación
automática de columnas sensibles (PII). Es gobernanza aplicada: cada regla es un control documentado.
"""
import re

import duckdb
import numpy as np
import pandas as pd

from common import bars, export, table
from p16_dwh_snowflake import DB, run as construir
from validacion import model_card

# (id, tabla, dimensión, descripción, SQL que cuenta filas que VIOLAN la regla, umbral % tolerado)
REGLAS = [
    ("C01", "dim_cliente", "Completitud", "email no nulo", "email IS NULL", 0.5),
    ("C02", "dim_cliente", "Completitud", "estado no nulo", "estado IS NULL", 0.0),
    ("U01", "dim_cliente", "Unicidad", "id_cliente único",
     "id_cliente IN (SELECT id_cliente FROM dim_cliente GROUP BY 1 HAVING COUNT(*) > 1)", 0.0),
    ("R01", "dim_cliente", "Rango", "edad entre 18 y 90", "edad NOT BETWEEN 18 AND 90", 0.0),
    ("R02", "dim_cliente", "Rango", "score de buró entre 300 y 850", "score_buro NOT BETWEEN 300 AND 850", 0.0),
    ("R03", "fact_credito", "Rango", "monto original positivo", "monto_original <= 0", 0.0),
    ("R04", "fact_credito", "Catálogo", "plazo en {24, 36, 48, 60}", "plazo_meses NOT IN (24, 36, 48, 60)", 0.0),
    ("I01", "fact_credito", "Integridad", "todo crédito tiene cliente",
     "id_cliente NOT IN (SELECT id_cliente FROM dim_cliente)", 0.0),
    ("I02", "fact_poliza", "Integridad", "toda póliza tiene crédito",
     "id_credito NOT IN (SELECT id_credito FROM fact_credito)", 0.0),
    ("K01", "fact_pago", "Catálogo", "días de atraso en múltiplos de 30", "dias_atraso % 30 <> 0", 0.0),
    ("K02", "fact_pago", "Consistencia", "saldo no negativo", "saldo_insoluto < 0", 0.0),
    ("K03", "fact_pago", "Consistencia", "si hay atraso no hay pago registrado",
     "dias_atraso > 0 AND pago_realizado > 0", 0.1),
]

PII = {  # patrón en nombre de columna → categoría de dato personal
    r"nombre": "Nombre (PII directa)", r"email|correo": "Correo (PII directa)",
    r"telefono|celular": "Teléfono (PII directa)", r"rfc|curp": "Identificador fiscal (PII directa)",
    r"ingreso|score": "Financiero sensible", r"edad|estado": "Cuasi-identificador",
}


def semaforo(pct, umbral):
    """Verde dentro del umbral; amarillo hasta 5× el umbral; rojo si es de tolerancia cero o lo rebasa."""
    if pct <= umbral:
        return "Verde"
    return "Amarillo" if pct <= umbral * 5 else "Rojo"


CAPA_CRUDA = {"dim_cliente": "stg_cliente", "fact_credito": "stg_credito", "fact_pago": "stg_pago",
              "fact_poliza": "stg_poliza"}


def evaluar(con, capa: str) -> pd.DataFrame:
    filas = []
    for rid, tabla, dim, desc, cond, umbral in REGLAS:
        t = CAPA_CRUDA.get(tabla, tabla) if capa == "Staging" else tabla
        c = cond
        if capa == "Staging":
            for cur, raw in CAPA_CRUDA.items():
                c = c.replace(cur, raw)
        total = con.execute(f"SELECT COUNT(*) FROM {t}").fetchone()[0]
        malas = con.execute(f"SELECT COUNT(*) FROM {t} WHERE {c}").fetchone()[0]
        pct = malas / total * 100
        filas.append([rid, t, dim, desc, "Crítica" if umbral == 0 else "Advertencia", total, malas,
                      round(pct, 3), umbral, semaforo(pct, umbral)])
    return pd.DataFrame(filas, columns=["Regla", "Tabla", "Dimensión", "Descripción", "Severidad", "Filas",
                                        "Violaciones", "% violación", "Umbral %", "Semáforo"])


def indice(res: pd.DataFrame) -> float:
    """Índice 0–100: verde = 1, amarillo = 0.5, rojo = 0; las reglas críticas pesan el doble."""
    puntos = res["Semáforo"].map({"Verde": 1, "Amarillo": 0.5, "Rojo": 0})
    peso = res["Severidad"].map({"Crítica": 2, "Advertencia": 1})
    return float((puntos * peso).sum() / peso.sum() * 100)


def enmascarar(col: str, v) -> str:
    """Enmascaramiento por rol: el analista ve la forma del dato, no el dato."""
    if v is None or (isinstance(v, float) and np.isnan(v)):
        return "—"
    v = str(v)
    if "email" in col:
        u, _, d = v.partition("@")
        return f"{u[0]}***@{d}"
    if "telefono" in col:
        return "******" + v[-4:]
    if "nombre" in col:
        return " ".join(p[0] + "." for p in v.split())
    return v


def run():
    if not DB.exists():
        construir()
    con = duckdb.connect(str(DB), read_only=True)
    crudo, res = evaluar(con, "Staging"), evaluar(con, "Curada")
    ind_crudo, ind_cur = indice(crudo), indice(res)

    # Diccionario de datos automático + clasificación PII (capa curada)
    dicc = []
    for t in ["dim_cliente", "dim_producto", "fact_credito", "fact_pago", "fact_poliza"]:
        info = con.execute(f"DESCRIBE {t}").df()
        stats = con.execute(f"SUMMARIZE {t}").df().set_index("column_name")
        for _, c in info.iterrows():
            cat = next((v for k, v in PII.items() if re.search(k, c.column_name)), "No sensible")
            dicc.append([t, c.column_name, c.column_type, f"{stats.loc[c.column_name, 'null_percentage']}",
                         int(stats.loc[c.column_name, "approx_unique"]), cat])
    dicc = pd.DataFrame(dicc, columns=["Tabla", "Columna", "Tipo", "% nulos", "Valores distintos (aprox.)",
                                       "Clasificación"])
    muestra = con.execute("SELECT id_cliente, nombre, email, telefono, estado, edad FROM dim_cliente LIMIT 5").df()
    mascara = muestra.apply(lambda col: col.map(lambda v: enmascarar(col.name, v)))
    # Perfilado: cuantiles de las variables numéricas clave (detecta valores imposibles de un vistazo)
    perfil = con.execute("""
        SELECT 'edad (staging)' variable, MIN(edad) minimo, QUANTILE_CONT(edad, 0.5) mediana, MAX(edad) maximo,
               AVG(CASE WHEN edad IS NULL THEN 1 ELSE 0 END) * 100 pct_nulos FROM stg_cliente
        UNION ALL SELECT 'edad (curada)', MIN(edad), QUANTILE_CONT(edad, 0.5), MAX(edad),
               AVG(CASE WHEN edad IS NULL THEN 1 ELSE 0 END) * 100 FROM dim_cliente
        UNION ALL SELECT 'monto_original (staging)', MIN(monto_original), QUANTILE_CONT(monto_original, 0.5),
               MAX(monto_original), 0 FROM stg_credito
        UNION ALL SELECT 'monto_original (curada)', MIN(monto_original), QUANTILE_CONT(monto_original, 0.5),
               MAX(monto_original), 0 FROM fact_credito""").df().round(2)
    perfil.columns = ["Variable", "Mínimo", "Mediana", "Máximo", "% nulos"]
    con.close()

    conteo = res["Semáforo"].value_counts().reindex(["Verde", "Amarillo", "Rojo"], fill_value=0)
    conteo_c = crudo["Semáforo"].value_counts().reindex(["Verde", "Amarillo", "Rojo"], fill_value=0)
    pii = dicc[dicc.Clasificación != "No sensible"]
    comp = pd.DataFrame({"Regla": res.Regla, "Descripción": res.Descripción, "Severidad": res.Severidad,
                         "Staging %": crudo["% violación"], "Staging": crudo.Semáforo,
                         "Curada %": res["% violación"], "Curada": res.Semáforo})
    result = {
        "title": "Calidad de datos y catálogo (gobernanza)",
        "summary": [
            f"{len(res)} reglas declarativas en 6 dimensiones, con severidad y umbral, evaluadas en las dos capas del DWH.",
            f"Staging (fuente): {conteo_c['Verde']} verdes, {conteo_c['Amarillo']} amarillas y {conteo_c['Rojo']} rojas — "
            f"índice {ind_crudo:.0f}/100. Capa curada: {conteo['Verde']} verdes, {conteo['Rojo']} rojas — índice {ind_cur:.0f}/100.",
            "Lo que no se puede corregir no se borra: se manda a cuarentena con su motivo (proyecto #16), y lo que "
            "queda en amarillo en la capa curada (correos faltantes) se reporta al dueño del dato en la fuente.",
            f"Catálogo automático de {len(dicc)} columnas; {len(pii)} sensibles (PII directa, financiera o "
            "cuasi-identificador) con enmascaramiento por rol para analistas.",
        ],
        "metrics": [
            {"label": "Índice staging", "value": f"{ind_crudo:.0f} / 100"},
            {"label": "Índice curada", "value": f"{ind_cur:.0f} / 100"},
            {"label": "Rojas staging → curada", "value": f"{conteo_c['Rojo']} → {conteo['Rojo']}"},
            {"label": "Columnas PII", "value": str(len(pii))},
        ],
        "charts": [
            {"id": "semaforo", "type": "status", "title": "Semáforo en la capa curada",
             "items": [{"label": f"{r['Regla']} · {r['Descripción']} ({r['Severidad'].lower()})", "status": r["Semáforo"],
                        "value": f"{r['% violación']:.2f} % (umbral {r['Umbral %']} %)"} for _, r in res.iterrows()]},
            {"id": "semaforo_crudo", "type": "status", "title": "Semáforo en staging (fuente)",
             "items": [{"label": f"{r['Regla']} · {r['Descripción']}", "status": r["Semáforo"],
                        "value": f"{r['% violación']:.2f} %"} for _, r in crudo.iterrows()]},
            bars("indice", "Índice de calidad por capa (0–100)", ["Staging", "Curada"], [round(ind_crudo, 1), round(ind_cur, 1)],
                 fmt="num", subtitle="Verde = 1, amarillo = 0.5, rojo = 0; críticas pesan doble"),
        ],
        "tables": [table("Reglas: staging vs. capa curada", comp),
                   table("Perfilado de variables clave", perfil),
                   table("Vista enmascarada para el rol analista", mascara),
                   table("Diccionario de datos y clasificación PII", dicc)],
        "card": model_card(
            "Reglas de calidad como SQL declarativo con dimensión, severidad, umbral y semáforo; catálogo y "
            "clasificación PII generados del esquema.",
            ["Las mismas reglas sobre staging y capa curada para medir el efecto de la limpieza.",
             "Perfilado de cuantiles para detectar valores imposibles.",
             "Los defectos sembrados en el generador (#16) funcionan como pruebas de que cada regla detecta lo que debe."],
            ["Umbrales definidos con el dueño de cada dato (aquí, supuestos razonables)."],
            ["La clasificación PII por nombre de columna es una primera pasada; se complementa con revisión manual."]),
    }
    export("p17-calidad-datos", result)
    return result


if __name__ == "__main__":
    run()