fn_registry/python/functions/datascience/suggest_intratable_fk_candidates.py

"""suggest_intratable_fk_candidates — heuristica de FK intra-tabla del grupo `eda`.

Sobre el TableProfile de UNA tabla (el dict que produce ``profile_table``), sugiere
por heuristica de NOMBRE + CARDINALIDAD que columnas PARECEN una clave foranea hacia
otra tabla, util cuando no hay relaciones inter-tabla disponibles (una sola tabla y,
por tanto, sin containment cruzado que medir). Es una SUGERENCIA, no una afirmacion:
no confirma que exista la tabla referida ni que los valores esten contenidos en ella.

La consume el capitulo RELACIONES de AutomaticEDA cuando solo hay una tabla.

Funcion PURA: solo lee el dict (lectura defensiva con ``.get``), no hace I/O y nunca
lanza por inputs raros (devuelve ``[]``).
"""

# inferred_type que es compatible con una clave foranea (entero/categorico).
_FK_INFERRED_OK = {"numeric", "categorical", "integer"}

# Prefijos de physical_type que admiten ser clave foranea (enteros, texto, uuid).
_FK_PHYSICAL_PREFIXES = (
    "int", "bigint", "smallint", "tinyint", "hugeint", "uint",
    "varchar", "text", "char", "bpchar", "string", "uuid",
)

# Prefijos de physical_type que EXCLUYEN ser clave foranea: medidas en coma flotante
# (float/double/decimal/numeric/real), temporales (date/time/timestamp/interval) y
# boolean. Se comprueban ANTES que las senales positivas (la exclusion gana: una
# columna numeric con physical DOUBLE es una medida, no una FK).
_FK_PHYSICAL_EXCLUDE = (
    "float", "double", "decimal", "numeric", "real",
    "date", "time", "timestamp", "interval",
    "bool",
)


def _fk_name_signal(name):
    """Detecta el sufijo de clave foranea en el nombre y devuelve ``(stem, sufijo)``.

    Reconoce ``<algo>_id`` (snake), ``<Algo>Id`` y ``<algo>ID`` (camel). NO reconoce
    el ``id``/``Id``/``ID`` generico a secas (suele ser la PK propia de la tabla, no
    una referencia). En camelCase la ``I`` mayuscula marca el limite de palabra, asi
    que ``paid``/``valid``/``grid`` (``id`` en minuscula y sin separador) NO matchean.

    El ``stem`` se devuelve en minusculas y sirve de ``ref_table_guess`` (la tabla a
    la que probablemente apunta): ``customer_id`` -> ``"customer"``, ``AlbumId`` ->
    ``"album"``, ``manager_staff_id`` -> ``"manager_staff"``. Devuelve ``None`` si no
    hay senal de nombre.
    """
    if not isinstance(name, str):
        return None
    raw = name.strip()
    if not raw:
        return None
    # Snake: termina en "_id" (indiferente a mayusculas en la parte "id").
    if raw.lower().endswith("_id"):
        stem = raw[:-3].rstrip("_-. ")
        if not stem:
            return None
        return (stem.lower(), "_id")
    # Camel todo-mayuscula: "...ID" (p.ej. customerID).
    if raw.endswith("ID"):
        stem = raw[:-2].rstrip("_-. ")
        if not stem:
            return None
        return (stem.lower(), "ID")
    # Camel: "...Id" (p.ej. AlbumId).
    if raw.endswith("Id"):
        stem = raw[:-2].rstrip("_-. ")
        if not stem:
            return None
        return (stem.lower(), "Id")
    return None


def _fk_type_compatible(col):
    """True si el tipo de la columna admite ser clave foranea.

    Compatible si el ``physical_type`` NO es una medida flotante, una temporal ni
    boolean, Y ademas (``inferred_type`` en {numeric, categorical, integer} O el
    ``physical_type`` empieza por entero/varchar/text/char/uuid). La comparacion es
    indistinta a mayusculas/minusculas.
    """
    phys = (col.get("physical_type") or "").strip().lower()
    inferred = (col.get("inferred_type") or "").strip().lower()
    # Exclusion por tipo fisico (gana sobre cualquier senal positiva).
    for bad in _FK_PHYSICAL_EXCLUDE:
        if phys.startswith(bad):
            return False
    # Senal positiva por tipo inferido.
    if inferred in _FK_INFERRED_OK:
        return True
    # Senal positiva por tipo fisico (entero/texto/uuid).
    for good in _FK_PHYSICAL_PREFIXES:
        if phys.startswith(good):
            return True
    return False


