Tutorial Python: Archivos comprimidos

Trabajar con archivos ZIP

Los archivos ZIP son uno de los formatos de compresión más populares y ampliamente utilizados. En esencia, funcionan como "cajas digitales" que permiten almacenar múltiples archivos en un solo contenedor, reduciendo su tamaño total y facilitando su distribución. Python ofrece soporte nativo para trabajar con estos archivos a través del módulo zipfile de su biblioteca estándar.

El módulo zipfile proporciona herramientas para crear, leer, escribir, añadir y extraer archivos de contenedores ZIP. Veamos cómo podemos utilizarlo para las operaciones más comunes:

Creación de archivos ZIP

Para crear un nuevo archivo ZIP, utilizamos la clase ZipFile con el modo 'w' (escritura):

import zipfile

# Crear un nuevo archivo ZIP
with zipfile.ZipFile('mi_archivo.zip', 'w') as archivo_zip:
    # Añadir archivos al ZIP
    archivo_zip.write('documento.txt')
    archivo_zip.write('imagen.jpg')
    
    # Podemos añadir archivos desde diferentes carpetas
    archivo_zip.write('carpeta/archivo.pdf')

Por defecto, los archivos se almacenan con su ruta relativa. Si queremos cambiar el nombre o la ruta dentro del ZIP, podemos usar el parámetro arcname:

with zipfile.ZipFile('backup.zip', 'w') as zip_file:
    # Guardar 'config.ini' como 'configuracion/config.ini' dentro del ZIP
    zip_file.write('config.ini', arcname='configuracion/config.ini')

Compresión de archivos

El módulo zipfile ofrece diferentes niveles de compresión. Podemos especificar el método de compresión al crear el archivo:

import zipfile

# Crear ZIP con máxima compresión
with zipfile.ZipFile('datos_comprimidos.zip', 'w', 
                     compression=zipfile.ZIP_DEFLATED, 
                     compresslevel=9) as zip_file:
    zip_file.write('datos_grandes.csv')

Los métodos de compresión disponibles son:

• ZIP_STORED (sin compresión, solo almacenamiento)
• ZIP_DEFLATED (método estándar de compresión)
• ZIP_BZIP2 (compresión BZIP2, generalmente mejor que DEFLATED)
• ZIP_LZMA (compresión LZMA, excelente ratio pero más lenta)

El parámetro compresslevel va de 0 (sin compresión) a 9 (máxima compresión).

Listar contenido de un archivo ZIP

Para ver qué archivos contiene un ZIP sin extraerlos:

import zipfile

with zipfile.ZipFile('proyecto.zip', 'r') as zip_file:
    # Listar todos los archivos en el ZIP
    contenido = zip_file.namelist()
    print(f"El archivo ZIP contiene {len(contenido)} elementos:")
    for item in contenido:
        print(f" - {item}")
        
    # Obtener información detallada de un archivo específico
    info = zip_file.getinfo('README.md')
    print(f"Tamaño original: {info.file_size} bytes")
    print(f"Tamaño comprimido: {info.compress_size} bytes")
    ratio = (info.file_size - info.compress_size) / info.file_size * 100
    print(f"Ratio de compresión: {ratio:.1f}%")

Extracción de archivos

Para extraer todos los archivos de un ZIP:

import zipfile

with zipfile.ZipFile('datos.zip', 'r') as zip_file:
    # Extraer todos los archivos a la carpeta actual
    zip_file.extractall()
    
    # O extraer a una carpeta específica
    zip_file.extractall(path='carpeta_destino')

Si solo queremos extraer archivos específicos:

with zipfile.ZipFile('proyecto.zip', 'r') as zip_file:
    # Extraer un solo archivo
    zip_file.extract('config.json')
    
    # Extraer varios archivos específicos
    archivos_a_extraer = ['README.md', 'docs/manual.pdf']
    for archivo in archivos_a_extraer:
        zip_file.extract(archivo, path='carpeta_destino')

Lectura de archivos sin extraerlos

Una característica muy útil es la capacidad de leer el contenido de un archivo dentro del ZIP sin necesidad de extraerlo al disco:

import zipfile

with zipfile.ZipFile('datos.zip', 'r') as zip_file:
    # Leer un archivo de texto
    with zip_file.open('config.json') as archivo:
        contenido = archivo.read()
        # Si es texto, podemos decodificarlo
        texto = contenido.decode('utf-8')
        print(texto)
        
    # Procesar un archivo CSV sin extraerlo
    import csv
    from io import TextIOWrapper
    
    with zip_file.open('datos.csv') as archivo_csv:
        # Convertir el archivo binario a texto
        csv_file = TextIOWrapper(archivo_csv)
        # Procesar el CSV
        reader = csv.reader(csv_file)
        for fila in reader:
            print(fila)

Añadir archivos a un ZIP existente

Para añadir archivos a un ZIP existente sin recrearlo completamente, usamos el modo 'a' (append):

import zipfile

# Abrir en modo append
with zipfile.ZipFile('proyecto.zip', 'a') as zip_file:
    # Añadir nuevos archivos
    zip_file.write('nuevos_datos.txt')
    zip_file.write('imagen_actualizada.png')

Protección con contraseña

Python 3.9+ permite crear archivos ZIP protegidos con contraseña:

import zipfile

# Crear ZIP con contraseña
with zipfile.ZipFile('confidencial.zip', 'w') as zip_file:
    zip_file.setpassword(b'mi_contraseña_secreta')
    zip_file.write('documento_secreto.pdf')

Para extraer archivos de un ZIP protegido:

with zipfile.ZipFile('confidencial.zip', 'r') as zip_file:
    # Proporcionar la contraseña para extraer
    zip_file.extractall(pwd=b'mi_contraseña_secreta')

Compresión de directorios completos

Para comprimir un directorio completo con todos sus subdirectorios, podemos combinar zipfile con el módulo os:

import zipfile
import os

def comprimir_directorio(directorio, archivo_zip):
    with zipfile.ZipFile(archivo_zip, 'w', zipfile.ZIP_DEFLATED) as zipf:
        # Recorrer todo el árbol de directorios
        for carpeta_actual, subcarpetas, archivos in os.walk(directorio):
            for archivo in archivos:
                # Ruta completa del archivo
                ruta_completa = os.path.join(carpeta_actual, archivo)
                # Ruta relativa para mantener la estructura de directorios
                ruta_relativa = os.path.relpath(ruta_completa, os.path.dirname(directorio))
                # Añadir al ZIP
                zipf.write(ruta_completa, ruta_relativa)

