API — Procesador

rutificador.procesador.ProcesadorLotesRut

Servicio para validar y formatear RUTs en lotes.

Source code in rutificador/procesador.py

class ProcesadorLotesRut:
    """Servicio para validar y formatear RUTs en lotes."""

    def __init__(
        self,
        validador: ValidadorRut | None = None,
        max_trabajadores: int | None = None,
        motor_paralelo: Literal["thread", "process"] = "process",
    ) -> None:
        """Inicializa el procesador de lotes.

        Args:
            validador: Validador personalizado. Por defecto crea uno
                en modo ``ESTRICTO``.
            max_trabajadores: Máximo de workers paralelos. Por defecto
                usa ``os.cpu_count()``.
            motor_paralelo: Tipo de executor (``"thread"`` para
                ``ThreadPoolExecutor``, ``"process"`` para
                ``ProcessPoolExecutor``).
        """
        self.validador = validador or ValidadorRut()
        self.max_trabajadores: int | None = max_trabajadores
        self.motor_paralelo = motor_paralelo

    def obtener_clase_ejecutor(
        self,
    ) -> "type[ThreadPoolExecutor | ProcessPoolExecutor]":
        # Resolución vía atributo de módulo (PEP 562 `__getattr__`) en vez
        # de `from concurrent.futures import ...` directo: el import del
        # paquete sigue sin arrastrar `multiprocessing` (ausente en Pyodide)
        # y el parcheo de `rutificador.procesador.ThreadPoolExecutor` /
        # `.ProcessPoolExecutor` en tests sigue funcionando.
        modulo = sys.modules[__name__]
        ejecutor_hilos = cast("type[ThreadPoolExecutor]", modulo.ThreadPoolExecutor)
        ejecutor_procesos = cast(
            "type[ProcessPoolExecutor]", modulo.ProcessPoolExecutor
        )

        if self.motor_paralelo == "process" and sys.platform == "win32":
            logger.warning(
                "Procesamiento en paralelo con procesos no está soportado en Windows; "
                "usando ThreadPoolExecutor"
            )
            return ejecutor_hilos
        return ejecutor_procesos if self.motor_paralelo == "process" else ejecutor_hilos

    def _calcular_chunksize(self, n_items: int, max_workers: int) -> int:
        """Calcula un chunksize óptimo para el procesamiento paralelo."""
        if n_items <= 0:
            return 1
        return max(1, min(1000, n_items // (max_workers * 4)))

    @monitor_de_rendimiento
    def validar_lista_ruts(
        self,
        ruts: Sequence[str],
        paralelo: bool = False,
        chunksize: int | None = None,
    ) -> ResultadoLote:
        """Valida una secuencia de RUTs.

        Args:
            ruts: Secuencia de cadenas con RUTs a validar.
            paralelo: Si ``True``, distribuye la validación entre
                múltiples workers.
            chunksize: Tamaño de lote para procesamiento paralelo.
                Por defecto se calcula automáticamente.

        Returns:
            ``ResultadoLote`` con RUTs válidos e inválidos.
        """
        inicio = time.perf_counter()
        resultado = ResultadoLote()

        if paralelo:
            configuracion = self.validador.configuracion
            modo = self.validador.modo
            cargas = ((cadena, configuracion, modo) for cadena in ruts)
            cls_ejecutor = self.obtener_clase_ejecutor()
            n_workers = self.max_trabajadores or (os.cpu_count() or 1)
            c_size = chunksize or self._calcular_chunksize(len(ruts), n_workers)

            with cls_ejecutor(max_workers=self.max_trabajadores) as ejecutor:
                resultados = ejecutor.map(
                    _validar_rut_en_proceso, cargas, chunksize=c_size
                )
        else:
            resultados = (_validar_rut_local(cadena, self.validador) for cadena in ruts)

        for es_valido, valor in resultados:
            if es_valido:
                assert isinstance(valor, RutProcesado)
                detalle = valor
                resultado.ruts_validos.append(detalle.valor)
                resultado.detalles_validos.append(detalle)
            else:
                assert isinstance(valor, DetalleError)
                resultado.ruts_invalidos.append(valor)

        resultado.total_procesados = len(ruts)
        resultado.tiempo_procesamiento = time.perf_counter() - inicio
        return resultado

    def flujo(self, ruts: Iterable[str | int]) -> Iterator[ValidacionResultado]:
        """Procesa RUTs en flujo continuo (streaming) sin materializar el lote completo."""
        for rut in ruts:
            yield Rut.parse(
                rut,
                modo=self.validador.modo,
                configuracion=self.validador.configuracion,
            )

    @monitor_de_rendimiento
    def formatear_lista_ruts(
        self,
        ruts: Sequence[str],
        separador_miles: bool = False,
        mayusculas: bool = False,
        formato: str | None = None,
        paralelo: bool = False,
        **kwargs_formateador: Any,
    ) -> str:
        """Valida y formatea una secuencia de RUTs.

        Args:
            ruts: Secuencia de cadenas con RUTs a validar.
            separador_miles: Si ``True``, incluye separadores de miles.
            mayusculas: Si ``True``, el DV se muestra en mayúscula.
            formato: Formato de salida (``None`` para texto plano,
                ``"csv"``, ``"json"``, ``"xml"``).
            paralelo: Si ``True``, procesa en paralelo.
            **kwargs_formateador: Argumentos adicionales para el
                formateador.

        Returns:
            Cadena con los RUTs formateados.
        """
        if not isinstance(ruts, (list, tuple)):
            raise TypeError(
                f"ruts debe ser una secuencia, se recibió: {type(ruts).__name__}"
            )
        asegurar_booleano(separador_miles, "separador_miles")
        asegurar_booleano(mayusculas, "mayusculas")

        resultado_validacion = self.validar_lista_ruts(ruts, paralelo=paralelo)
        partes: list[str] = ["RUTs válidos:"]
        fuentes = resultado_validacion.detalles_validos

        formateador_detalle = partial(
            _formatear_detalle,
            separador_miles=separador_miles,
            mayusculas=mayusculas,
        )

        if paralelo:
            cls_ejecutor = self.obtener_clase_ejecutor()
            with cls_ejecutor(max_workers=self.max_trabajadores) as ejecutor:
                ruts_formateados = list(ejecutor.map(formateador_detalle, fuentes))
        else:
            ruts_formateados = [formateador_detalle(item) for item in fuentes]

        if formato:
            formatter = FabricaFormateadorRut.obtener_formateador(
                formato, **kwargs_formateador
            )
            if not formatter:
                disponibles = FabricaFormateadorRut.obtener_formatos_disponibles()
                raise ValueError(
                    f"Formato '{formato}' no soportado. "
                    f"Formatos disponibles: {disponibles}"
                )
            partes.append(formatter.formatear(ruts_formateados))
        else:
            partes.append("\n".join(ruts_formateados))

        if resultado_validacion.ruts_invalidos:
            partes.append("")
            partes.append("RUTs inválidos:")
            for detalle in resultado_validacion.ruts_invalidos:
                codigo = detalle.codigo or "SIN_CODIGO"
                partes.append(f"{detalle.rut} [{codigo}] - {detalle.mensaje}")

        tiempo_procesamiento = resultado_validacion.tiempo_procesamiento

        partes.extend(
            [
                "",
                "Estadísticas de procesamiento:",
                f"- Total procesados: {resultado_validacion.total_procesados}",
                f"- RUTs válidos: {len(resultado_validacion.ruts_validos)}",
                f"- RUTs inválidos: {len(resultado_validacion.ruts_invalidos)}",
                f"- Tasa de éxito: {resultado_validacion.tasa_exito:.1f}%",
                f"- Tiempo de procesamiento: {tiempo_procesamiento:.4f}s",
            ]
        )
        return "\n".join(partes)

__init__

__init__(validador: ValidadorRut | None = None, max_trabajadores: int | None = None, motor_paralelo: Literal['thread', 'process'] = 'process') -> None

Inicializa el procesador de lotes.

Parameters:

Name	Type	Description	Default
validador	ValidadorRut | None	Validador personalizado. Por defecto crea uno en modo ESTRICTO.	None
max_trabajadores	int | None	Máximo de workers paralelos. Por defecto usa os.cpu_count().	None
motor_paralelo	Literal['thread', 'process']	Tipo de executor ("thread" para ThreadPoolExecutor, "process" para ProcessPoolExecutor).	'process'

Source code in rutificador/procesador.py

def __init__(
    self,
    validador: ValidadorRut | None = None,
    max_trabajadores: int | None = None,
    motor_paralelo: Literal["thread", "process"] = "process",
) -> None:
    """Inicializa el procesador de lotes.

    Args:
        validador: Validador personalizado. Por defecto crea uno
            en modo ``ESTRICTO``.
        max_trabajadores: Máximo de workers paralelos. Por defecto
            usa ``os.cpu_count()``.
        motor_paralelo: Tipo de executor (``"thread"`` para
            ``ThreadPoolExecutor``, ``"process"`` para
            ``ProcessPoolExecutor``).
    """
    self.validador = validador or ValidadorRut()
    self.max_trabajadores: int | None = max_trabajadores
    self.motor_paralelo = motor_paralelo

validar_lista_ruts

validar_lista_ruts(ruts: Sequence[str], paralelo: bool = False, chunksize: int | None = None) -> ResultadoLote

Valida una secuencia de RUTs.

Parameters:

Name	Type	Description	Default
ruts	Sequence[str]	Secuencia de cadenas con RUTs a validar.	required
paralelo	bool	Si True, distribuye la validación entre múltiples workers.	False
chunksize	int | None	Tamaño de lote para procesamiento paralelo. Por defecto se calcula automáticamente.	None

Returns:

Type	Description
ResultadoLote	ResultadoLote con RUTs válidos e inválidos.

Source code in rutificador/procesador.py

@monitor_de_rendimiento
def validar_lista_ruts(
    self,
    ruts: Sequence[str],
    paralelo: bool = False,
    chunksize: int | None = None,
) -> ResultadoLote:
    """Valida una secuencia de RUTs.

    Args:
        ruts: Secuencia de cadenas con RUTs a validar.
        paralelo: Si ``True``, distribuye la validación entre
            múltiples workers.
        chunksize: Tamaño de lote para procesamiento paralelo.
            Por defecto se calcula automáticamente.

    Returns:
        ``ResultadoLote`` con RUTs válidos e inválidos.
    """
    inicio = time.perf_counter()
    resultado = ResultadoLote()

    if paralelo:
        configuracion = self.validador.configuracion
        modo = self.validador.modo
        cargas = ((cadena, configuracion, modo) for cadena in ruts)
        cls_ejecutor = self.obtener_clase_ejecutor()
        n_workers = self.max_trabajadores or (os.cpu_count() or 1)
        c_size = chunksize or self._calcular_chunksize(len(ruts), n_workers)

        with cls_ejecutor(max_workers=self.max_trabajadores) as ejecutor:
            resultados = ejecutor.map(
                _validar_rut_en_proceso, cargas, chunksize=c_size
            )
    else:
        resultados = (_validar_rut_local(cadena, self.validador) for cadena in ruts)

    for es_valido, valor in resultados:
        if es_valido:
            assert isinstance(valor, RutProcesado)
            detalle = valor
            resultado.ruts_validos.append(detalle.valor)
            resultado.detalles_validos.append(detalle)
        else:
            assert isinstance(valor, DetalleError)
            resultado.ruts_invalidos.append(valor)

    resultado.total_procesados = len(ruts)
    resultado.tiempo_procesamiento = time.perf_counter() - inicio
    return resultado

flujo

flujo(ruts: Iterable[str | int]) -> Iterator[ValidacionResultado]

Procesa RUTs en flujo continuo (streaming) sin materializar el lote completo.

Source code in rutificador/procesador.py

def flujo(self, ruts: Iterable[str | int]) -> Iterator[ValidacionResultado]:
    """Procesa RUTs en flujo continuo (streaming) sin materializar el lote completo."""
    for rut in ruts:
        yield Rut.parse(
            rut,
            modo=self.validador.modo,
            configuracion=self.validador.configuracion,
        )

formatear_lista_ruts

formatear_lista_ruts(ruts: Sequence[str], separador_miles: bool = False, mayusculas: bool = False, formato: str | None = None, paralelo: bool = False, **kwargs_formateador: Any) -> str

Valida y formatea una secuencia de RUTs.

Parameters:

Name	Type	Description	Default
ruts	Sequence[str]	Secuencia de cadenas con RUTs a validar.	required
separador_miles	bool	Si True, incluye separadores de miles.	False
mayusculas	bool	Si True, el DV se muestra en mayúscula.	False
formato	str | None	Formato de salida (None para texto plano, "csv", "json", "xml").	None
paralelo	bool	Si True, procesa en paralelo.	False
**kwargs_formateador	Any	Argumentos adicionales para el formateador.	{}

Returns:

Type	Description
str	Cadena con los RUTs formateados.

Source code in rutificador/procesador.py

@monitor_de_rendimiento
def formatear_lista_ruts(
    self,
    ruts: Sequence[str],
    separador_miles: bool = False,
    mayusculas: bool = False,
    formato: str | None = None,
    paralelo: bool = False,
    **kwargs_formateador: Any,
) -> str:
    """Valida y formatea una secuencia de RUTs.

    Args:
        ruts: Secuencia de cadenas con RUTs a validar.
        separador_miles: Si ``True``, incluye separadores de miles.
        mayusculas: Si ``True``, el DV se muestra en mayúscula.
        formato: Formato de salida (``None`` para texto plano,
            ``"csv"``, ``"json"``, ``"xml"``).
        paralelo: Si ``True``, procesa en paralelo.
        **kwargs_formateador: Argumentos adicionales para el
            formateador.

    Returns:
        Cadena con los RUTs formateados.
    """
    if not isinstance(ruts, (list, tuple)):
        raise TypeError(
            f"ruts debe ser una secuencia, se recibió: {type(ruts).__name__}"
        )
    asegurar_booleano(separador_miles, "separador_miles")
    asegurar_booleano(mayusculas, "mayusculas")

    resultado_validacion = self.validar_lista_ruts(ruts, paralelo=paralelo)
    partes: list[str] = ["RUTs válidos:"]
    fuentes = resultado_validacion.detalles_validos

    formateador_detalle = partial(
        _formatear_detalle,
        separador_miles=separador_miles,
        mayusculas=mayusculas,
    )

    if paralelo:
        cls_ejecutor = self.obtener_clase_ejecutor()
        with cls_ejecutor(max_workers=self.max_trabajadores) as ejecutor:
            ruts_formateados = list(ejecutor.map(formateador_detalle, fuentes))
    else:
        ruts_formateados = [formateador_detalle(item) for item in fuentes]

    if formato:
        formatter = FabricaFormateadorRut.obtener_formateador(
            formato, **kwargs_formateador
        )
        if not formatter:
            disponibles = FabricaFormateadorRut.obtener_formatos_disponibles()
            raise ValueError(
                f"Formato '{formato}' no soportado. "
                f"Formatos disponibles: {disponibles}"
            )
        partes.append(formatter.formatear(ruts_formateados))
    else:
        partes.append("\n".join(ruts_formateados))

    if resultado_validacion.ruts_invalidos:
        partes.append("")
        partes.append("RUTs inválidos:")
        for detalle in resultado_validacion.ruts_invalidos:
            codigo = detalle.codigo or "SIN_CODIGO"
            partes.append(f"{detalle.rut} [{codigo}] - {detalle.mensaje}")

    tiempo_procesamiento = resultado_validacion.tiempo_procesamiento

    partes.extend(
        [
            "",
            "Estadísticas de procesamiento:",
            f"- Total procesados: {resultado_validacion.total_procesados}",
            f"- RUTs válidos: {len(resultado_validacion.ruts_validos)}",
            f"- RUTs inválidos: {len(resultado_validacion.ruts_invalidos)}",
            f"- Tasa de éxito: {resultado_validacion.tasa_exito:.1f}%",
            f"- Tiempo de procesamiento: {tiempo_procesamiento:.4f}s",
        ]
    )
    return "\n".join(partes)

rutificador.procesador.RutProcesado dataclass

Representa un RUT validado con metadatos adicionales.

Source code in rutificador/procesador.py

@dataclass
class RutProcesado:
    """Representa un RUT validado con metadatos adicionales."""

    valor: str
    base: str
    digito: str
    validador_modo: str
    duracion: float = 0.0

    def formatear(
        self,
        separador_miles: bool = False,
        mayusculas: bool = False,
        separador_personalizado: str = ".",
    ) -> str:
        """Formatea el RUT reutilizando la información ya validada."""
        resultado = self.valor
        if separador_miles:
            base_formateada = Rut.agregar_separador_miles(
                self.base, separador_personalizado
            )
            resultado = f"{base_formateada}-{self.digito}"
        if mayusculas:
            resultado = resultado.upper()
        return resultado

formatear

formatear(separador_miles: bool = False, mayusculas: bool = False, separador_personalizado: str = '.') -> str

Formatea el RUT reutilizando la información ya validada.

Source code in rutificador/procesador.py

def formatear(
    self,
    separador_miles: bool = False,
    mayusculas: bool = False,
    separador_personalizado: str = ".",
) -> str:
    """Formatea el RUT reutilizando la información ya validada."""
    resultado = self.valor
    if separador_miles:
        base_formateada = Rut.agregar_separador_miles(
            self.base, separador_personalizado
        )
        resultado = f"{base_formateada}-{self.digito}"
    if mayusculas:
        resultado = resultado.upper()
    return resultado

rutificador.procesador.ResultadoLote dataclass

Contenedor de resultados para operaciones por lotes.

Source code in rutificador/procesador.py

@dataclass
class ResultadoLote:
    """Contenedor de resultados para operaciones por lotes."""

    detalles_validos: list[RutProcesado] = field(default_factory=list)
    ruts_validos: list[str] = field(default_factory=list)
    ruts_invalidos: list[DetalleError] = field(default_factory=list)
    tiempo_procesamiento: float = 0.0
    total_procesados: int = 0

    @property
    def tasa_exito(self) -> float:
        if self.total_procesados == 0:
            return 0.0
        return (len(self.ruts_validos) / self.total_procesados) * 100

rutificador.procesador.DetalleError dataclass

Detalle estructurado de error o advertencia.

Source code in rutificador/errores.py

@dataclass(frozen=True)
class DetalleError:
    """Detalle estructurado de error o advertencia."""

    codigo: str
    mensaje: str
    hint: str
    severidad: Severidad
    recuperable: bool
    rut: str | None = None
    duracion: float = 0.0

    def __eq__(self, other: object) -> bool:  # pragma: no cover - comparación simple
        if not isinstance(other, DetalleError):
            return NotImplemented
        return (
            self.codigo,
            self.mensaje,
            self.hint,
            self.severidad,
            self.recuperable,
            self.rut,
        ) == (
            other.codigo,
            other.mensaje,
            other.hint,
            other.severidad,
            other.recuperable,
            other.rut,
        )

rutificador.procesador.validar_flujo_ruts

validar_flujo_ruts(ruts: Iterable[str | int], paralelo: bool = False, max_trabajadores: int | None = None, motor_paralelo: Literal['thread', 'process'] = 'process', chunksize: int = CHUNKSIZE_FLUJO_POR_DEFECTO) -> Iterator[tuple[bool, RutProcesado | DetalleError]]

Valida RUTs desde cualquier iterable y produce resultados uno a uno.

Si paralelo es True, distribuye la carga entre múltiples trabajadores manteniendo la naturaleza de generador (no materializa el lote).

Parameters:

Name	Type	Description	Default
ruts	Iterable[str | int]	Iterable de RUTs (puede ser un generador ilimitado).	required
paralelo	bool	Si True, usa un executor multiproceso o multihilo.	False
max_trabajadores	int | None	Número máximo de workers. Por defecto usa os.cpu_count().	None
motor_paralelo	Literal['thread', 'process']	"process" (por defecto) o "thread".	'process'
chunksize	int	Cantidad de ítems que se envían juntos a cada worker en el modo paralelo. Un valor mayor reduce el overhead de serialización inter-proceso. No usar len(ruts) aquí: el iterable puede ser un generador sin longitud definida.	CHUNKSIZE_FLUJO_POR_DEFECTO

Source code in rutificador/procesador.py

def validar_flujo_ruts(
    ruts: Iterable[str | int],
    paralelo: bool = False,
    max_trabajadores: int | None = None,
    motor_paralelo: Literal["thread", "process"] = "process",
    chunksize: int = CHUNKSIZE_FLUJO_POR_DEFECTO,
) -> Iterator[tuple[bool, RutProcesado | DetalleError]]:
    """Valida RUTs desde cualquier iterable y produce resultados uno a uno.

    Si paralelo es True, distribuye la carga entre múltiples trabajadores
    manteniendo la naturaleza de generador (no materializa el lote).

    Args:
        ruts: Iterable de RUTs (puede ser un generador ilimitado).
        paralelo: Si ``True``, usa un executor multiproceso o multihilo.
        max_trabajadores: Número máximo de workers. Por defecto usa
            ``os.cpu_count()``.
        motor_paralelo: ``"process"`` (por defecto) o ``"thread"``.
        chunksize: Cantidad de ítems que se envían juntos a cada worker en
            el modo paralelo. Un valor mayor reduce el overhead de
            serialización inter-proceso. No usar ``len(ruts)`` aquí: el
            iterable puede ser un generador sin longitud definida.
    """
    procesador = ProcesadorLotesRut(
        max_trabajadores=max_trabajadores, motor_paralelo=motor_paralelo
    )

    if paralelo:
        configuracion = procesador.validador.configuracion
        modo = procesador.validador.modo
        cargas = ((str(rut), configuracion, modo) for rut in ruts)
        cls_ejecutor = procesador.obtener_clase_ejecutor()

        with cls_ejecutor(max_workers=procesador.max_trabajadores) as ejecutor:
            for es_valido, detalle in ejecutor.map(
                _validar_rut_en_proceso, cargas, chunksize=chunksize
            ):
                yield es_valido, detalle
    else:
        for rut in ruts:
            es_valido, detalle = _validar_rut_local(str(rut), procesador.validador)
            yield es_valido, detalle

rutificador.procesador.formatear_flujo_ruts

formatear_flujo_ruts(ruts: Iterable[str | int], separador_miles: bool = False, mayusculas: bool = False, paralelo: bool = False, max_trabajadores: int | None = None, motor_paralelo: Literal['thread', 'process'] = 'process', chunksize: int = CHUNKSIZE_FLUJO_POR_DEFECTO) -> Iterator[tuple[bool, str | DetalleError]]

Valida y formatea RUTs provenientes de cualquier iterable.

Soporta opcionalmente procesamiento paralelo manteniendo el flujo iterativo.

Parameters:

Name	Type	Description	Default
ruts	Iterable[str | int]	Iterable de RUTs (puede ser un generador ilimitado).	required
separador_miles	bool	Si True, incluye separadores de miles.	False
mayusculas	bool	Si True, el DV se muestra en mayúscula.	False
paralelo	bool	Si True, usa un executor multiproceso o multihilo.	False
max_trabajadores	int | None	Número máximo de workers.	None
motor_paralelo	Literal['thread', 'process']	"process" (por defecto) o "thread".	'process'
chunksize	int	Tamaño de lote para el mapeo paralelo. Ver :func:validar_flujo_ruts.	CHUNKSIZE_FLUJO_POR_DEFECTO

Source code in rutificador/procesador.py

def formatear_flujo_ruts(
    ruts: Iterable[str | int],
    separador_miles: bool = False,
    mayusculas: bool = False,
    paralelo: bool = False,
    max_trabajadores: int | None = None,
    motor_paralelo: Literal["thread", "process"] = "process",
    chunksize: int = CHUNKSIZE_FLUJO_POR_DEFECTO,
) -> Iterator[tuple[bool, str | DetalleError]]:
    """Valida y formatea RUTs provenientes de cualquier iterable.

    Soporta opcionalmente procesamiento paralelo manteniendo el flujo iterativo.

    Args:
        ruts: Iterable de RUTs (puede ser un generador ilimitado).
        separador_miles: Si ``True``, incluye separadores de miles.
        mayusculas: Si ``True``, el DV se muestra en mayúscula.
        paralelo: Si ``True``, usa un executor multiproceso o multihilo.
        max_trabajadores: Número máximo de workers.
        motor_paralelo: ``"process"`` (por defecto) o ``"thread"``.
        chunksize: Tamaño de lote para el mapeo paralelo. Ver
            :func:`validar_flujo_ruts`.
    """
    asegurar_booleano(separador_miles, "separador_miles")
    asegurar_booleano(mayusculas, "mayusculas")

    # Reutilizamos la lógica de validación de flujo y simplemente mapeamos el formateo
    for es_valido, detalle in validar_flujo_ruts(
        ruts,
        paralelo=paralelo,
        max_trabajadores=max_trabajadores,
        motor_paralelo=motor_paralelo,
        chunksize=chunksize,
    ):
        if es_valido:
            # Si el detalle es de una validación exitosa, es un RutProcesado
            # que ya tiene el método .formatear()
            yield (
                True,
                detalle.formatear(  # type: ignore[union-attr]
                    separador_miles=separador_miles, mayusculas=mayusculas
                ),
            )
        else:
            yield False, detalle  # type: ignore[misc]

rutificador.procesador.validar_lista_ruts

validar_lista_ruts(ruts: Sequence[str], paralelo: bool = False, max_trabajadores: int | None = None, motor_paralelo: Literal['thread', 'process'] = 'process') -> ResumenValidacion

Valida una secuencia de RUTs utilizando ProcesadorLotesRut.

Source code in rutificador/procesador.py

def validar_lista_ruts(
    ruts: Sequence[str],
    paralelo: bool = False,
    max_trabajadores: int | None = None,
    motor_paralelo: Literal["thread", "process"] = "process",
) -> ResumenValidacion:
    """Valida una secuencia de RUTs utilizando ``ProcesadorLotesRut``."""

    procesador = ProcesadorLotesRut(
        max_trabajadores=max_trabajadores, motor_paralelo=motor_paralelo
    )
    resultado = procesador.validar_lista_ruts(ruts, paralelo=paralelo)
    return {"validos": resultado.ruts_validos, "invalidos": resultado.ruts_invalidos}

rutificador.procesador.formatear_lista_ruts

formatear_lista_ruts(ruts: Sequence[str], separador_miles: bool = False, mayusculas: bool = False, formato: str | None = None, paralelo: bool = False, max_trabajadores: int | None = None, motor_paralelo: Literal['thread', 'process'] = 'process', **kwargs_formateador: Any) -> str

Valida y formatea una secuencia de RUTs.

Source code in rutificador/procesador.py

def formatear_lista_ruts(
    ruts: Sequence[str],
    separador_miles: bool = False,
    mayusculas: bool = False,
    formato: str | None = None,
    paralelo: bool = False,
    max_trabajadores: int | None = None,
    motor_paralelo: Literal["thread", "process"] = "process",
    **kwargs_formateador: Any,
) -> str:
    """Valida y formatea una secuencia de RUTs."""

    procesador = ProcesadorLotesRut(
        max_trabajadores=max_trabajadores, motor_paralelo=motor_paralelo
    )
    return procesador.formatear_lista_ruts(
        ruts,
        separador_miles=separador_miles,
        mayusculas=mayusculas,
        formato=formato,
        paralelo=paralelo,
        **kwargs_formateador,
    )

rutificador.procesador.evaluar_rendimiento

evaluar_rendimiento(num_ruts: int = 10000, paralelo: bool = True) -> dict[str, Any]

Evalúa el rendimiento del procesamiento de RUTs.

Genera RUTs aleatorios y mide el tiempo de validación y formateo en serie y en paralelo.

Parameters:

Name	Type	Description	Default
num_ruts	int	Cantidad de RUTs a generar para la prueba.	10000
paralelo	bool	Si True, incluye medición con paralelismo.	True

Returns:

Type	Description
dict[str, Any]	Diccionario con métricas de rendimiento (tiempos, tasa de éxito,
dict[str, Any]	conteo de RUTs).

Source code in rutificador/procesador.py

def evaluar_rendimiento(num_ruts: int = 10000, paralelo: bool = True) -> dict[str, Any]:
    """Evalúa el rendimiento del procesamiento de RUTs.

    Genera RUTs aleatorios y mide el tiempo de validación y formateo
    en serie y en paralelo.

    Args:
        num_ruts: Cantidad de RUTs a generar para la prueba.
        paralelo: Si ``True``, incluye medición con paralelismo.

    Returns:
        Diccionario con métricas de rendimiento (tiempos, tasa de éxito,
        conteo de RUTs).
    """
    pruebas = []
    for _ in range(num_ruts):
        base = str(random.randint(1_000_000, 99_999_999))  # nosec B311
        dv = calcular_digito_verificador(base)
        pruebas.append(f"{base}-{dv}")

    procesador = ProcesadorLotesRut()
    inicio = time.perf_counter()
    resultado = procesador.validar_lista_ruts(pruebas, paralelo=paralelo)
    tiempo_validacion = time.perf_counter() - inicio

    inicio = time.perf_counter()
    procesador.formatear_lista_ruts(pruebas[:1000], paralelo=paralelo)
    tiempo_formato = time.perf_counter() - inicio

    return {
        "conteo_ruts_prueba": num_ruts,
        "procesamiento_paralelo": paralelo,
        "tiempo_validacion": tiempo_validacion,
        "tasa_validacion": num_ruts / tiempo_validacion if tiempo_validacion else 0,
        "tiempo_formateo": tiempo_formato,
        "tasa_formateo": 1000 / tiempo_formato if tiempo_formato else 0,
        "tasa_exito": resultado.tasa_exito,
        "estadisticas_cache": Rut.estadisticas_cache(),
    }