def suggest_intratable_fk_candidates(profile: dict, max_candidates: int = 20) -> list:
    """Sugiere columnas que parecen una FK intra-tabla por nombre + cardinalidad.

    Heuristica (no afirma nada): una columna es candidata a clave foranea si su nombre
    tiene sufijo de id con stem no vacio (``<algo>_id`` / ``<Algo>Id`` / ``<algo>ID``,
    NUNCA el ``id`` generico), no es ya candidata a PK, no es constante, tiene
    cardinalidad alta pero por debajo del numero de filas (N:1, no unica) y un tipo
    compatible con clave (entero/categorico/texto/uuid; nunca float/fecha/boolean).

    Args:
        profile: TableProfile (dict de ``profile_table``). Se leen, de forma
            defensiva, ``columns`` (lista de ColumnProfile), ``n_rows`` y
            ``key_candidates`` (nombres de columna ya candidatos a PK).
        max_candidates: tope de sugerencias devueltas (default 20). Las columnas se
            ordenan por ``distinct_count`` descendente (mas informativas primero)
            antes de cortar.

    Returns:
        list de dicts (posiblemente vacia), uno por columna sugerida, con claves:
        ``column``, ``ref_table_guess`` (stem del nombre), ``reason`` (frase humana),
        ``distinct_count``, ``unique_pct`` (fraccion 0-1 tal como viene del profile),
        ``inferred_type``, ``physical_type``. Nunca lanza: si ``profile`` no es dict o
        no hay columnas, devuelve ``[]``.
    """
    if not isinstance(profile, dict):
        return []
    columns = profile.get("columns")
    if not isinstance(columns, list):
        return []

    n_rows = profile.get("n_rows")
    has_n_rows = (
        isinstance(n_rows, int) and not isinstance(n_rows, bool) and n_rows > 0
    )

    key_candidates = profile.get("key_candidates")
    if not isinstance(key_candidates, (list, tuple, set)):
        key_candidates = []
    key_set = set(key_candidates)

    out = []
    for col in columns:
        if not isinstance(col, dict):
            continue
        name = col.get("name")

        # 1) Senal de nombre: sufijo de id con stem no vacio.
        signal = _fk_name_signal(name)
        if signal is None:
            continue
        ref_guess, suffix = signal

        # 2) No es ya candidata a PK (clave primaria de la propia tabla).
        if name in key_set:
            continue

        # 3) No constante y con >= 2 valores distintos.
        flags = col.get("flags") or []
        if "constant" in flags:
            continue
        dc = col.get("distinct_count")
        if not (isinstance(dc, int) and not isinstance(dc, bool) and dc >= 2):
            continue

        # 4) Cardinalidad alta pero < n_rows (no es PK) y no parece unica.
        if has_n_rows and dc >= n_rows:
            continue
        unique_pct = col.get("unique_pct")
        has_unique = (
            isinstance(unique_pct, (int, float)) and not isinstance(unique_pct, bool)
        )
        if has_unique and unique_pct >= 0.99:
            continue

        # 5) Tipo compatible con clave foranea (entero/categorico/texto; no medida).
        if not _fk_type_compatible(col):
            continue

        out.append(
            {
                "column": name,
                "ref_table_guess": ref_guess,
                "reason": _build_reason(suffix, dc, n_rows if has_n_rows else None, ref_guess),
                "distinct_count": dc,
                "unique_pct": float(unique_pct) if has_unique else None,
                "inferred_type": col.get("inferred_type") or "",
                "physical_type": col.get("physical_type") or "",
            }
        )

    # Mas informativas primero (mayor cardinalidad), luego corte.
    out.sort(key=lambda d: d.get("distinct_count") or 0, reverse=True)
    return out[: max(0, int(max_candidates))]


def _build_reason(suffix, dc, n_rows, ref_guess):
    """Frase humana que deja claro que la sugerencia es heuristica, no confirmada."""
    if n_rows is not None:
        card = f"es N:1 ({dc} valores distintos < {n_rows} filas)"
    else:
        card = f"tiene {dc} valores distintos que se repiten (cardinalidad N:1)"
    return (
        f"el nombre termina en '{suffix}' y {card}: parece (heuristica por nombre, "
        f"sin confirmar containment) una referencia a una tabla «{ref_guess}»"
    )