# Ejemplo: comprimir el directorio 'mi_proyecto'
comprimir_directorio('mi_proyecto', 'mi_proyecto.zip')

Casos de uso prácticos

Los archivos ZIP son especialmente útiles en varios escenarios:

• Backup de proyectos: Comprimir carpetas de proyectos para respaldo.
• Distribución de software: Empaquetar aplicaciones y sus dependencias.
• Transferencia de datos: Reducir el tamaño de archivos para enviarlos por correo o subirlos a la nube.
• Procesamiento por lotes: Procesar múltiples archivos empaquetados en un solo contenedor.

Veamos un ejemplo de backup automático de un proyecto:

import zipfile
import os
from datetime import datetime

def crear_backup(directorio_proyecto):
    # Crear nombre con fecha y hora
    fecha = datetime.now().strftime("%Y%m%d_%H%M%S")
    nombre_zip = f"backup_{os.path.basename(directorio_proyecto)}_{fecha}.zip"
    
    # Comprimir el proyecto
    with zipfile.ZipFile(nombre_zip, 'w', zipfile.ZIP_DEFLATED) as zip_file:
        for carpeta_actual, subcarpetas, archivos in os.walk(directorio_proyecto):
            # Excluir directorios como .git, __pycache__, etc.
            subcarpetas[:] = [d for d in subcarpetas if d not in ['.git', '__pycache__', 'venv']]
            
            for archivo in archivos:
                # Excluir archivos temporales o binarios
                if not archivo.endswith(('.pyc', '.tmp', '.log')):
                    ruta_completa = os.path.join(carpeta_actual, archivo)
                    ruta_relativa = os.path.relpath(ruta_completa, os.path.dirname(directorio_proyecto))
                    zip_file.write(ruta_completa, ruta_relativa)
    
    print(f"Backup creado: {nombre_zip}")
    return nombre_zip

# Crear backup del proyecto actual
crear_backup('./mi_proyecto_python')

Este script crea un archivo ZIP con el contenido del proyecto, excluyendo archivos y carpetas que típicamente no necesitan respaldo (como archivos temporales o entornos virtuales).

Módulo tarfile

El formato TAR (Tape Archive) es una forma tradicional de agrupar múltiples archivos en un solo contenedor, originalmente diseñado para copias de seguridad en cintas magnéticas. A diferencia de ZIP, el formato TAR por sí solo no proporciona compresión, sino que simplemente agrupa archivos. Sin embargo, cuando se combina con algoritmos de compresión como gzip o bzip2, se convierte en una solución completa para archivar y comprimir datos.

Python incluye el módulo tarfile en su biblioteca estándar, que proporciona una interfaz completa para trabajar con archivos TAR, incluyendo soporte para los formatos comprimidos más comunes.

Creación de archivos TAR

Para crear un nuevo archivo TAR, utilizamos la clase TarFile con el modo 'w':

import tarfile

# Crear un archivo TAR sin compresión
with tarfile.open('mi_archivo.tar', 'w') as tar:
    # Añadir archivos al TAR
    tar.add('documento.txt')
    tar.add('imagen.jpg')
    
    # Añadir una carpeta completa (incluye subcarpetas)
    tar.add('mi_carpeta')

Para crear un archivo TAR comprimido, especificamos el modo adecuado:

import tarfile

# Crear un archivo TAR comprimido con gzip
with tarfile.open('mi_archivo.tar.gz', 'w:gz') as tar:
    tar.add('datos.csv')
    tar.add('config.json')

# Crear un archivo TAR comprimido con bzip2 (mejor compresión)
with tarfile.open('mi_archivo.tar.bz2', 'w:bz2') as tar:
    tar.add('carpeta_grande')

Los modos de apertura más comunes son:

• 'r' - Lectura de un archivo TAR sin compresión
• 'w' - Escritura de un archivo TAR sin compresión
• 'a' - Añadir a un archivo TAR existente
• 'r:gz' - Lectura de un archivo TAR comprimido con gzip
• 'w:gz' - Escritura de un archivo TAR comprimido con gzip
• 'r:bz2' - Lectura de un archivo TAR comprimido con bzip2
• 'w:bz2' - Escritura de un archivo TAR comprimido con bzip2
• 'r:xz' - Lectura de un archivo TAR comprimido con lzma/xz
• 'w:xz' - Escritura de un archivo TAR comprimido con lzma/xz

Listar contenido de un archivo TAR

Para examinar el contenido de un archivo TAR sin extraerlo:

import tarfile

with tarfile.open('proyecto.tar.gz', 'r:gz') as tar:
    # Listar todos los miembros del archivo TAR
    print("Contenido del archivo TAR:")
    for miembro in tar.getmembers():
        print(f" - {miembro.name} ({miembro.size} bytes)")
        
    # Obtener información detallada
    info = tar.getmember('README.md')
    print(f"\nInformación de README.md:")
    print(f"Tamaño: {info.size} bytes")
    print(f"Tipo: {'Directorio' if info.isdir() else 'Archivo'}")
    print(f"Modo: {info.mode:o}")  # Permisos en formato octal
    print(f"Modificado: {info.mtime}")

El método getmembers() devuelve una lista de objetos TarInfo que contienen metadatos sobre cada archivo o directorio en el archivo TAR. Estos objetos incluyen información como nombre, tamaño, permisos, fechas de modificación y tipo de archivo.

Extracción de archivos

Para extraer todos los archivos de un TAR:

import tarfile

with tarfile.open('datos.tar.gz', 'r:gz') as tar:
    # Extraer todos los archivos a la carpeta actual
    tar.extractall()
    
    # O extraer a una carpeta específica
    tar.extractall(path='carpeta_destino')

Si queremos extraer archivos específicos:

import tarfile

with tarfile.open('proyecto.tar.bz2', 'r:bz2') as tar:
    # Extraer archivos específicos
    miembros = [tar.getmember('config.json'), tar.getmember('README.md')]
    tar.extractall(members=miembros, path='carpeta_destino')
    
    # O extraer un solo archivo
    tar.extract('docs/manual.pdf')

Filtrado de archivos durante la extracción

Una característica útil es la capacidad de filtrar qué archivos extraer basándose en ciertos criterios. Esto es especialmente importante por razones de seguridad, ya que los archivos TAR pueden contener rutas absolutas o referencias a directorios superiores:

import tarfile
import os

