fn_registry/python/functions/datascience/generate_synthetic_eda_table.py

"""generate_synthetic_eda_table — fixture sintetico para ejercitar el motor AutomaticEDA.

Funcion impura (escribe un archivo DuckDB a disco) y determinista por ``seed``:
construye una unica tabla cuyo CONTENIDO esta disenado para ACTIVAR el maximo
numero de capitulos del motor AutomaticEDA del grupo `eda` (num_distr, cat_distr,
text_distr, correlacion, missingness, modelos, timeseries, geospatial, relaciones,
calidad, agregacion) y los detectores semanticos / PII (`infer_semantic_type`).

Estilo dict-no-throw del grupo `eda`: NUNCA lanza; captura cualquier error y
devuelve ``{"status": "error", "error": str}``.

Determinismo: con el mismo ``seed`` el DataFrame y, por tanto, la tabla DuckDB
resultante son identicos byte a byte. Se siembra Faker (``Faker.seed``) y numpy
(``np.random.default_rng(seed)``) al inicio de cada generacion.
"""

import re

# Lista fija de paises (12 -> cardinalidad media para cat_distr / agregacion).
_COUNTRIES = [
    "ES", "FR", "DE", "IT", "PT", "NL",
    "BE", "US", "GB", "IE", "SE", "PL",
]

# Lista fija de categorias de producto (6 -> cardinalidad media).
_CATEGORIES = [
    "electronics", "clothing", "home", "sports", "books", "toys",
]

# Niveles de plan con probabilidades DESBALANCEADAS (entropia baja para cat_distr).
_PLANS = ["baja", "media", "alta"]
_PLAN_PROBS = [0.70, 0.25, 0.05]

# Centroides (lat, lon) aproximados por pais: muestrean coordenadas validas
# dentro de [-90, 90] x [-180, 180] para que detect_latlon_columns las acepte.
_CENTROIDS = {
    "ES": (40.4, -3.7), "FR": (46.6, 2.2), "DE": (51.1, 10.4), "IT": (41.9, 12.5),
    "PT": (39.4, -8.2), "NL": (52.1, 5.3), "BE": (50.5, 4.5), "US": (39.0, -98.0),
    "GB": (54.0, -2.0), "IE": (53.4, -8.0), "SE": (60.1, 18.6), "PL": (52.0, 19.1),
}

# Locales rotados para generar texto multi-idioma (es/en/fr).
_TEXT_LOCALES = ["es_ES", "en_US", "fr_FR"]

# Identificador SQL valido (DuckDB no parametriza el nombre de tabla en DDL).
_IDENT_RE = re.compile(r"^[A-Za-z_][A-Za-z0-9_]*$")


def _make_fakers(seed):
    """Crea los Faker por locale tras sembrar el generador compartido.

    ``Faker.seed(seed)`` siembra el ``random.Random`` compartido por todas las
    instancias Faker que usan el generador por defecto, asi que el orden de
    llamadas determina por completo la salida (determinismo).
    """
    from faker import Faker

    Faker.seed(seed)
    es_es, en_us, fr_fr = (Faker(loc) for loc in _TEXT_LOCALES)
    return {"es_ES": es_es, "en_US": en_us, "fr_FR": fr_fr}


# Texto duplicado canonico (multi-idioma, > 20 palabras) que se inyecta en una
# fraccion de las filas para que el analisis de duplicados exactos lo detecte.
_DUP_REVIEW = (
    "Servicio excelente y entrega muy rapida, el producto llego en perfecto "
    "estado y coincide con la descripcion publicada en la tienda. The customer "
    "support team answered every question quickly and the packaging was solid "
    "and well protected during shipping. Je recommande vivement ce vendeur a "
    "tous mes amis, la qualite est vraiment au rendez-vous cette fois."
)


