"""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, }