def filtro_seguro(miembro):
    """Filtro para evitar rutas absolutas o referencias a directorios superiores."""
    ruta = miembro.name
    if ruta.startswith(('/', '..')):
        return None  # No extraer este archivo
    
    # Convertir a ruta segura
    ruta_segura = os.path.normpath(os.path.join('.', ruta))
    if ruta_segura.startswith(('/', '..')):
        return None
    
    # Actualizar el nombre del miembro con la ruta segura
    miembro.name = ruta_segura
    return miembro

# Extraer archivos de forma segura
with tarfile.open('archivo.tar.gz', 'r:gz') as tar:
    tar.extractall(filter=filtro_seguro)

Este filtro es crucial para evitar ataques de "path traversal" donde un archivo malicioso podría sobrescribir archivos importantes del sistema.

Lectura de archivos sin extraerlos

Al igual que con ZIP, podemos leer el contenido de archivos dentro de un TAR sin extraerlos al disco:

import tarfile

with tarfile.open('datos.tar.gz', 'r:gz') as tar:
    # Obtener un archivo específico
    archivo_info = tar.getmember('config.json')
    
    # Abrir el archivo como un objeto similar a un archivo
    with tar.extractfile(archivo_info) as f:
        contenido = f.read()
        # Si es texto, decodificarlo
        texto = contenido.decode('utf-8')
        print(texto)

Para procesar archivos de texto como CSV dentro del TAR:

import tarfile
import csv
from io import TextIOWrapper

with tarfile.open('datos.tar.gz', 'r:gz') as tar:
    # Obtener el archivo CSV
    csv_info = tar.getmember('datos.csv')
    
    # Abrir y procesar el CSV
    with tar.extractfile(csv_info) as binario:
        # Convertir de binario a texto
        csv_file = TextIOWrapper(binario)
        # Leer el CSV
        reader = csv.reader(csv_file)
        for fila in reader:
            print(fila)

Añadir archivos con control de metadatos

El módulo tarfile permite un control detallado sobre los metadatos de los archivos que añadimos:

import tarfile
import time
import os

with tarfile.open('personalizado.tar', 'w') as tar:
    # Crear un objeto TarInfo para metadatos personalizados
    info = tarfile.TarInfo('archivo_virtual.txt')
    
    # Establecer metadatos
    info.size = 14  # Tamaño en bytes
    info.mtime = time.time()  # Tiempo de modificación actual
    info.mode = 0o644  # Permisos (rw-r--r--)
    info.type = tarfile.REGTYPE  # Archivo regular
    info.uid = os.getuid()  # ID de usuario actual
    info.gid = os.getgid()  # ID de grupo actual
    
    # Contenido del archivo
    contenido = b'Hola, mundo!\n'
    
    # Añadir el archivo con sus metadatos y contenido
    tar.addfile(info, fileobj=open('archivo_real.txt', 'rb'))
    
    # O añadir contenido desde un objeto bytes
    from io import BytesIO
    tar.addfile(info, fileobj=BytesIO(contenido))

Casos de uso prácticos

Los archivos TAR son especialmente útiles en entornos Unix/Linux y para ciertos escenarios específicos:

• Backup de sistemas Unix/Linux: El formato TAR preserva permisos, propietarios y enlaces simbólicos.
• Distribución de software: Muchos paquetes de software en Linux se distribuyen como archivos .tar.gz o .tar.bz2.
• Archivado de proyectos: Ideal para preservar la estructura exacta de directorios y metadatos.

Veamos un ejemplo de script para crear un backup incremental de un directorio:

import tarfile
import os
import time
from datetime import datetime

def backup_incremental(directorio, archivo_base):
    """Crea un backup incremental basado en la fecha de modificación."""
    # Determinar la fecha del último backup
    ultima_fecha = 0
    if os.path.exists(archivo_base):
        ultima_fecha = os.path.getmtime(archivo_base)
    
    # Nombre para el nuevo backup incremental
    fecha_str = datetime.now().strftime("%Y%m%d_%H%M%S")
    archivo_incremental = f"incremental_{fecha_str}.tar.gz"
    
    # Crear el archivo TAR incremental
    with tarfile.open(archivo_incremental, 'w:gz') as tar:
        archivos_nuevos = 0
        
        # Recorrer el directorio
        for carpeta_actual, subcarpetas, archivos in os.walk(directorio):
            for archivo in archivos:
                ruta_completa = os.path.join(carpeta_actual, archivo)
                # Verificar si el archivo es más reciente que el último backup
                if os.path.getmtime(ruta_completa) > ultima_fecha:
                    # Añadir solo archivos modificados desde el último backup
                    ruta_relativa = os.path.relpath(ruta_completa, os.path.dirname(directorio))
                    tar.add(ruta_completa, arcname=ruta_relativa)
                    archivos_nuevos += 1
        
        print(f"Backup incremental creado: {archivo_incremental}")
        print(f"Archivos añadidos: {archivos_nuevos}")
    
    # Si no hay backup base, este se convierte en el base
    if not os.path.exists(archivo_base):
        os.rename(archivo_incremental, archivo_base)
        print(f"Creado backup base: {archivo_base}")
    
    return archivo_incremental

# Ejemplo de uso
backup_incremental('./mi_proyecto', 'backup_base.tar.gz')

Este script crea un backup que solo incluye archivos modificados desde el último backup, lo que ahorra espacio y tiempo.

Comparación con otros formatos

El formato TAR tiene algunas características distintivas en comparación con ZIP:

• Preservación de metadatos: TAR conserva permisos, propietarios y enlaces simbólicos, lo que lo hace ideal para sistemas Unix/Linux.
• Compresión separada: TAR separa el agrupamiento (tarring) de la compresión, permitiendo elegir diferentes algoritmos.
• Streaming: TAR fue diseñado para trabajar con cintas, por lo que es adecuado para operaciones de streaming secuencial.

Sin embargo, ZIP tiene ventajas como la compresión individual de archivos y mejor soporte en Windows.

Compresión y descompresión con gzip

El módulo gzip de Python proporciona una interfaz simple para trabajar con archivos comprimidos en formato gzip. A diferencia de los formatos ZIP o TAR que pueden contener múltiples archivos, gzip está diseñado para comprimir un único archivo a la vez, funcionando como una "bolsa compresora" individual para cada archivo que queramos reducir de tamaño.

El formato gzip utiliza el algoritmo de compresión DEFLATE, que ofrece un buen equilibrio entre velocidad y ratio de compresión, haciéndolo ideal para reducir el tamaño de archivos de texto, logs, o datos que necesitamos transmitir o almacenar de manera eficiente.