def _make_reviews(n, rng, fakers, dup_frac=0.04, null_frac=0.08):
    """Genera ``n`` reviews de texto libre largo multi-idioma (es/en/fr).

    Cada review concatena dos parrafos de Faker en el idioma rotado por fila, de
    modo que la MEDIANA de palabras por documento queda muy por encima de 20 y la
    media de caracteres por encima de 50 (gates del capitulo text_distr). Se
    inyectan duplicados exactos (``dup_frac``) y nulos (``null_frac``).

    Devuelve una ``list`` de ``str`` o ``None`` (nulos) de longitud ``n``.
    """
    # Numero de frases por parrafo precomputado con numpy (determinista) para no
    # interleavar draws de rng dentro del bucle de faker.
    nb1 = rng.integers(4, 8, n)
    nb2 = rng.integers(3, 7, n)

    reviews = []
    for i in range(n):
        fk = fakers[_TEXT_LOCALES[i % 3]]
        p1 = fk.paragraph(nb_sentences=int(nb1[i]))
        p2 = fk.paragraph(nb_sentences=int(nb2[i]))
        reviews.append(f"{p1} {p2}")

    # Duplicados exactos: una fraccion de filas comparte un review identico.
    if n > 0 and dup_frac > 0:
        k_dup = max(1, int(n * dup_frac))
        dup_idx = rng.choice(n, size=min(k_dup, n), replace=False)
        for j in dup_idx:
            reviews[int(j)] = _DUP_REVIEW

    # Nulos MCAR-ish: una fraccion de filas al azar queda en None.
    if n > 0 and null_frac > 0:
        k_null = max(1, int(n * null_frac))
        null_idx = rng.choice(n, size=min(k_null, n), replace=False)
        for j in null_idx:
            reviews[int(j)] = None

    return reviews


def _make_phone_intl(rng):
    """Construye un telefono en formato internacional que casa phone_intl.

    Regex objetivo (fullmatch): ``\\+\\d[\\d\\s()-]{6,}\\d``. Empieza por '+',
    digito, bloques de digitos separados por espacios y termina en digito.
    """
    cc = int(rng.integers(1, 99))
    a = int(rng.integers(100, 999))
    b = int(rng.integers(100, 999))
    c = int(rng.integers(100, 999))
    return f"+{cc} {a} {b} {c}"


def _make_latlon(countries, rng):
    """Devuelve (latitudes, longitudes) muestreando centroides de pais + jitter.

    Mantiene los valores dentro de [-90, 90] y [-180, 180] (validez exigida por
    detect_latlon_columns). El jitter es pequeno para no salirse del rango.
    """
    import numpy as np

    lats = np.empty(len(countries), dtype=float)
    lons = np.empty(len(countries), dtype=float)
    jitter_lat = rng.normal(0.0, 0.5, len(countries))
    jitter_lon = rng.normal(0.0, 0.5, len(countries))
    for i, code in enumerate(countries):
        base_lat, base_lon = _CENTROIDS[code]
        lats[i] = float(np.clip(base_lat + jitter_lat[i], -90.0, 90.0))
        lons[i] = float(np.clip(base_lon + jitter_lon[i], -180.0, 180.0))
    return lats, lons


def _amount_with_outliers(n, rng, n_extreme=6, factor=50.0):
    """Serie lognormal de cola pesada con ~``n_extreme`` outliers altos (x``factor``)."""
    import numpy as np

    amount = rng.lognormal(mean=4.0, sigma=1.0, size=n)
    if n > 0 and n_extreme > 0:
        idx = rng.choice(n, size=min(n_extreme, n), replace=False)
        amount[idx] = amount[idx] * factor
    return amount


