Detección y eliminación de secuencias duplicadas ignorando identificadores

Este script analiza archivos FASTA para localizar secuencias biológicas idénticas, independientemente del nombre del encabezado. Cuando encuentra repeticionse, solicita al usuario elegir manualmente cuál conservar o mantener todas.

import os
from collections import defaultdict

def cargar_fasta(ruta_archivo):
    registros = []
    with open(ruta_archivo, "r", encoding="utf-8") as f:
        cabecera = None
        lineas_secuencia = []

        for linea in f:
            linea = linea.rstrip()
            if linea.startswith(">"):
                if cabecera is not None:
                    registros.append((cabecera, "".join(lineas_secuencia)))
                cabecera = linea
                lineas_secuencia = []
            else:
                lineas_secuencia.append(linea)

        if cabecera is not None:
            registros.append((cabecera, "".join(lineas_secuencia)))

    return registros

def depurar_manual(registros):
    mapa_secuencias = defaultdict(list)
    for cab, sec in registros:
        mapa_secuencias[sec].append(cab)

    registros_filtrados = []

    for sec, cabeceras in mapa_secuencias.items():
        if len(cabeceras) == 1:
            registros_filtrados.append((cabeceras[0], sec))
            continue

        print("\n\u26a0\ufe0f Duplicado detectado (secuencia idéntica):")
        for i, cab in enumerate(cabeceras):
            print(f"  [{i}] {cab}")

        print("Indica el número de la secuencia a CONSERVAR (ej. 0)")
        print("o presiona Enter para guardar todas:")

        eleccion = input("Tu elección: ").strip()

        if eleccion == "":
            for cab in cabeceras:
                registros_filtrados.append((cab, sec))
        else:
            try:
                indice = int(eleccion)
                registros_filtrados.append((cabeceras[indice], sec))
            except (ValueError, IndexError):
                print("Entrada no válida, se mantienen todas por defecto")
                for cab in cabeceras:
                    registros_filtrados.append((cab, sec))

    return registros_filtrados

def guardar_fasta(registros, ruta_salida):
    with open(ruta_salida, "w", encoding="utf-8") as out:
        for cab, sec in registros:
            out.write(cab + "\n")
            out.write(sec + "\n")

if __name__ == "__main__":
    entrada = r"D:\ejemplos\secuencias_original.fasta"
    salida = r"D:\ejemplos\secuencias_sin_duplicados.fasta"

    datos = cargar_fasta(entrada)
    datos_depurados = depurar_manual(datos)
    guardar_fasta(datos_depurados, salida)

    print(f"\nProceso completado. Se conservaron {len(datos_depurados)} secuencias")
    print(f"Archivo generado: {salida}")

Detección de secuencias idénticas y mismo identificador

En esta segunda versión, primero se eliminan secuencias duplicadas con intervención manual y luego se procesan aquellas entradas que comparten el mismo identificador pero tienen secuencias diferentes, permitiendo renombrarlas interactivamente.

from collections import defaultdict

def leer_fasta(archivo):
    entradas = []
    with open(archivo, "r", encoding="utf-8") as f:
        cabecera = None
        fragmentos = []
        for linea in f:
            linea = linea.rstrip()
            if linea.startswith(">"):
                if cabecera is not None:
                    entradas.append((cabecera, "".join(fragmentos)))
                cabecera = linea
                fragmentos = []
            else:
                fragmentos.append(linea)
        if cabecera is not None:
            entradas.append((cabecera, "".join(fragmentos)))
    return entradas

def extraer_id(encabezado):
    return encabezado[1:].split()[0]

def depurar_por_secuencia(entradas):
    agrupacion = defaultdict(list)
    for cab, seq in entradas:
        agrupacion[seq].append(cab)

    resultado = []
    for seq, cabeceras in agrupacion.items():
        if len(cabeceras) == 1:
            resultado.append((cabeceras[0], seq))
            continue

        print("\n\u26a0\ufe0f Secuencias duplicadas encontradas:")
        for i, cab in enumerate(cabeceras):
            print(f"  [{i}] {cab}")

        print("Elige el índice a CONSERVAR (ej. 0) o Enter para mantener todo:")
        opcion = input("Opción: ").strip()

        if opcion == "":
            for cab in cabeceras:
                resultado.append((cab, seq))
        else:
            try:
                idx = int(opcion)
                resultado.append((cabeceras[idx], seq))
            except (ValueError, IndexError):
                print("Opción inválida, conservando todos")
                for cab in cabeceras:
                    resultado.append((cab, seq))
    return resultado

def renombrar_por_id(entradas):
    grupos_id = defaultdict(list)
    for cab, seq in entradas:
        id_seq = extraer_id(cab)
        grupos_id[id_seq].append((cab, seq))

    resultado_final = []
    for id_seq, items in grupos_id.items():
        if len(items) == 1:
            resultado_final.extend(items)
            continue

        print(f"\n\u26a0\ufe0f ID '{id_seq}' repetido con secuencias diferentes:")
        for i, (cab, seq) in enumerate(items):
            sugerencia = f"{id_seq}_v{i+1}"
            print(f"\n[{i}]")
            print(f"Original: {cab}")
            print(f"Longitud: {len(seq)}")
            print(f"Sugerencia: {sugerencia}")

            nuevo_id = input("Nuevo ID (sin >, Enter mantiene original): ").strip()
            if nuevo_id:
                cab_mod = ">" + nuevo_id
            else:
                cab_mod = cab
            resultado_final.append((cab_mod, seq))
    return resultado_final

def escribir_fasta(entradas, salida):
    with open(salida, "w", encoding="utf-8") as f:
        for cab, seq in entradas:
            f.write(cab + "\n")
            f.write(seq + "\n")

if __name__ == "__main__":
    archivo_in = r"D:\datos\raw_sequences.fasta"
    archivo_out = r"D:\datos\sequences_clean.fasta"

    datos = leer_fasta(archivo_in)
    datos = depurar_por_secuencia(datos)
    datos = renombrar_por_id(datos)
    escribir_fasta(datos, archivo_out)

    print(f"\n\u2705 Limpieza finalizada. {len(datos)} secuencias guardadas")
    print(f"\U0001f4c4 Archivo: {archivo_out}")