Compresión básica de archivos

La forma más sencilla de comprimir un archivo con gzip es utilizando la función open() del módulo:

import gzip

# Comprimir un archivo de texto
with open('datos.txt', 'rb') as archivo_original:
    with gzip.open('datos.txt.gz', 'wb') as archivo_comprimido:
        archivo_comprimido.write(archivo_original.read())

También podemos comprimir datos directamente desde una variable en memoria:

import gzip

# Datos que queremos comprimir
datos = b"Este es un texto de ejemplo que ocupara menos espacio al comprimirse."

# Comprimir los datos
with gzip.open('datos_memoria.gz', 'wb') as f:
    f.write(datos)

Control del nivel de compresión

El módulo gzip permite ajustar el nivel de compresión según nuestras necesidades, balanceando entre velocidad y tamaño final:

import gzip

# Comprimir con nivel máximo (9)
with open('archivo_grande.log', 'rb') as entrada:
    with gzip.open('archivo_grande.log.gz', 'wb', compresslevel=9) as salida:
        salida.write(entrada.read())

# Comprimir rápidamente con nivel bajo (1)
with open('archivo_temporal.txt', 'rb') as entrada:
    with gzip.open('archivo_temporal.txt.gz', 'wb', compresslevel=1) as salida:
        salida.write(entrada.read())

Los niveles de compresión van de 1 (más rápido, menos compresión) a 9 (más lento, mejor compresión). El nivel predeterminado es 9, que prioriza el tamaño final sobre la velocidad.

Descompresión de archivos

Descomprimir un archivo gzip es igualmente sencillo:

import gzip

# Descomprimir un archivo
with gzip.open('datos.txt.gz', 'rb') as archivo_comprimido:
    with open('datos_descomprimidos.txt', 'wb') as archivo_descomprimido:
        archivo_descomprimido.write(archivo_comprimido.read())

Trabajando con archivos de texto

Cuando trabajamos con archivos de texto, podemos usar el parámetro 't' para manejar automáticamente la codificación:

import gzip

# Comprimir texto directamente
texto = "Este es un ejemplo de texto que será comprimido con gzip.\n" * 100
with gzip.open('ejemplo.txt.gz', 'wt', encoding='utf-8') as f:
    f.write(texto)

# Leer texto comprimido
with gzip.open('ejemplo.txt.gz', 'rt', encoding='utf-8') as f:
    contenido = f.read()
    print(f"Primeras 50 caracteres: {contenido[:50]}...")

Los modos de apertura más comunes son:

• 'rb' - Lectura binaria (predeterminado)
• 'wb' - Escritura binaria
• 'ab' - Añadir a un archivo existente en modo binario
• 'rt' - Lectura de texto (con decodificación)
• 'wt' - Escritura de texto (con codificación)
• 'at' - Añadir texto a un archivo existente

Compresión y descompresión en memoria

En muchos casos, necesitamos comprimir o descomprimir datos sin escribirlos en disco. Python permite hacer esto fácilmente:

import gzip

# Comprimir datos en memoria
datos_originales = b"Datos que ocupan espacio y queremos comprimir" * 1000
datos_comprimidos = gzip.compress(datos_originales, compresslevel=6)

print(f"Tamaño original: {len(datos_originales)} bytes")
print(f"Tamaño comprimido: {len(datos_comprimidos)} bytes")
print(f"Ratio de compresión: {(1 - len(datos_comprimidos) / len(datos_originales)) * 100:.2f}%")

# Descomprimir datos en memoria
datos_descomprimidos = gzip.decompress(datos_comprimidos)

# Verificar que los datos son idénticos
if datos_originales == datos_descomprimidos:
    print("¡Descompresión exitosa! Los datos son idénticos.")

Las funciones compress() y decompress() son ideales para aplicaciones web, procesamiento de datos o cualquier situación donde necesitemos manipular datos comprimidos sin archivos intermedios.

Procesamiento de archivos grandes por bloques

Cuando trabajamos con archivos muy grandes, es mejor procesarlos por bloques para evitar cargar todo en memoria:

import gzip

def comprimir_por_bloques(archivo_entrada, archivo_salida, tamaño_bloque=8192):
    """Comprime un archivo grande procesándolo por bloques."""
    with open(archivo_entrada, 'rb') as f_in:
        with gzip.open(archivo_salida, 'wb') as f_out:
            while True:
                bloque = f_in.read(tamaño_bloque)
                if not bloque:
                    break
                f_out.write(bloque)

def descomprimir_por_bloques(archivo_entrada, archivo_salida, tamaño_bloque=8192):
    """Descomprime un archivo grande procesándolo por bloques."""
    with gzip.open(archivo_entrada, 'rb') as f_in:
        with open(archivo_salida, 'wb') as f_out:
            while True:
                bloque = f_in.read(tamaño_bloque)
                if not bloque:
                    break
                f_out.write(bloque)

# Ejemplo de uso
comprimir_por_bloques('archivo_grande.log', 'archivo_grande.log.gz')
descomprimir_por_bloques('archivo_grande.log.gz', 'archivo_grande_recuperado.log')

Este enfoque es mucho más eficiente en términos de memoria para archivos de varios gigabytes.

Comprobación de archivos gzip

Podemos verificar si un archivo tiene formato gzip válido antes de intentar descomprimirlo:

import gzip

def es_archivo_gzip(ruta_archivo):
    """Verifica si un archivo tiene formato gzip válido."""
    try:
        with gzip.open(ruta_archivo, 'rb') as f:
            # Intentar leer un byte
            f.read(1)
            return True
    except gzip.BadGzipFile:
        return False
    except Exception as e:
        print(f"Error al verificar el archivo: {e}")
        return False

# Ejemplo de uso
archivos = ['datos.txt.gz', 'archivo_normal.txt', 'corrupto.gz']
for archivo in archivos:
    if es_archivo_gzip(archivo):
        print(f"{archivo} es un archivo gzip válido")
    else:
        print(f"{archivo} NO es un archivo gzip válido")

Casos de uso prácticos

El formato gzip es especialmente útil en varios escenarios cotidianos:

1. Compresión de logs

Los archivos de registro suelen crecer rápidamente y contener texto repetitivo, haciéndolos ideales para compresión:

import gzip
import logging
from datetime import datetime

