cmconsultor/pdf-toolkit
2
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix: compressão real p/ stream DCTDecode + auth c/ chave vazia

Browse Source 
- Substitui xobj._data por page.images[].replace() em compress.py e pdfa.py
- Adiciona compress_with_stats() com feedback honesto na UI
- Trata PDFTOOLKIT_API_KEY vazia como modo dev (auth desabilitado)
- Atualiza engine.py e app.py para exibir redução real ou 'já otimizado'
This commit is contained in:
CM Consultor
2026-08-09 05:01:16 +00:00
parent 4496e9cb55
commit e97adca143
5 changed files with 239 additions and 111 deletions
Download Patch File Download Diff File Expand all files Collapse all files

18
core/auth.py
View File

@@ -2,8 +2,8 @@
Autenticação via API Key para a API REST.
A chave é lida da variável de ambiente PDFTOOLKIT_API_KEY.
Se a variável não estiver definida, a autenticação fica desabilitada
(útil para desenvolvimento local).
Se a variável não estiver definida **ou estiver vazia**, a
autenticação fica desabilitada (útil para desenvolvimento local).
Uso:
export PDFTOOLKIT_API_KEY="sua-chave-secreta-aqui"
@@ -15,8 +15,9 @@ from fastapi.security import APIKeyHeader
API_KEY_ENV = "PDFTOOLKIT_API_KEY"
# Lê a chave do ambiente (None se não configurada)
_API_KEY = os.environ.get(API_KEY_ENV, None)
# Lê a chave do ambiente (None se não configurada ou vazia)
_API_KEY_CANDIDATE = os.environ.get(API_KEY_ENV, None)
_API_KEY = _API_KEY_CANDIDATE if _API_KEY_CANDIDATE else None
# Schema de header: X-API-Key
api_key_header = APIKeyHeader(name="X-API-Key", auto_error=False)
@@ -26,11 +27,12 @@ def verify_api_key(api_key: str = Security(api_key_header)) -> str:
"""
Dependência FastAPI que valida a API Key.
- Se PDFTOOLKIT_API_KEY não foi definida no ambiente → permite acesso livre
(modo desenvolvimento).
- Se foi definida → exige que o header X-API-Key corresponda exatamente.
- Se PDFTOOLKIT_API_KEY não foi definida ou está vazia → permite
acesso livre (modo desenvolvimento).
- Se foi definida → exige que o header X-API-Key corresponda
exatamente.
Retorna a própria chave se válida, ou levanta 401.
Retorna a própria chave se válida, ou levanta 401/403.
"""
# Modo desenvolvimento: sem chave configurada
if _API_KEY is None:

254
core/compress.py
View File

