115 lines
3.4 KiB
Python
115 lines
3.4 KiB
Python
import base64
|
|
import zlib
|
|
|
|
|
|
class DataCompressor:
|
|
"""
|
|
Classe utilitária para compressão e descompressão de dados.
|
|
|
|
Funções:
|
|
- compress(): compacta texto/bytes com zlib e opcional Base64
|
|
- decompress(): decodifica Base64, descompacta e retorna texto legível
|
|
"""
|
|
|
|
@staticmethod
|
|
def decompress(data):
|
|
"""
|
|
Descomprime dados vindos de Firebird/Delphi/strings compactadas.
|
|
|
|
Processos executados automaticamente:
|
|
1. Se vier Base64 → faz base64.b64decode()
|
|
2. Se vier stream/BLOB → lê com .read()
|
|
3. Se for string → converte para bytes
|
|
4. Detecta se é zlib → descompacta
|
|
5. Retorna texto ISO-8859-1
|
|
|
|
Retorna:
|
|
Texto descompactado (str)
|
|
"""
|
|
|
|
# Caso seja None ou vazio
|
|
if not data:
|
|
return ""
|
|
|
|
# 1. Se vier como stream (ex.: campo BLOB do Firebird)
|
|
if hasattr(data, "read"):
|
|
data = data.read()
|
|
|
|
# 2. Se for string, pode ser Base64 → tentamos primeiro decodificar base64
|
|
if isinstance(data, str):
|
|
try:
|
|
# Tentamos decodificar Base64
|
|
data_bytes = base64.b64decode(data, validate=False)
|
|
except Exception:
|
|
# Não era Base64, convertemos string para bytes ISO-8859-1
|
|
data_bytes = data.encode("latin1", errors="ignore")
|
|
else:
|
|
# Já são bytes
|
|
data_bytes = data
|
|
|
|
# Tentativa extra: bytes podem ter vindo em Base64 também
|
|
try:
|
|
data_bytes = base64.b64decode(data_bytes, validate=False)
|
|
except Exception:
|
|
pass
|
|
|
|
# 3. Detectar assinatura zlib (0x78 0x9C ou 0x78 0xDA)
|
|
is_zlib = (
|
|
len(data_bytes) > 2
|
|
and data_bytes[0] == 0x78
|
|
and data_bytes[1] in (0x9C, 0xDA)
|
|
)
|
|
|
|
if is_zlib:
|
|
try:
|
|
text = zlib.decompress(data_bytes).decode("iso-8859-1", errors="ignore")
|
|
return text
|
|
except Exception:
|
|
pass
|
|
|
|
# 4. Se não era zlib, tentamos decodificar direto
|
|
try:
|
|
return data_bytes.decode("iso-8859-1", errors="ignore")
|
|
except Exception:
|
|
return str(data)
|
|
|
|
# =====================================================================
|
|
|
|
@staticmethod
|
|
def compress(text, *, encoding="iso-8859-1", as_base64=True):
|
|
"""
|
|
Compacta texto/bytes usando zlib.
|
|
|
|
Parâmetros:
|
|
text -> str | bytes | stream
|
|
encoding -> usado ao converter str para bytes
|
|
as_base64 -> retorna Base64 (True) ou bytes puros (False)
|
|
|
|
Retorno:
|
|
bytes (quando as_base64=False)
|
|
str Base64 (quando as_base64=True)
|
|
"""
|
|
|
|
if text is None or text == "":
|
|
return "" if as_base64 else b""
|
|
|
|
# Se for stream/BLOB
|
|
if hasattr(text, "read"):
|
|
raw = text.read()
|
|
else:
|
|
raw = text
|
|
|
|
# Garante bytes
|
|
if isinstance(raw, str):
|
|
raw_bytes = raw.encode(encoding, errors="ignore")
|
|
else:
|
|
raw_bytes = bytes(raw)
|
|
|
|
# Compressão zlib
|
|
comp = zlib.compress(raw_bytes)
|
|
|
|
# Retorna Base64 se solicitado
|
|
if as_base64:
|
|
return base64.b64encode(comp).decode("ascii")
|
|
|
|
return comp
|