class GzipRotatingFileHandler(logging.Handler):
    """Manejador de logs que comprime automáticamente archivos antiguos."""
    
    def __init__(self, filename, max_bytes=10485760):
        super().__init__()
        self.filename = filename
        self.max_bytes = max_bytes
        self.current_size = 0
        
        # Crear o abrir el archivo de log
        self.file = open(filename, 'a', encoding='utf-8')
    
    def emit(self, record):
        msg = self.format(record) + '\n'
        encoded_msg = msg.encode('utf-8')
        msg_size = len(encoded_msg)
        
        # Verificar si debemos rotar el archivo
        if self.current_size + msg_size > self.max_bytes:
            self.rotate()
        
        # Escribir el mensaje y actualizar tamaño
        self.file.write(msg)
        self.file.flush()
        self.current_size += msg_size
    
    def rotate(self):
        """Cierra el archivo actual y lo comprime."""
        self.file.close()
        
        # Nombre del archivo comprimido con timestamp
        timestamp = datetime.now().strftime("%Y%m%d-%H%M%S")
        gz_filename = f"{self.filename}.{timestamp}.gz"
        
        # Comprimir el archivo
        with open(self.filename, 'rb') as f_in:
            with gzip.open(gz_filename, 'wb') as f_out:
                f_out.write(f_in.read())
        
        # Crear un nuevo archivo vacío
        self.file = open(self.filename, 'w', encoding='utf-8')
        self.current_size = 0

# Ejemplo de uso
logger = logging.getLogger('mi_aplicacion')
handler = GzipRotatingFileHandler('aplicacion.log', max_bytes=1024*100)  # Rotar cada 100KB
formatter = logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s')
handler.setFormatter(formatter)
logger.addHandler(handler)
logger.setLevel(logging.INFO)

# Generar algunos logs
for i in range(10000):
    logger.info(f"Mensaje de log de ejemplo número {i}")

2. Transferencia eficiente de datos

Cuando necesitamos enviar datos a través de la red, comprimir primero puede ahorrar ancho de banda:

import gzip
import requests
import json

def enviar_datos_comprimidos(url, datos):
    """Envía datos JSON comprimidos a un servidor."""
    # Convertir datos a JSON y luego a bytes
    json_data = json.dumps(datos).encode('utf-8')
    
    # Comprimir los datos
    datos_comprimidos = gzip.compress(json_data)
    
    # Configurar cabeceras para indicar que enviamos contenido comprimido
    headers = {
        'Content-Encoding': 'gzip',
        'Content-Type': 'application/json'
    }
    
    # Enviar la solicitud
    response = requests.post(url, data=datos_comprimidos, headers=headers)
    return response

# Ejemplo: enviar un conjunto grande de datos
datos_grandes = {
    "registros": [{"id": i, "valor": f"dato_{i}"} for i in range(10000)]
}

respuesta = enviar_datos_comprimidos('https://api.ejemplo.com/datos', datos_grandes)
print(f"Código de respuesta: {respuesta.status_code}")

3. Almacenamiento eficiente de datos científicos

En análisis de datos y ciencia de datos, podemos comprimir resultados intermedios para ahorrar espacio:

import gzip
import numpy as np
import pickle

def guardar_array_comprimido(array, nombre_archivo):
    """Guarda un array de NumPy en formato comprimido."""
    with gzip.open(f"{nombre_archivo}.npy.gz", 'wb') as f:
        np.save(f, array)

def cargar_array_comprimido(nombre_archivo):
    """Carga un array de NumPy desde un archivo comprimido."""
    with gzip.open(f"{nombre_archivo}.npy.gz", 'rb') as f:
        return np.load(f)

# Crear un array grande
datos = np.random.rand(1000, 1000)

# Guardar de forma comprimida
guardar_array_comprimido(datos, 'matriz_aleatoria')

# Cargar desde archivo comprimido
datos_cargados = cargar_array_comprimido('matriz_aleatoria')

print(f"¿Arrays idénticos? {np.array_equal(datos, datos_cargados)}")

Para objetos Python más complejos, podemos combinar gzip con pickle:

def guardar_objeto_comprimido(objeto, nombre_archivo):
    """Guarda cualquier objeto Python en formato comprimido."""
    with gzip.open(f"{nombre_archivo}.pkl.gz", 'wb') as f:
        pickle.dump(objeto, f)

def cargar_objeto_comprimido(nombre_archivo):
    """Carga un objeto Python desde un archivo comprimido."""
    with gzip.open(f"{nombre_archivo}.pkl.gz", 'rb') as f:
        return pickle.load(f)

# Ejemplo con un diccionario complejo
datos_complejos = {
    'resultados': [{'experimento': i, 'valores': np.random.rand(100)} for i in range(50)],
    'metadatos': {'fecha': '2023-10-15', 'versión': '1.2.3'}
}

guardar_objeto_comprimido(datos_complejos, 'resultados_experimento')
datos_recuperados = cargar_objeto_comprimido('resultados_experimento')

El formato gzip es particularmente eficiente para datos científicos que contienen patrones repetitivos o valores similares, como matrices de datos experimentales o resultados de simulaciones.

Lectura y escritura de archivos comprimidos

Cuando trabajamos con archivos comprimidos en Python, a menudo necesitamos leer o escribir datos sin tener que descomprimir completamente los archivos en el disco. Esta capacidad es especialmente útil para procesar grandes volúmenes de datos de manera eficiente, como archivos de registro, conjuntos de datos científicos o backups.

Python ofrece varias formas de interactuar directamente con el contenido de archivos comprimidos, permitiéndonos tratarlos casi como si fueran archivos normales. Veamos cómo podemos implementar estas operaciones con diferentes formatos de compresión.

Lectura y escritura con objetos tipo archivo

La característica más potente de los módulos de compresión en Python es que proporcionan interfaces similares a los archivos estándar. Esto significa que podemos usar métodos familiares como read(), write(), readline() y otros:

import gzip

# Escribir líneas de texto en un archivo comprimido
with gzip.open('datos.txt.gz', 'wt', encoding='utf-8') as f:
    for i in range(1000):
        f.write(f"Esta es la línea {i} de nuestro archivo de ejemplo.\n")

# Leer el archivo línea por línea
with gzip.open('datos.txt.gz', 'rt', encoding='utf-8') as f:
    # Leer las primeras 5 líneas
    for _ in range(5):
        linea = f.readline()
        print(linea.strip())

Este patrón funciona de manera similar con otros formatos de compresión:

import bz2