def generate_synthetic_eda_table(
    out_db_path, table="synthetic", n_rows=2000, seed=42
):
    """Genera una tabla DuckDB sintetica que activa el maximo de capitulos del EDA.

    Construye un DataFrame de ``n_rows`` clientes unicos con columnas elegidas para
    disparar detectores concretos del motor AutomaticEDA (numericas continuas con
    correlaciones lineal/no-lineal, numericas con outliers, categoricas
    desbalanceadas, texto libre multi-idioma con duplicados, fecha para serie
    temporal, lat/lon validas, semanticos/PII y nulos con patron MCAR/MAR), y la
    materializa en ``out_db_path`` con ``CREATE OR REPLACE TABLE``.

    Funcion impura (escribe a disco) y determinista por ``seed``: con el mismo
    seed la tabla resultante es identica byte a byte. NUNCA lanza.

    Args:
        out_db_path: ruta al archivo DuckDB de salida. Se crea (o reutiliza) y la
            tabla se reemplaza si ya existe.
        table: nombre de la tabla a crear. Se valida contra
            ``^[A-Za-z_][A-Za-z0-9_]*$`` y se cita en el DDL.
        n_rows: numero de filas (clientes unicos). Default 2000.
        seed: semilla para Faker y numpy. Default 42.

    Returns:
        dict dict-no-throw. En exito::

            {"status": "ok", "db_path": out_db_path, "table": table,
             "n_rows": n_rows, "columns": [<nombres de columna>], "seed": seed}

        En error (sin lanzar)::

            {"status": "error", "error": str}
    """
    try:
        import duckdb
        import numpy as np
        import pandas as pd

        if not _IDENT_RE.match(table or ""):
            return {
                "status": "error",
                "error": (
                    f"nombre de tabla invalido: {table!r} "
                    "(debe casar con ^[A-Za-z_][A-Za-z0-9_]*$)"
                ),
            }
        n = int(n_rows)
        if n <= 0:
            return {"status": "error", "error": f"n_rows debe ser > 0, dado {n_rows!r}"}

        fakers = _make_fakers(seed)
        rng = np.random.default_rng(seed)

        # --- Numericas continuas (distinct alto, correlaciones) ---
        income = np.clip(rng.normal(40000.0, 12000.0, n), 1000.0, None)
        spending = income * 0.35 + rng.normal(0.0, 2000.0, n)  # corr POSITIVA fuerte
        age = rng.integers(18, 91, n)
        risk_score = 90.0 - age * 0.7 + rng.normal(0.0, 5.0, n)  # corr NEGATIVA con age
        tenure_months = rng.uniform(0.0, 60.0, n)
        engagement_quad = ((tenure_months - 30.0) ** 2) / 30.0 + rng.normal(0.0, 1.0, n)

        # --- Numericas con outliers claros ---
        amount = _amount_with_outliers(n, rng)
        n_purchases = rng.poisson(3.0, n).astype(float)
        if n > 0:
            k_hi = min(max(1, int(n * 0.002)) + 2, n)  # ~3-5 valores altisimos
            hi_idx = rng.choice(n, size=k_hi, replace=False)
            n_purchases[hi_idx] = rng.integers(200, 400, len(hi_idx)).astype(float)

        # --- Categoricas ---
        country = rng.choice(_COUNTRIES, n)
        category = rng.choice(_CATEGORIES, n)
        plan = rng.choice(_PLANS, n, p=_PLAN_PROBS)

        # --- Texto libre multi-idioma con duplicados ---
        review = _make_reviews(n, rng, fakers)

        # --- Fecha / serie temporal (rango ~2 anios, cadencia ~diaria) ---
        base = np.datetime64("2022-01-01")
        offsets = rng.integers(0, 730, n)
        signup_date = pd.to_datetime(base) + pd.to_timedelta(offsets, unit="D")

        # --- Geo lat/lon validas ---
        latitude, longitude = _make_latlon(country, rng)

        # --- Semanticos / PII (>=80% match para infer_semantic_type) ---
        customer_id = [fakers["en_US"].uuid4() for _ in range(n)]
        email = [fakers["en_US"].email() for _ in range(n)]
        iban = [fakers["en_US"].iban() for _ in range(n)]
        phone = [_make_phone_intl(rng) for _ in range(n)]

        df = pd.DataFrame(
            {
                "customer_id": customer_id,
                "email": email,
                "iban": iban,
                "phone": phone,
                "income": income,
                "spending": spending,
                "age": age,
                "risk_score": risk_score,
                "tenure_months": tenure_months,
                "engagement_quad": engagement_quad,
                "amount": amount,
                "n_purchases": n_purchases,
                "country": country,
                "category": category,
                "plan": plan,
                "review": review,
                "signup_date": signup_date,
                "latitude": latitude,
                "longitude": longitude,
            }
        )

        # --- Nulos con patron ---
        # income + spending faltan JUNTAS en las MISMAS filas (co-ocurrencia -> MAR).
        k_co = max(1, int(n * 0.12))
        co_idx = rng.choice(n, size=min(k_co, n), replace=False)
        df.loc[co_idx, "income"] = np.nan
        df.loc[co_idx, "spending"] = np.nan
        # risk_score falta cuando plan == "alta" (mas una pizca de azar) -> MAR.
        risk_mask = (df["plan"] == "alta").to_numpy() | (rng.random(n) < 0.02)
        df.loc[risk_mask, "risk_score"] = np.nan

        columns = list(df.columns)

        con = duckdb.connect(out_db_path)
        try:
            con.register("df_synth_eda", df)
            con.execute(
                f'CREATE OR REPLACE TABLE "{table}" AS SELECT * FROM df_synth_eda'
            )
            con.unregister("df_synth_eda")
        finally:
            con.close()

        return {
            "status": "ok",
            "db_path": out_db_path,
            "table": table,
            "n_rows": n,
            "columns": columns,
            "seed": seed,
        }
    except Exception as exc:  # noqa: BLE001 — dict-no-throw del grupo eda.
        return {"status": "error", "error": str(exc)}


if __name__ == "__main__":
    import json
    import sys

    args = sys.argv[1:]
    db_path = args[0] if len(args) > 0 else "/tmp/synthetic_eda.duckdb"
    tbl = args[1] if len(args) > 1 else "synthetic"
    rows = int(args[2]) if len(args) > 2 else 2000
    sd = int(args[3]) if len(args) > 3 else 42
    print(json.dumps(generate_synthetic_eda_table(db_path, tbl, rows, sd), indent=2))