fn_registry/python/functions/datascience/compute_text_duplicates.py

"""Detección de documentos duplicados en un corpus de texto.

Función pura, estilo dict-no-throw del grupo `eda`: nunca lanza, siempre
devuelve el mismo contrato de claves. Los duplicados EXACTOS se calculan
siempre con la stdlib (normalización + hash SHA-1). Los CASI-duplicados
(near-dup) requieren la dependencia opcional `datasketch`; si no está
instalada, esa parte degrada limpiamente a ``available: False`` sin afectar
al resto del cálculo.
"""

import hashlib
import re


def _compute_near_dup(valid, near_threshold, sample_max):
    """Cuenta documentos con al menos otro casi-duplicado vía MinHash + LSH.

    Import perezoso de ``datasketch``. Si la librería no está disponible (o
    cualquier paso falla), degrada a ``{"available": False, "n_near_dup_docs": 0}``
    sin propagar la excepción.

    Args:
        valid: lista de str ya filtrada (sin None ni no-str).
        near_threshold: umbral de similitud Jaccard para LSH.
        sample_max: número máximo de documentos a muestrear.

    Returns:
        dict con ``available`` (bool) y ``n_near_dup_docs`` (int). Cuando
        ``available`` es True, incluye además ``threshold``.
    """
    try:
        from datasketch import MinHash, MinHashLSH
    except Exception:
        return {"available": False, "n_near_dup_docs": 0}

    try:
        docs = valid[:sample_max]
        num_perm = 128
        lsh = MinHashLSH(threshold=near_threshold, num_perm=num_perm)
        minhashes = {}

        for i, doc in enumerate(docs):
            tokens = re.findall(r"\w+", doc.lower())
            shingles = set()
            for j in range(len(tokens) - 2):
                shingles.add(" ".join(tokens[j:j + 3]))
            # Documentos con menos de 3 tokens no generan 3-shingles: caemos a
            # los tokens sueltos para no perderlos del todo.
            if not shingles:
                shingles = set(tokens)
            if not shingles:
                # Documento sin tokens (cadena vacía / solo símbolos): se omite.
                continue
            m = MinHash(num_perm=num_perm)
            for sh in shingles:
                m.update(sh.encode("utf-8"))
            key = "d{}".format(i)
            minhashes[key] = m
            lsh.insert(key, m)

        n_near = 0
        for key, m in minhashes.items():
            matches = lsh.query(m)
            if len(matches) > 1:
                n_near += 1

        return {
            "available": True,
            "n_near_dup_docs": int(n_near),
            "threshold": near_threshold,
        }
    except Exception:
        return {"available": False, "n_near_dup_docs": 0}


def compute_text_duplicates(texts, near_threshold=0.85, sample_max=2000) -> dict:
    """Detecta duplicados exactos y casi-duplicados en un corpus de texto.

    Args:
        texts: lista de documentos. Los elementos None o que no sean str se
            descartan; ``n_docs`` cuenta solo los válidos.
        near_threshold: umbral de similitud Jaccard para considerar dos
            documentos casi-duplicados (solo near-dup, requiere datasketch).
        sample_max: tope de documentos muestreados para el cálculo near-dup.

    Returns:
        dict con las claves ``n_docs``, ``n_exact_dup``, ``exact_dup_pct``
        (float redondeado a 2 decimales, o None si el corpus está vacío),
        ``n_unique`` y ``near_dup`` (sub-dict con ``available`` y
        ``n_near_dup_docs``, más ``threshold`` cuando está disponible).
        Nunca lanza: captura toda excepción y degrada.
    """
    # Filtrado defensivo de documentos válidos.
    try:
        valid = [t for t in texts if isinstance(t, str)] if texts is not None else []
    except Exception:
        valid = []

    n_docs = len(valid)

    # Duplicados exactos: normalizar + hash SHA-1 (stdlib, siempre disponible).
    try:
        seen = set()
        n_exact_dup = 0
        for doc in valid:
            norm = " ".join(doc.split()).strip().lower()
            digest = hashlib.sha1(norm.encode("utf-8")).hexdigest()
            if digest in seen:
                n_exact_dup += 1
            else:
                seen.add(digest)
        n_unique = len(seen)
    except Exception:
        n_exact_dup = 0
        n_unique = 0

    exact_dup_pct = round(n_exact_dup / n_docs * 100, 2) if n_docs > 0 else None

    # Casi-duplicados: opcional vía datasketch, degrada solo.
    near_dup = _compute_near_dup(valid, near_threshold, sample_max)

    return {
        "n_docs": n_docs,
        "n_exact_dup": n_exact_dup,
        "exact_dup_pct": exact_dup_pct,
        "n_unique": n_unique,
        "near_dup": near_dup,
    }