# Escribir en un archivo bz2
with bz2.open('datos.txt.bz2', 'wt', encoding='utf-8') as f:
    f.write("Datos comprimidos con bzip2, que ofrece mejor compresión que gzip.\n")
    f.write("Ideal para archivos que no necesitan acceso aleatorio frecuente.\n")
    
# Leer desde un archivo bz2
with bz2.open('datos.txt.bz2', 'rt', encoding='utf-8') as f:
    contenido = f.read()
    print(contenido)

Procesamiento de datos estructurados

Una aplicación común es procesar datos estructurados como CSV o JSON directamente desde archivos comprimidos:

import gzip
import csv
import json
from io import TextIOWrapper

# Escribir datos CSV en un archivo comprimido
datos = [
    ['Nombre', 'Edad', 'Ciudad'],
    ['Ana', '28', 'Madrid'],
    ['Carlos', '35', 'Barcelona'],
    ['Elena', '42', 'Valencia']
]

with gzip.open('personas.csv.gz', 'wt', newline='', encoding='utf-8') as f_gz:
    writer = csv.writer(f_gz)
    writer.writerows(datos)

# Leer datos CSV desde un archivo comprimido
with gzip.open('personas.csv.gz', 'rt', newline='', encoding='utf-8') as f_gz:
    reader = csv.reader(f_gz)
    for fila in reader:
        print(f"Fila: {fila}")

Para archivos JSON:

import gzip
import json

# Datos de ejemplo
datos_json = {
    "usuarios": [
        {"id": 1, "nombre": "Laura", "activo": True},
        {"id": 2, "nombre": "Miguel", "activo": False},
        {"id": 3, "nombre": "Sara", "activo": True}
    ],
    "total": 3,
    "página": 1
}

# Escribir JSON comprimido
with gzip.open('usuarios.json.gz', 'wt', encoding='utf-8') as f:
    json.dump(datos_json, f, indent=2)

# Leer JSON comprimido
with gzip.open('usuarios.json.gz', 'rt', encoding='utf-8') as f:
    datos_cargados = json.load(f)
    print(f"Total de usuarios: {datos_cargados['total']}")
    print(f"Primer usuario: {datos_cargados['usuarios'][0]['nombre']}")

Trabajando con archivos ZIP de forma eficiente

Aunque ya vimos el módulo zipfile en detalle, es importante destacar cómo podemos leer y escribir archivos dentro de un ZIP sin extraerlos completamente:

import zipfile
import csv
from io import StringIO, BytesIO

# Crear un archivo ZIP en memoria
buffer_zip = BytesIO()

# Escribir varios archivos CSV en el ZIP
with zipfile.ZipFile(buffer_zip, 'w', zipfile.ZIP_DEFLATED) as zip_file:
    # Primer archivo CSV
    datos_ventas = [
        ['Producto', 'Cantidad', 'Precio'],
        ['Laptop', '5', '899.99'],
        ['Monitor', '10', '249.50'],
        ['Teclado', '15', '45.75']
    ]
    
    csv_buffer = StringIO()
    writer = csv.writer(csv_buffer)
    writer.writerows(datos_ventas)
    
    # Añadir el CSV al ZIP
    zip_file.writestr('ventas.csv', csv_buffer.getvalue())
    
    # Segundo archivo: texto simple
    zip_file.writestr('README.txt', 'Este archivo ZIP contiene datos de ventas del último trimestre.')

# Ahora leemos los archivos del ZIP (sin extraerlos)
buffer_zip.seek(0)  # Volver al inicio del buffer

with zipfile.ZipFile(buffer_zip, 'r') as zip_file:
    # Listar contenido
    print("Archivos en el ZIP:")
    for archivo in zip_file.namelist():
        print(f" - {archivo}")
    
    # Leer el archivo CSV
    with zip_file.open('ventas.csv') as f:
        # Convertir a texto para CSV
        csv_texto = TextIOWrapper(f, encoding='utf-8')
        reader = csv.reader(csv_texto)
        print("\nContenido de ventas.csv:")
        for fila in reader:
            print(fila)
    
    # Leer el archivo de texto
    with zip_file.open('README.txt') as f:
        contenido = f.read().decode('utf-8')
        print(f"\nContenido de README.txt:\n{contenido}")

Este ejemplo muestra cómo podemos crear y manipular archivos ZIP completamente en memoria, sin necesidad de archivos temporales en disco.

Compresión en streaming

Para casos donde los datos se generan o procesan de forma continua, podemos implementar un enfoque de streaming:

import gzip
import time
import random

def generador_datos():
    """Simula una fuente de datos continua."""
    for _ in range(100):
        yield f"Datos generados en {time.time()}: {random.random()}\n"
        time.sleep(0.1)  # Simular tiempo de procesamiento

# Escribir datos en streaming
with gzip.open('streaming.log.gz', 'wt', encoding='utf-8') as f:
    for datos in generador_datos():
        f.write(datos)
        f.flush()  # Asegurar que los datos se escriban inmediatamente
        print(".", end="", flush=True)  # Indicador de progreso

Combinando compresión con procesamiento paralelo

Para archivos muy grandes, podemos combinar la compresión con procesamiento paralelo:

import gzip
import concurrent.futures
import os

def procesar_bloque(bloque):
    """Procesa un bloque de texto (ejemplo: contar palabras)."""
    palabras = bloque.split()
    return len(palabras)

def procesar_archivo_comprimido_paralelo(archivo_gz, num_workers=4):
    """Procesa un archivo gzip grande usando múltiples workers."""
    resultados = []
    
    with gzip.open(archivo_gz, 'rt', encoding='utf-8') as f:
        # Leer bloques de texto (por ejemplo, 1000 líneas por bloque)
        with concurrent.futures.ProcessPoolExecutor(max_workers=num_workers) as executor:
            bloques = []
            lineas_acumuladas = []
            lineas_por_bloque = 1000
            
            for linea in f:
                lineas_acumuladas.append(linea)
                
                if len(lineas_acumuladas) >= lineas_por_bloque:
                    bloques.append(''.join(lineas_acumuladas))
                    lineas_acumuladas = []
            
            # Añadir el último bloque si queda algo
            if lineas_acumuladas:
                bloques.append(''.join(lineas_acumuladas))
            
            # Procesar bloques en paralelo
            resultados = list(executor.map(procesar_bloque, bloques))
    
    return sum(resultados)  # Total de palabras

