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.
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/",
)
