0. Introducción
Esto parece una película. Primeramente busqué en Google "Algoritmo de Huffman en Visual Basic" y me hace referencia a esta URL de jrubi que ha encontrado el algoritmo de Huffman realizado por James Vincent Carnicelli en agosto del 2000, y le pasé el código a Claude para que me lo convirtiera a Python.
Como Claude tenía ya muestras de ficheros originales y comprimidos, me contesta que el algoritmo casi daba resultados correctos pero que era una versión antigua, en vez de darle cabecera "HF3" daba "HF2", y me propone buscar el algoritmo más reciente en internet y le digo que si.
Entonces encuentra esta URL de post.bytes que dice que es :"Es un hilo de 2005 (espejo de un antiguo grupo de noticias de Microsoft sobre VB.NET) donde un usuario llamado "Crouchie1998" responde a una pregunta pegando el código VB6 completo de la clase clsHuffman, atribuido originalmente a David Midkiff. Ese mismo mensaje también enlaza a dos copias del programa VB6 ya compilado (Planet Source Code y a1vbcode.com), aunque esos enlaces concretos ya no funcionan hoy en día — el texto del código en sí es lo que pude usar."
Y primeramente prueba el algoritmo en VB.NET sobre los ficheros que tenía y me suelta "BINGO!" y me genera el algoritmo en Python completamente funcional.
Con esto conseguimos no utilizar código que puede ser propiedad intelectual de Aytos como compress.dll (aunque el algoritmo subyacente no sea de su propiedad) y poder trabajar perfectamente en Linux para obtener estos ficheros y tener un histórico documental
1. Algoritmo de Huffman de compresión y decompresión
Veamos pues el preciado algoritmo pero teniendo en cuenta que:
- En la función main_edu() hay que darle la relació de ficheros a descomprimir
- Los nombres de ficheros serán del tipo "12345.EXT.bin" donde "12345" representa el campo IDX_ID de la tabla BIB_AY.dbo.IDX que es numérico; "EXT" represnta la extensión por ejemplo "DOC", "PDF" o la que sea, y "bin" es constante y significa que está comprido. Ejemplos son "345123.PDF.bin", "24512.DOC.bin"
- Al final genera el fichero con el mismo nombre al que se le ha añadido la extensiónb correspondiente, por ejemplo en los dos casos anteriores generaría "345123.PDF.bin.PDF", "24512.DOC.bin.DOC"
#!/usr/bin/env python3 """ decode_he3.py -------------- Codificador/decodificador puro-Python del formato "HE3" usado por compress.dll (clase clsHuffman, de David Midkiff). Reimplementa exactamente el algoritmo Huffman original (encontrado en su código fuente VB6), sin depender de Windows, Wine ni la DLL original. Verificado byte a byte contra ficheros reales: tanto DecodeFile como EncodeFile producen un resultado IDÉNTICO al de la DLL original (incluido el checksum interno del propio formato). USO: Descomprimir un fichero: python3 decode_he3.py decode origen.bin destino.pdf Comprimir un fichero: python3 decode_he3.py encode origen.pdf destino.bin Toda una carpeta (descomprimir todos los *.bin): python3 decode_he3.py decode --batch carpeta_origen carpeta_destino Toda una carpeta (comprimir): python3 decode_he3.py encode --batch carpeta_origen carpeta_destino --ext .pdf --suffix .aytos FORMATO DEL FICHERO (para referencia): offset 0-3 : magia "HE3" + CR (0x48 0x45 0x33 0x0D) (si el 3er byte es '0'=0x30, el fichero no se llegó a comprimir y el resto son los datos originales tal cual) offset 4 : checksum (XOR de todos los bytes originales) offset 5-8 : Long (LE32) = tamaño del fichero original offset 9-10 : Integer (LE16) = número de símbolos (valores de byte) distintos usados en el fichero luego, por cada símbolo (en orden ascendente de valor de byte): 1 byte = valor del símbolo (0-255) 1 byte = longitud en bits de su código Huffman luego, en el mismo orden ascendente de símbolo: los propios patrones de bits de cada código, empaquetados LSB-first, concatenados y sin relleno entre símbolos (solo relleno al final, hasta completar el byte) a partir de ahí, alineado a byte: los datos reales comprimidos, empaquetados también LSB-first, decodificables recorriendo el árbol binario reconstruido (bit 0 = rama izquierda, bit 1 = rama derecha) hasta producir tantos bytes como indica el tamaño original. """ import argparse import os import struct import sys class HE3Error(Exception): pass class _BitReader: """Lee bits LSB-first de un buffer, empezando en un byte concreto.""" __slots__ = ('data', 'pos', 'bitpos') def __init__(self, data: bytes, start_byte: int = 0): self.data = data self.pos = start_byte self.bitpos = 0 def read_bit(self) -> int: byte = self.data[self.pos] bit = (byte >> self.bitpos) & 1 self.bitpos += 1 if self.bitpos == 8: self.bitpos = 0 self.pos += 1 return bit def align_byte(self) -> None: if self.bitpos != 0: self.bitpos = 0 self.pos += 1 class _Node: __slots__ = ('left', 'right', 'value') def __init__(self): self.left = None self.right = None self.value = -1 def decode_he3(data: bytes) -> bytes: """Decodifica el contenido binario (formato HE3) y devuelve los bytes originales. Lanza HE3Error si la cabecera o el checksum no son validos.""" if len(data) < 4 or data[0] != 72 or data[1] != 69 or data[3] != 13: raise HE3Error("Cabecera invalida: no es un fichero HE0/HE3") if data[2] == 48: # '0' -> el fichero se guardo sin comprimir return data[4:] if data[2] != 51: # '3' raise HE3Error(f"Version de formato no soportada (byte2={data[2]})") checksum = data[4] orig_size = struct.unpack('<I', data[5:9])[0] if orig_size == 0: return b'' count = struct.unpack('<H', data[9:11])[0] pos = 11 symbols = [] lengths = {} for _ in range(count): sym = data[pos]; pos += 1 length = data[pos]; pos += 1 symbols.append(sym) lengths[sym] = length # Leer los patrones de bits de cada codigo (orden raiz->hoja) br = _BitReader(data, pos) codes = {} for sym in symbols: bits = [br.read_bit() for _ in range(lengths[sym])] codes[sym] = bits br.align_byte() # Reconstruir el arbol binario: 0 = izquierda, 1 = derecha root = _Node() for sym, bits in codes.items(): node = root for b in bits: if b == 0: if node.left is None: node.left = _Node() node = node.left else: if node.right is None: node.right = _Node() node = node.right node.value = sym # Decodificar el resto del fichero recorriendo el arbol out = bytearray() node = root for byte in data[br.pos:]: for bitpos in range(8): bit = (byte >> bitpos) & 1 node = node.right if bit else node.left if node is None: raise HE3Error("Arbol Huffman corrupto durante la decodificacion") if node.value > -1: out.append(node.value) if len(out) == orig_size: chk = 0 for b in out: chk ^= b if chk != checksum: raise HE3Error( f"Checksum no coincide (esperado {checksum}, " f"calculado {chk}) - el fichero podria estar corrupto" ) return bytes(out) node = root raise HE3Error("Fin del fichero antes de completar la decodificacion") def encode_he3(data: bytes) -> bytes: """Comprime bytes al formato HE3. Devuelve el mismo resultado, byte a byte, que produciria clsHuffman.EncodeFile().""" byte_len = len(data) if byte_len == 0: return bytes([72, 69, 48, 13]) # "HE0" + CR, fichero vacio counts = [0] * 256 for b in data: counts[b] += 1 class _ENode: __slots__ = ('weight', 'left', 'right', 'parent', 'value') def __init__(self, weight, value=-1, left=-1, right=-1): self.weight = weight self.value = value self.left = left self.right = right self.parent = -1 nodes = [_ENode(counts[i], value=i) for i in range(256) if counts[i] > 0] nodes_count = len(nodes) nodes.extend([None] * (nodes_count - 1)) # espacio para nodos internos n = nodes_count while n > 1: lnode1 = lnode2 = -1 lweight1 = lweight2 = 0 # Replica exacta del algoritmo O(n^2) de "dos mas pequenos" del # codigo VB original, incluido su criterio de desempate. for i, node in enumerate(nodes): if node is None or node.parent != -1: continue if lnode1 == -1: lweight1 = node.weight; lnode1 = i elif lnode2 == -1: lweight2 = node.weight; lnode2 = i elif node.weight < lweight1: if node.weight < lweight2: if lweight1 < lweight2: lweight2 = node.weight; lnode2 = i else: lweight1 = node.weight; lnode1 = i else: lweight1 = node.weight; lnode1 = i elif node.weight < lweight2: lweight2 = node.weight; lnode2 = i new_index = next(i for i, node in enumerate(nodes) if node is None) nodes[new_index] = _ENode(lweight1 + lweight2, left=lnode1, right=lnode2) nodes[lnode1].parent = new_index nodes[lnode2].parent = new_index n -= 1 root_index = next(i for i, node in enumerate(nodes) if node is not None and node.parent == -1) codes = {} def walk(idx, bits): node = nodes[idx] if node.value > -1: codes[node.value] = bits return if node.left != -1: walk(node.left, bits + [0]) if node.right != -1: walk(node.right, bits + [1]) walk(root_index, []) used_symbols = sorted(codes.keys()) total_bits = sum(len(codes[i]) * counts[i] for i in used_symbols) l_length = (total_bits + 7) // 8 if l_length == 0 or l_length > byte_len: return bytes([72, 69, 48, 13]) + data # fallback: sin comprimir checksum = 0 for b in data: checksum ^= b out = bytearray() out += bytes([72, 69, 51, 13]) out.append(checksum) out += struct.pack('<I', byte_len) out += struct.pack('<H', len(used_symbols)) for sym in used_symbols: out.append(sym) out.append(len(codes[sym])) bitbuf = 0 bitcount = 0 def push_bit(bit): nonlocal bitbuf, bitcount if bit: bitbuf |= (1 << bitcount) bitcount += 1 if bitcount == 8: out.append(bitbuf) bitbuf = 0 bitcount = 0 for sym in used_symbols: for bit in codes[sym]: push_bit(bit) if bitcount > 0: out.append(bitbuf) bitbuf = 0 bitcount = 0 for b in data: for bit in codes[b]: push_bit(bit) if bitcount > 0: out.append(bitbuf) return bytes(out) def process_file(src: str, dst: str, mode: str) -> None: with open(src, 'rb') as f: data = f.read() result = decode_he3(data) if mode == 'decode' else encode_he3(data) out_dir = os.path.dirname(os.path.abspath(dst)) if out_dir: os.makedirs(out_dir, exist_ok=True) with open(dst, 'wb') as f: f.write(result) def main_edu() -> None: files=[ "/home/user/kk/DOCUMENTS_AYTOS/238476.PDF.bin", "/home/user/kk/DOCUMENTS_AYTOS/233831.DOC.bin", ] fail=0 ok=0 for fname in files: parts=fname.split(".") f_out=fname + "." + parts[1] try: process_file(fname, f_out, "decode") print(f"[OK] {fname} -> {os.path.basename(f_out)}") ok += 1 except Exception as exc: print(f"[ERROR] {fname}: {exc}") fail += 1 print(f"\nHecho. {ok} correctos, {fail} con error.") if __name__ == "__main__": main_edu()
2. Descargar los ficheros desde BD a disco
Veamos ahora el módulo completo para descargar los ficheros descomprimidos o descifrados a disco
Aquí ya no escribimos el fichero encriptado/comprimido a disco sinó que escribimos el fichero drirectamente a disco con su nombre y extensión correspondiente.
En mi caso he guardado el fichero anterior en la misma carpeta y lo he llamado xmhuffman.py Por tanto la tercera línea se debe de ajustar al nombre que se le haya dado a dicho fichero. En la función main hay que darle los parámetros de acceso a la BD y también la rudta donde guardar los ficheros.
Aquí ya no escribimos el fichero encriptado/comprimido a disco sinó que escribimos el fichero drirectamente a disco con su nombre y extensión correspondiente.
En mi caso he guardado el fichero anterior en la misma carpeta y lo he llamado xmhuffman.py Por tanto la tercera línea se debe de ajustar al nombre que se le haya dado a dicho fichero. En la función main hay que darle los parámetros de acceso a la BD y también la rudta donde guardar los ficheros.
from sqlalchemy import create_engine, text from urllib.parse import quote_plus from xmhuffman import decode_he3 # -------------------------------------------------------------------------- # Utilidades de configuración # -------------------------------------------------------------------------- def _connection_string(c: dict) -> str: usuario = quote_plus(str(c["dbUser"])) password = quote_plus(str(c["dbPwd"])) host = f"{c['dbHost']}:{c['dbPort']}" if c["dbPort"] else c["dbHost"] url = f"{c['dbType']}://{usuario}:{password}@{host}/{c['dbName']}" params = {} if c["dbDriver"]: params["driver"] = c["dbDriver"] if c["dbTrustServerCertificate"]: params["TrustServerCertificate"] = c["dbTrustServerCertificate"] if params: url += "?" + "&".join(f"{k}={quote_plus(str(v))}" for k, v in params.items()) return url def _create_sync_engine(conf: dict): """Engine síncrono, para SQL Server (pyodbc).""" return create_engine(_connection_string(conf)) def copy_all_blobs2files( query_ms: str, conf_ms: dict, ruta_salida: str, ): engine_ms = _create_sync_engine(conf_ms) archivos_creados = 0 try: with engine_ms.connect() as conn: # Usamos fetchall() para traer todos los registros de la consulta resultados = conn.execute(text(query_ms)).fetchall() if not resultados: raise ValueError(f"No se encontraron filas para la consulta: {query_ms}") i=0 for fila in resultados: if i==2000: break # fila[0] debe ser el ID (o nombre del archivo) y fila[1] el BLOB id_registro = fila[0] extension = fila[1] contenido = fila[2] nombre_archivo=f"{id_registro}.{extension}" print(nombre_archivo) # Omitimos registros si el BLOB está vacío if contenido is None: print(f"Advertencia: El campo BLOB está vacío para el ID {id_registro}. Saltando...") continue # Construimos la ruta completa del archivo nombre_archivo = f"{ruta_salida}{nombre_archivo}" result = decode_he3(contenido) with open(nombre_archivo, 'wb') as f: f.write(result) archivos_creados += 1 i+=1 return archivos_creados finally: engine_ms.dispose() if __name__ == "__main__": conf_ms = { "dbType": "mssql+pyodbc", "dbUser": "myuser", "dbPwd": "mypassword", "dbHost": "192.168.xxx.xxx", "dbPort": 1433, "dbName": "BIB_AY", "dbDriver": "ODBC Driver 18 for SQL Server", "dbTrustServerCertificate": "yes", } query_ms_all = """ SELECT B.IDX_ID, B.IDX_EXT, ISNULL(d1.DOC_CNT,d2.ORI_CNT) as doc_blob,B.IDX_ALI, d1.DOC_TAM, d2.ORI_TAM, 10000*YEAR(B.IDX_FCH)+100*MONTH(B.IDX_FCH)+DAY(B.IDX_FCH) as FOLDER FROM BIB_AY.dbo.IDX B LEFT JOIN BIB_AY.dbo.DOC d1 ON d1.DOC_ID=B.IDX_ID LEFT JOIN BIB_AY.dbo.DOC_ORI d2 ON d2.ORI_ID=B.IDX_ID WHERE d1.DOC_TAM>0 or d2.ORI_TAM>0; """ resultado = copy_all_blobs2files( query_ms=query_ms_all, conf_ms=conf_ms, ruta_salida="/home/user/kk/DOCUMENTS_AYTOS/", )