# Ejemplo de uso (asumiendo que tenemos un archivo grande)
if os.path.exists('archivo_grande.txt.gz'):
    total_palabras = procesar_archivo_comprimido_paralelo('archivo_grande.txt.gz')
    print(f"Total de palabras en el archivo: {total_palabras}")

Trabajando con archivos TAR comprimidos

Para archivos TAR comprimidos, podemos leer y escribir archivos individuales sin extraer todo el archivo:

import tarfile
import io

# Crear un archivo tar.gz con varios archivos
with tarfile.open('documentos.tar.gz', 'w:gz') as tar:
    # Añadir un archivo desde una cadena
    datos = "Este es el contenido del primer archivo.\n"
    info = tarfile.TarInfo(name="archivo1.txt")
    info.size = len(datos)
    tar.addfile(info, io.BytesIO(datos.encode('utf-8')))
    
    # Añadir otro archivo
    datos2 = "Contenido del segundo archivo con más texto.\n" * 10
    info2 = tarfile.TarInfo(name="carpeta/archivo2.txt")
    info2.size = len(datos2)
    tar.addfile(info2, io.BytesIO(datos2.encode('utf-8')))

# Leer archivos específicos del tar.gz
with tarfile.open('documentos.tar.gz', 'r:gz') as tar:
    # Leer el primer archivo
    miembro = tar.getmember("archivo1.txt")
    f = tar.extractfile(miembro)
    contenido = f.read().decode('utf-8')
    print(f"Contenido de archivo1.txt:\n{contenido}")
    
    # Buscar y leer archivos que coincidan con un patrón
    for miembro in tar.getmembers():
        if miembro.name.endswith('.txt') and '/' in miembro.name:
            print(f"\nArchivo encontrado en subcarpeta: {miembro.name}")
            f = tar.extractfile(miembro)
            if f:  # Podría ser None si es un directorio
                primeras_lineas = f.read(100).decode('utf-8')
                print(f"Primeras líneas: {primeras_lineas}...")

Compresión adaptativa según el tipo de contenido

Diferentes tipos de datos se comprimen de manera distinta. Podemos adaptar nuestra estrategia según el contenido:

import gzip
import bz2
import lzma
import os

def comprimir_archivo(archivo_entrada, metodo='auto'):
    """Comprime un archivo usando el método más adecuado según su tipo."""
    # Determinar extensión para inferir tipo
    _, extension = os.path.splitext(archivo_entrada)
    extension = extension.lower()
    
    # Leer el archivo original
    with open(archivo_entrada, 'rb') as f:
        datos = f.read()
    
    # Seleccionar método de compresión
    if metodo == 'auto':
        if extension in ['.txt', '.csv', '.json', '.xml', '.html', '.md']:
            metodo = 'gzip'  # Bueno para texto
        elif extension in ['.log', '.sql']:
            metodo = 'bzip2'  # Mejor para datos repetitivos
        elif extension in ['.bin', '.dat']:
            metodo = 'lzma'   # Mejor para datos binarios
        else:
            metodo = 'gzip'   # Por defecto
    
    # Comprimir según el método elegido
    if metodo == 'gzip':
        datos_comprimidos = gzip.compress(datos, compresslevel=9)
        extension_salida = '.gz'
    elif metodo == 'bzip2':
        datos_comprimidos = bz2.compress(datos, compresslevel=9)
        extension_salida = '.bz2'
    elif metodo == 'lzma':
        datos_comprimidos = lzma.compress(datos, preset=9)
        extension_salida = '.xz'
    else:
        raise ValueError(f"Método de compresión no válido: {metodo}")
    
    # Guardar archivo comprimido
    archivo_salida = archivo_entrada + extension_salida
    with open(archivo_salida, 'wb') as f:
        f.write(datos_comprimidos)
    
    # Calcular ratio de compresión
    ratio = (1 - len(datos_comprimidos) / len(datos)) * 100
    
    return {
        'archivo_original': archivo_entrada,
        'archivo_comprimido': archivo_salida,
        'tamaño_original': len(datos),
        'tamaño_comprimido': len(datos_comprimidos),
        'ratio_compresión': ratio,
        'método': metodo
    }

# Ejemplo de uso
archivos = ['datos.txt', 'registro.log', 'configuracion.json']
for archivo in archivos:
    if os.path.exists(archivo):
        resultado = comprimir_archivo(archivo)
        print(f"Archivo: {resultado['archivo_original']}")
        print(f"Método: {resultado['método']}")
        print(f"Ratio de compresión: {resultado['ratio_compresión']:.2f}%")
        print(f"Tamaño original: {resultado['tamaño_original']} bytes")
        print(f"Tamaño comprimido: {resultado['tamaño_comprimido']} bytes")
        print("-" * 40)

Lectura y escritura de archivos comprimidos en aplicaciones web

En aplicaciones web, a menudo necesitamos generar archivos comprimidos al vuelo para descargas:

import io
import zipfile
import csv

def generar_informe_zip():
    """Genera un archivo ZIP con informes en memoria para descarga web."""
    # Crear buffer en memoria
    buffer = io.BytesIO()
    
    # Crear archivo ZIP en el buffer
    with zipfile.ZipFile(buffer, 'w', zipfile.ZIP_DEFLATED) as zip_file:
        # Generar informe CSV
        csv_buffer = io.StringIO()
        writer = csv.writer(csv_buffer)
        writer.writerow(['Mes', 'Ventas', 'Gastos', 'Beneficio'])
        writer.writerow(['Enero', '10000', '7500', '2500'])
        writer.writerow(['Febrero', '12000', '8000', '4000'])
        writer.writerow(['Marzo', '15000', '9000', '6000'])
        
        # Añadir CSV al ZIP
        zip_file.writestr('informe_trimestral.csv', csv_buffer.getvalue())
        
        # Generar informe de texto
        resumen = "RESUMEN TRIMESTRAL\n\n"
        resumen += "Total Ventas: 37000€\n"
        resumen += "Total Gastos: 24500€\n"
        resumen += "Beneficio Neto: 12500€\n"
        
        zip_file.writestr('resumen.txt', resumen)
    
    # Mover el cursor al inicio del buffer
    buffer.seek(0)
    return buffer

# En un framework web como Flask, podríamos usarlo así:
"""
@app.route('/descargar-informe')
def descargar_informe():
    buffer = generar_informe_zip()
    return send_file(
        buffer,
        as_attachment=True,
        download_name='informe_trimestral.zip',
        mimetype='application/zip'
    )
"""

Compresión y descompresión con contraseña