@@ -1,20 +1,64 @@
"""
Estratégias de compressão e otimização de PDF.
Usa pypdf + Pillow para re-amostragem de imagens e remoção de
objetos redundantes. Mantém a estrutura original do PDF (inclusive
PDF/A) quando possível.
Usa pypdf (API oficial de ImageFile.replace) + Pillow para
re-amostragem de imagens. A abordagem antiga de gravar em
`xobj._data` quebrava o /Filter do stream e falhava em silêncio
na maioria dos PDFs (FlateDecode / pixels crus).
"""
from __future__ import annotations
import io
from typing import Optional
import logging
from typing import Optional, Tuple
from PIL import Image
from pypdf import PdfReader, PdfWriter
from pypdf.constants import ImageAttributes as IA
from config import QUALITY_PRESETS
logger = logging.getLogger(__name__)
def _prepare_image(
img: Image.Image,
quality: int = 60,
max_dim: int = 2000,
) -> Tuple[Image.Image, dict]:
"""
Normaliza modo de cor e redimensiona para o preset.
Retorna (imagem_pronta, kwargs para Image.save / replace).
"""
# JPEG (via replace→PDF) não aceita alpha/CMYK/paleta de forma confiável
if img.mode in ("RGBA", "LA"):
background = Image.new("RGB", img.size, (255, 255, 255))
alpha = img.split()[-1]
background.paste(img.convert("RGBA"), mask=alpha)
img = background
elif img.mode == "P":
img = img.convert("RGBA")
if "A" in img.getbands():
background = Image.new("RGB", img.size, (255, 255, 255))
background.paste(img, mask=img.split()[-1])
img = background
else:
img = img.convert("RGB")
elif img.mode == "CMYK":
img = img.convert("RGB")
elif img.mode not in ("RGB", "L"):
img = img.convert("RGB")
w, h = img.size
if max(w, h) > max_dim:
ratio = max_dim / max(w, h)
new_size = (max(1, int(w * ratio)), max(1, int(h * ratio)))
img = img.resize(new_size, Image.LANCZOS)
save_kwargs = {"quality": quality, "optimize": True}
return img, save_kwargs
def _downsample_image(
img_bytes: bytes,
@@ -23,34 +67,23 @@ def _downsample_image(
fmt: str = "JPEG",
) -> bytes:
"""
Redimensiona e re-comprime uma imagem.
- quality: 1-100 (JPEG)
- max_dim: lado máximo em px
- fmt: "JPEG" ou "PNG"
Redimensiona e re-comprime bytes de imagem (helper legado / pdfa).
Aceita apenas containers que o Pillow abra (JPEG/PNG/...).
Pixels crus de streams FlateDecode NÃO funcionam aqui — use
compress_pdf(), que passa por page.images (PIL Image já montada).
"""
img = Image.open(io.BytesIO(img_bytes))
# Converte RGBA/CMYK/P (palette) para RGB para compressão JPEG
if img.mode not in ("RGB", "L"):
if fmt == "JPEG":
img = img.convert("RGB")
elif img.mode == "P":
img = img.convert("RGB")
# Redimensiona proporcionalmente se exceder max_dim
w, h = img.size
if max(w, h) > max_dim:
ratio = max_dim / max(w, h)
new_size = (int(w * ratio), int(h * ratio))
# Usa ANTIALIAS/LANCZOS para melhor qualidade na redução
img = img.resize(new_size, Image.LANCZOS)
img, save_kwargs = _prepare_image(img, quality=quality, max_dim=max_dim)
buf = io.BytesIO()
if fmt == "JPEG":
img.save(buf, format="JPEG", quality=quality, optimize=True)
out_fmt = "JPEG" if fmt.upper() == "JPEG" else "PNG"
if out_fmt == "JPEG" and img.mode not in ("RGB", "L"):
img = img.convert("RGB")
if out_fmt == "JPEG":
img.save(buf, format="JPEG", **save_kwargs)
else:
img.save(buf, format="PNG", optimize=True)
return buf.getvalue()
@@ -60,75 +93,148 @@ def compress_pdf(
remove_unused: bool = True,
) -> bytes:
"""
Comprime um PDF re-amostrando imagens e otimizando estrutura.
Comprime um PDF re-amostrando imagens embutidas.
Args:
data: Conteúdo do PDF original.
preset: Perfil de qualidade ("screen", "ebook", "printer", "prepress").
remove_unused: Remove objetos não referenciados.
remove_unused: Tenta deduplicar objetos idênticos no writer.
Retorna bytes do PDF comprimido.
Retorna bytes do PDF (pode ser maior em PDFs só-texto; o engine
decide se devolve o original nesse caso).
"""
opts = QUALITY_PRESETS.get(preset, QUALITY_PRESETS["ebook"])
quality = opts["quality"]
max_dim = opts["max_dim"]
reader = PdfReader(io.BytesIO(data))
writer = PdfWriter()
# clone_from é obrigatório para ImageFile.replace() funcionar
writer = PdfWriter(clone_from=reader)
# Copia metadados
writer.metadata = reader.metadata
replaced = 0
errors = 0
# Registra fontes e objetos que usamos (para limpeza)
used_objects = set()
for page in writer.pages:
try:
images = list(page.images)
except Exception as exc:
logger.debug("Falha ao listar imagens da página: %s", exc)
continue
for page_num, page in enumerate(reader.pages):
used_objects.add(id(page))
for img_file in images:
# Imagens inline não têm indirect_reference — replace() recusa
if getattr(img_file, "indirect_reference", None) is None:
continue
try:
pil = img_file.image
if pil is None:
continue
prepared, save_kwargs = _prepare_image(
pil, quality=quality, max_dim=max_dim
)
img_file.replace(prepared, **save_kwargs)
replaced += 1
except Exception as exc:
errors += 1
logger.debug(
"Falha ao recomprimir imagem %s: %s",
getattr(img_file, "name", "?"),
exc,
)
# Verifica se há imagens na página
if "/XObject" in page.get("/Resources", {}):
resources = page["/Resources"]
if "/XObject" in resources:
xobjects = resources["/XObject"]
for obj_name in list(xobjects.keys()):
xobj = xobjects[obj_name]
subtype = xobj.get("/Subtype", "")
if subtype == "/Image":
try:
raw = xobj.get_data()
if raw:
# Determina formato — PNG se tiver transparência
fmt = "PNG" if xobj.get("/ColorSpace") == "/DeviceCMYK" else "JPEG"
compressed = _downsample_image(raw, quality, max_dim, fmt)
xobj._data = compressed
xobj[IA.WIDTH], xobj[IA.HEIGHT] = Image.open(
io.BytesIO(compressed)
).size
except Exception:
# Se falhar na recompressão, mantém original
pass
if remove_unused:
try:
writer.compress_identical_objects()
except Exception:
pass
# Adiciona página modificada
writer.add_page(page)
# Escreve com compressão de streams
out_buf = io.BytesIO()
writer.write_stream(out_buf)
writer.write(out_buf)
result = out_buf.getvalue()
return out_buf.getvalue()
logger.info(
"compress_pdf preset=%s images_replaced=%d errors=%d "
"in=%d out=%d",
preset,
replaced,
errors,
len(data),
len(result),
)
return result
def compress_pdf_with_stats(
data: bytes,
preset: str = "ebook",
remove_unused: bool = True,
) -> Tuple[bytes, dict]:
"""
Igual a compress_pdf, mas devolve estatísticas úteis para UI/API.
"""
opts = QUALITY_PRESETS.get(preset, QUALITY_PRESETS["ebook"])
quality = opts["quality"]
max_dim = opts["max_dim"]
reader = PdfReader(io.BytesIO(data))
writer = PdfWriter(clone_from=reader)
replaced = 0
skipped = 0
errors = 0
for page in writer.pages:
try:
images = list(page.images)
except Exception:
continue
for img_file in images:
if getattr(img_file, "indirect_reference", None) is None:
skipped += 1
continue
try:
pil = img_file.image
if pil is None:
skipped += 1
continue
prepared, save_kwargs = _prepare_image(
pil, quality=quality, max_dim=max_dim
)
img_file.replace(prepared, **save_kwargs)
replaced += 1
except Exception:
errors += 1
if remove_unused:
try:
writer.compress_identical_objects()
except Exception:
pass
out_buf = io.BytesIO()
writer.write(out_buf)
result = out_buf.getvalue()
stats = {
"images_replaced": replaced,
"images_skipped": skipped,
"images_errors": errors,
"original_bytes": len(data),
"result_bytes": len(result),
"reduced": len(result) < len(data),
"preset": preset,
}
return result, stats
def linearize_pdf(data: bytes) -> bytes:
"""
Re-escreve o PDF de forma linearizada (otimizada para web).
Usa pypdf para reestruturar.
Re-escreve o PDF (clone limpo). Não é linearização real de web-fast,
mas reestrutura o arquivo via pypdf.
"""
reader = PdfReader(io.BytesIO(data))
writer = PdfWriter()
writer.metadata = reader.metadata
for page in reader.pages:
writer.add_page(page)
writer = PdfWriter(clone_from=reader)
out_buf = io.BytesIO()
writer.write_stream(out_buf)
return out_buf.getvalue()
writer.write(out_buf)
return out_buf.getvalue()

16
core/engine.py
View File

@@ -9,7 +9,7 @@ import io
from typing import List, Tuple
from .validators import validate_pdf, assert_safe_filename
from .compress import compress_pdf
from .compress import compress_pdf, compress_pdf_with_stats
from .pdfa import detect_pdfa, strip_pdfa, compress_pdfa
from .operations import merge_pdfs, split_pdf, extract_pages, rotate_pages
from .operations import extract_text, extract_images, preview_pages
@@ -38,8 +38,18 @@ class PDFEngine:
ou que não possuem imagens adequadas para reamostragem. Nesse caso,
preservamos o original em vez de entregar um arquivo maior.
"""
result = compress_pdf(data, preset=preset)
return result if len(result) < len(data) else data
result, stats = compress_pdf_with_stats(data, preset=preset)
if stats["reduced"]:
return result
return data
@staticmethod
def compress_with_stats(data: bytes, preset: str = "ebook") -> Tuple[bytes, dict]:
"""Comprime e retorna (bytes, estatísticas) para feedback na UI."""
result, stats = compress_pdf_with_stats(data, preset=preset)
if stats["reduced"]:
return result, stats
return data, stats
@staticmethod
def pdfa_to_regular(data: bytes) -> bytes:

46
core/pdfa.py
View File

@@ -139,30 +139,36 @@ def strip_pdfa(data: bytes) -> bytes:
def compress_pdfa(data: bytes, quality: int = 60, max_dim: int = 2000) -> bytes:
"""
Comprime imagens internas mantendo a estrutura PDF/A.
Usa a mesma lógica de compress.compress_pdf mas preserva
metadados XMP e OutputIntents.
Usa a mesma lógica de compress_pdf (ImageFile.replace)
mas preserva metadados XMP e OutputIntents.
"""
from .compress import _downsample_image
from PIL import Image as PILImage
from .compress import _prepare_image
reader = PdfReader(io.BytesIO(data))
writer = PdfWriter()
writer.metadata = reader.metadata
writer = PdfWriter(clone_from=reader)
for page in reader.pages:
if "/XObject" in page.get("/Resources", {}):
xobjects = page["/Resources"]["/XObject"]
for obj_name in list(xobjects.keys()):
xobj = xobjects[obj_name]
if xobj.get("/Subtype") == "/Image":
try:
raw = xobj.get_data()
if raw:
compressed = _downsample_image(raw, quality, max_dim)
xobj._data = compressed
except Exception:
pass
writer.add_page(page)
replaced = 0
for page in writer.pages:
try:
images = list(page.images)
except Exception:
continue
for img_file in images:
if getattr(img_file, "indirect_reference", None) is None:
continue
try:
pil = img_file.image
if pil is None:
continue
prepared, save_kwargs = _prepare_image(
pil, quality=quality, max_dim=max_dim
)
img_file.replace(prepared, **save_kwargs)
replaced += 1
except Exception:
pass
out_buf = io.BytesIO()
writer.write_stream(out_buf)
writer.write(out_buf)
return out_buf.getvalue()