Para datos sensibles, podemos usar bibliotecas como pyzipper que extienden zipfile con soporte para encriptación AES:

# Necesita: pip install pyzipper
import pyzipper

def crear_zip_encriptado(archivos, archivo_salida, contraseña):
    """Crea un archivo ZIP encriptado con AES."""
    with pyzipper.AESZipFile(archivo_salida, 
                           'w', 
                           compression=pyzipper.ZIP_LZMA,
                           encryption=pyzipper.WZ_AES) as zip_file:
        zip_file.setpassword(contraseña.encode('utf-8'))
        
        for archivo in archivos:
            zip_file.write(archivo)
            
    print(f"Archivo ZIP encriptado creado: {archivo_salida}")

def leer_zip_encriptado(archivo_zip, contraseña):
    """Lee el contenido de un ZIP encriptado."""
    with pyzipper.AESZipFile(archivo_zip) as zip_file:
        zip_file.setpassword(contraseña.encode('utf-8'))
        
        print(f"Contenido del ZIP encriptado:")
        for archivo in zip_file.namelist():
            print(f" - {archivo}")
            
            # Si es un archivo de texto, mostrar contenido
            if archivo.endswith(('.txt', '.csv', '.json')):
                with zip_file.open(archivo) as f:
                    contenido = f.read().decode('utf-8')
                    print(f"   Primeras 50 caracteres: {contenido[:50]}...")

# Ejemplo de uso
archivos_confidenciales = ['contratos.txt', 'clientes.csv']
# crear_zip_encriptado(archivos_confidenciales, 'confidencial.zip', 'clave_segura123')
# leer_zip_encriptado('confidencial.zip', 'clave_segura123')

Compresión diferencial para backups incrementales

Para backups regulares, podemos implementar una estrategia de compresión diferencial que solo almacene los cambios:

import gzip
import os
import hashlib
import pickle
import time

class BackupIncremental:
    def __init__(self, directorio_base, directorio_backup):
        self.directorio_base = directorio_base
        self.directorio_backup = directorio_backup
        self.archivo_estado = os.path.join(directorio_backup, 'estado.pkl.gz')
        self.estado_anterior = self._cargar_estado()
    
    def _cargar_estado(self):
        """Carga el estado del backup anterior si existe."""
        if os.path.exists(self.archivo_estado):
            with gzip.open(self.archivo_estado, 'rb') as f:
                return pickle.load(f)
        return {}  # Estado vacío si es el primer backup
    
    def _guardar_estado(self, nuevo_estado):
        """Guarda el estado actual del backup."""
        with gzip.open(self.archivo_estado, 'wb') as f:
            pickle.dump(nuevo_estado, f)
    
    def _calcular_hash(self, ruta_archivo):
        """Calcula el hash SHA-256 de un archivo."""
        h = hashlib.sha256()
        with open(ruta_archivo, 'rb') as f:
            # Leer por bloques para archivos grandes
            for bloque in iter(lambda: f.read(4096), b''):
                h.update(bloque)
        return h.hexdigest()
    
    def realizar_backup(self):
        """Realiza un backup incremental."""
        nuevo_estado = {}
        archivos_nuevos = 0
        archivos_modificados = 0
        archivos_sin_cambios = 0
        
        # Timestamp para este backup
        timestamp = time.strftime("%Y%m%d_%H%M%S")
        directorio_actual = os.path.join(self.directorio_backup, timestamp)
        os.makedirs(directorio_actual, exist_ok=True)
        
        # Recorrer todos los archivos
        for carpeta_actual, _, archivos in os.walk(self.directorio_base):
            for archivo in archivos:
                # Ignorar el propio directorio de backup
                if self.directorio_backup in carpeta_actual:
                    continue
                
                ruta_completa = os.path.join(carpeta_actual, archivo)
                ruta_relativa = os.path.relpath(ruta_completa, self.directorio_base)
                
                # Calcular hash del archivo
                hash_actual = self._calcular_hash(ruta_completa)
                nuevo_estado[ruta_relativa] = hash_actual
                
                # Verificar si el archivo ha cambiado
                if (ruta_relativa not in self.estado_anterior or 
                    self.estado_anterior[ruta_relativa] != hash_actual):
                    
                    # Crear estructura de directorios en el backup
                    ruta_destino = os.path.join(directorio_actual, ruta_relativa)
                    os.makedirs(os.path.dirname(ruta_destino), exist_ok=True)
                    
                    # Comprimir y guardar el archivo
                    with open(ruta_completa, 'rb') as f_in:
                        with gzip.open(ruta_destino + '.gz', 'wb') as f_out:
                            f_out.write(f_in.read())
                    
                    if ruta_relativa in self.estado_anterior:
                        archivos_modificados += 1
                    else:
                        archivos_nuevos += 1
                else:
                    archivos_sin_cambios += 1
        
        # Guardar el nuevo estado
        self._guardar_estado(nuevo_estado)
        
        return {
            'timestamp': timestamp,
            'archivos_nuevos': archivos_nuevos,
            'archivos_modificados': archivos_modificados,
            'archivos_sin_cambios': archivos_sin_cambios,
            'total_archivos': len(nuevo_estado)
        }

# Ejemplo de uso
# backup = BackupIncremental('./mi_proyecto', './backups')
# resultado = backup.realizar_backup()
# print(f"Backup completado en {resultado['timestamp']}")
# print(f"Archivos nuevos: {resultado['archivos_nuevos']}")
# print(f"Archivos modificados: {resultado['archivos_modificados']}")
# print(f"Archivos sin cambios: {resultado['archivos_sin_cambios']}")

Este sistema de backup incremental solo comprime y almacena los archivos que han cambiado desde el último backup, ahorrando espacio y tiempo.

Consideraciones de rendimiento

Al trabajar con archivos comprimidos, es importante considerar algunos aspectos de rendimiento:

• Nivel de compresión: Un nivel más alto proporciona mejor compresión pero es más lento.
• Tamaño de buffer: Para archivos grandes, ajustar el tamaño del buffer puede mejorar el rendimiento.
• Formato adecuado: Cada formato tiene ventajas según el tipo de datos (gzip es rápido, bzip2 comprime mejor, lzma ofrece la mejor compresión pero es más lento).
• Procesamiento en memoria vs. disco: Para archivos muy grandes, el procesamiento por bloques evita problemas de memoria.

La elección del enfoque adecuado dependerá de tus necesidades específicas: velocidad, ratio de compresión o facilidad de uso.