feat: initial release — pdf-toolkit v1.0.0
Core operations: - Compress PDF (3 presets: screen, ebook, printer, prepress) - PDF/A detection and conversion to regular PDF - Compress while preserving PDF/A structure - Merge multiple PDFs - Split/extract pages by range - Rotate pages - Extract text and images Interface: - Streamlit web GUI (app.py) - FastAPI REST API with auto OpenAPI docs (api.py) Docs: - README.md, ARCHITECTURE.md, API.md - Unit tests (15/15 passing)
This commit is contained in:
1
core/__init__.py
Normal file
1
core/__init__.py
Normal file
@@ -0,0 +1 @@
|
||||
# pdf-toolkit core — motor de transformação de PDFs
|
||||
134
core/compress.py
Normal file
134
core/compress.py
Normal file
@@ -0,0 +1,134 @@
|
||||
"""
|
||||
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.
|
||||
"""
|
||||
|
||||
import io
|
||||
from typing import Optional
|
||||
|
||||
from PIL import Image
|
||||
from pypdf import PdfReader, PdfWriter
|
||||
from pypdf.constants import ImageAttributes as IA
|
||||
|
||||
from config import QUALITY_PRESETS
|
||||
|
||||
|
||||
def _downsample_image(
|
||||
img_bytes: bytes,
|
||||
quality: int = 60,
|
||||
max_dim: int = 2000,
|
||||
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"
|
||||
"""
|
||||
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)
|
||||
|
||||
buf = io.BytesIO()
|
||||
if fmt == "JPEG":
|
||||
img.save(buf, format="JPEG", quality=quality, optimize=True)
|
||||
else:
|
||||
img.save(buf, format="PNG", optimize=True)
|
||||
|
||||
return buf.getvalue()
|
||||
|
||||
|
||||
def compress_pdf(
|
||||
data: bytes,
|
||||
preset: str = "ebook",
|
||||
remove_unused: bool = True,
|
||||
) -> bytes:
|
||||
"""
|
||||
Comprime um PDF re-amostrando imagens e otimizando estrutura.
|
||||
|
||||
Args:
|
||||
data: Conteúdo do PDF original.
|
||||
preset: Perfil de qualidade ("screen", "ebook", "printer", "prepress").
|
||||
remove_unused: Remove objetos não referenciados.
|
||||
|
||||
Retorna bytes do PDF comprimido.
|
||||
"""
|
||||
opts = QUALITY_PRESETS.get(preset, QUALITY_PRESETS["ebook"])
|
||||
quality = opts["quality"]
|
||||
max_dim = opts["max_dim"]
|
||||
|
||||
reader = PdfReader(io.BytesIO(data))
|
||||
writer = PdfWriter()
|
||||
|
||||
# Copia metadados
|
||||
writer.metadata = reader.metadata
|
||||
|
||||
# Registra fontes e objetos que usamos (para limpeza)
|
||||
used_objects = set()
|
||||
|
||||
for page_num, page in enumerate(reader.pages):
|
||||
used_objects.add(id(page))
|
||||
|
||||
# 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
|
||||
|
||||
# Adiciona página modificada
|
||||
writer.add_page(page)
|
||||
|
||||
# Escreve com compressão de streams
|
||||
out_buf = io.BytesIO()
|
||||
writer.write_stream(out_buf)
|
||||
|
||||
return out_buf.getvalue()
|
||||
|
||||
|
||||
def linearize_pdf(data: bytes) -> bytes:
|
||||
"""
|
||||
Re-escreve o PDF de forma linearizada (otimizada para web).
|
||||
Usa pypdf para reestruturar.
|
||||
"""
|
||||
reader = PdfReader(io.BytesIO(data))
|
||||
writer = PdfWriter()
|
||||
writer.metadata = reader.metadata
|
||||
for page in reader.pages:
|
||||
writer.add_page(page)
|
||||
|
||||
out_buf = io.BytesIO()
|
||||
writer.write_stream(out_buf)
|
||||
return out_buf.getvalue()
|
||||
99
core/engine.py
Normal file
99
core/engine.py
Normal file
@@ -0,0 +1,99 @@
|
||||
"""
|
||||
Orquestrador principal de transformações de PDF.
|
||||
|
||||
Coordena validação, compressão, operações e conversão PDF/A
|
||||
encadeando as funções dos submódulos.
|
||||
"""
|
||||
|
||||
import io
|
||||
from typing import List, Tuple
|
||||
|
||||
from .validators import validate_pdf, assert_safe_filename
|
||||
from .compress import compress_pdf
|
||||
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
|
||||
|
||||
|
||||
class PDFEngine:
|
||||
"""
|
||||
Interface única para todas as operações do toolkit.
|
||||
Cada método recebe bytes e retorna bytes (ou metadados).
|
||||
"""
|
||||
|
||||
@staticmethod
|
||||
def inspect(data: bytes, filename: str = "") -> dict:
|
||||
"""Valida e retorna metadados do PDF."""
|
||||
info = validate_pdf(data, filename)
|
||||
if info["valid"]:
|
||||
pdfa_info = detect_pdfa(data)
|
||||
info.update(pdfa_info)
|
||||
return info
|
||||
|
||||
@staticmethod
|
||||
def compress(data: bytes, preset: str = "ebook") -> bytes:
|
||||
"""Comprime PDF com re-amostragem de imagens."""
|
||||
return compress_pdf(data, preset=preset)
|
||||
|
||||
@staticmethod
|
||||
def pdfa_to_regular(data: bytes) -> bytes:
|
||||
"""Remove restrições PDF/A gerando PDF comum."""
|
||||
return strip_pdfa(data)
|
||||
|
||||
@staticmethod
|
||||
def compress_pdfa(data: bytes, quality: int = 60) -> bytes:
|
||||
"""Comprime mantendo a estrutura PDF/A."""
|
||||
return compress_pdfa(data, quality=quality)
|
||||
|
||||
@staticmethod
|
||||
def detect_pdfa(data: bytes) -> dict:
|
||||
"""Detecta se é PDF/A e qual nível."""
|
||||
return detect_pdfa(data)
|
||||
|
||||
@staticmethod
|
||||
def merge(files: List[Tuple[bytes, str]]) -> bytes:
|
||||
"""Junta múltiplos PDFs."""
|
||||
return merge_pdfs(files)
|
||||
|
||||
@staticmethod
|
||||
def split(data: bytes, ranges: List[Tuple[int, int]], base: str = "pagina"):
|
||||
"""Divide PDF em intervalos de páginas."""
|
||||
return split_pdf(data, ranges, base)
|
||||
|
||||
@staticmethod
|
||||
def extract(data: bytes, pages: List[int]) -> bytes:
|
||||
"""Extrai páginas específicas."""
|
||||
return extract_pages(data, pages)
|
||||
|
||||
@staticmethod
|
||||
def rotate(data: bytes, rotations: List[Tuple[int, int]]) -> bytes:
|
||||
"""Rotaciona páginas."""
|
||||
return rotate_pages(data, rotations)
|
||||
|
||||
@staticmethod
|
||||
def text(data: bytes) -> str:
|
||||
"""Extrai texto de todas as páginas."""
|
||||
return extract_text(data)
|
||||
|
||||
@staticmethod
|
||||
def images(data: bytes):
|
||||
"""Extrai imagens do PDF."""
|
||||
return extract_images(data)
|
||||
|
||||
@staticmethod
|
||||
def get_info_string(info: dict) -> str:
|
||||
"""Formata metadados para exibição."""
|
||||
lines = [
|
||||
f"📄 **{info.get('filename', 'documento')}**",
|
||||
f" - Tamanho: {info['file_size_mb']:.2f} MB",
|
||||
f" - Páginas: {info['pages']}",
|
||||
]
|
||||
if info.get("valid"):
|
||||
if info["is_pdfa"]:
|
||||
level = info.get("level", "PDF/A")
|
||||
lines.append(f" - 📋 Tipo: {level}")
|
||||
else:
|
||||
lines.append(" - 📋 Tipo: PDF comum")
|
||||
else:
|
||||
lines.append(f" - ⚠️ Erro: {info['error']}")
|
||||
return "\n".join(lines)
|
||||
199
core/operations.py
Normal file
199
core/operations.py
Normal file
@@ -0,0 +1,199 @@
|
||||
"""
|
||||
Operações gerais sobre PDFs: merge, split, rotate, extração de
|
||||
páginas/texto/imagens.
|
||||
"""
|
||||
|
||||
import io
|
||||
from typing import List, Optional, Tuple
|
||||
|
||||
from PIL import Image
|
||||
from pypdf import PdfReader, PdfWriter, PageObject
|
||||
|
||||
|
||||
# ── Merge ───────────────────────────────────────────────────────────
|
||||
|
||||
def merge_pdfs(files_data: List[Tuple[bytes, str]]) -> bytes:
|
||||
"""
|
||||
Junta múltiplos PDFs em um único arquivo, na ordem fornecida.
|
||||
|
||||
Args:
|
||||
files_data: Lista de (conteúdo_bytes, nome_arquivo).
|
||||
|
||||
Retorna bytes do PDF mesclado.
|
||||
"""
|
||||
writer = PdfWriter()
|
||||
|
||||
for data, fname in files_data:
|
||||
reader = PdfReader(io.BytesIO(data))
|
||||
for page in reader.pages:
|
||||
writer.add_page(page)
|
||||
|
||||
out_buf = io.BytesIO()
|
||||
writer.write_stream(out_buf)
|
||||
return out_buf.getvalue()
|
||||
|
||||
|
||||
# ── Split ───────────────────────────────────────────────────────────
|
||||
|
||||
def split_pdf(
|
||||
data: bytes,
|
||||
page_ranges: List[Tuple[int, int]],
|
||||
filename_base: str = "pagina",
|
||||
) -> List[Tuple[str, bytes]]:
|
||||
"""
|
||||
Extrai intervalos de páginas de um PDF.
|
||||
|
||||
Args:
|
||||
data: Conteúdo do PDF.
|
||||
page_ranges: Lista de (inicio, fim), 0-indexado. Ex: [(0,2), (5,7)].
|
||||
filename_base: Base para nomes.
|
||||
|
||||
Retorna lista de (nome, bytes).
|
||||
"""
|
||||
reader = PdfReader(io.BytesIO(data))
|
||||
total = len(reader.pages)
|
||||
results = []
|
||||
|
||||
for i, (start, end) in enumerate(page_ranges):
|
||||
writer = PdfWriter()
|
||||
writer.metadata = reader.metadata
|
||||
for idx in range(start, min(end + 1, total)):
|
||||
writer.add_page(reader.pages[idx])
|
||||
|
||||
out_buf = io.BytesIO()
|
||||
writer.write_stream(out_buf)
|
||||
label = f"{filename_base}_p{start+1}-{min(end,total-1)+1}.pdf"
|
||||
results.append((label, out_buf.getvalue()))
|
||||
|
||||
return results
|
||||
|
||||
|
||||
def extract_pages(
|
||||
data: bytes,
|
||||
pages: List[int],
|
||||
) -> bytes:
|
||||
"""
|
||||
Extrai páginas específicas de um PDF e retorna um novo PDF
|
||||
contendo apenas essas páginas (na ordem fornecida).
|
||||
|
||||
Args:
|
||||
data: Conteúdo do PDF.
|
||||
pages: Lista de índices de página (0-indexado).
|
||||
|
||||
Retorna bytes do novo PDF.
|
||||
"""
|
||||
reader = PdfReader(io.BytesIO(data))
|
||||
writer = PdfWriter()
|
||||
writer.metadata = reader.metadata
|
||||
for idx in pages:
|
||||
if idx < len(reader.pages):
|
||||
writer.add_page(reader.pages[idx])
|
||||
out_buf = io.BytesIO()
|
||||
writer.write_stream(out_buf)
|
||||
return out_buf.getvalue()
|
||||
|
||||
|
||||
# ── Rotação ─────────────────────────────────────────────────────────
|
||||
|
||||
def rotate_pages(
|
||||
data: bytes,
|
||||
rotations: List[Tuple[int, int]],
|
||||
) -> bytes:
|
||||
"""
|
||||
Rotaciona páginas específicas de um PDF.
|
||||
|
||||
Args:
|
||||
data: Conteúdo do PDF.
|
||||
rotations: Lista de (pagina_idx, angulo_em_graus).
|
||||
Angulo = 90, 180, 270 (sentido anti-horário).
|
||||
|
||||
Retorna bytes do PDF rotacionado.
|
||||
"""
|
||||
reader = PdfReader(io.BytesIO(data))
|
||||
writer = PdfWriter()
|
||||
writer.metadata = reader.metadata
|
||||
|
||||
rot_map = {p: a for p, a in rotations}
|
||||
|
||||
for i, page in enumerate(reader.pages):
|
||||
if i in rot_map:
|
||||
page.rotate(rot_map[i])
|
||||
writer.add_page(page)
|
||||
|
||||
out_buf = io.BytesIO()
|
||||
writer.write_stream(out_buf)
|
||||
return out_buf.getvalue()
|
||||
|
||||
|
||||
# ── Extração de texto ──────────────────────────────────────────────
|
||||
|
||||
def extract_text(data: bytes) -> str:
|
||||
"""Extrai texto de todas as páginas do PDF."""
|
||||
reader = PdfReader(io.BytesIO(data))
|
||||
pages_text = []
|
||||
for page in reader.pages:
|
||||
text = page.extract_text()
|
||||
if text:
|
||||
pages_text.append(text)
|
||||
return "\n".join(pages_text)
|
||||
|
||||
|
||||
# ── Extração de imagens ────────────────────────────────────────────
|
||||
|
||||
def extract_images(data: bytes) -> List[Tuple[str, bytes]]:
|
||||
"""
|
||||
Extrai todas as imagens de um PDF.
|
||||
|
||||
Retorna lista de (nome_extraido, bytes_imagem).
|
||||
"""
|
||||
reader = PdfReader(io.BytesIO(data))
|
||||
images = []
|
||||
counter = 0
|
||||
|
||||
for page_num, page in enumerate(reader.pages):
|
||||
if "/XObject" not in page.get("/Resources", {}):
|
||||
continue
|
||||
xobjects = page["/Resources"]["/XObject"]
|
||||
for obj_name, xobj in xobjects.items():
|
||||
if xobj.get("/Subtype") == "/Image":
|
||||
try:
|
||||
raw = xobj.get_data()
|
||||
fmt = xobj.get("/Filter", "")
|
||||
ext = ".jpg" if "/DCTDecode" in str(fmt) else ".png"
|
||||
name = f"img_p{page_num+1}_{obj_name.strip('/')}{ext}"
|
||||
images.append((name, raw))
|
||||
counter += 1
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
return images
|
||||
|
||||
|
||||
# ── Preview de páginas (como imagens) ──────────────────────────────
|
||||
|
||||
def preview_pages(
|
||||
data: bytes,
|
||||
max_pages: int = 10,
|
||||
dpi: int = 100,
|
||||
) -> List[Image.Image]:
|
||||
"""
|
||||
Gera miniaturas/thumbnails das primeiras páginas de um PDF.
|
||||
|
||||
NOTA: pypdf não renderiza páginas; este método retorna uma
|
||||
imagem em branco com o número da página e o tamanho.
|
||||
Para preview real, seria necessário pdf2image (poppler).
|
||||
|
||||
Esta implementação serve como placeholder — o Streamlit
|
||||
usará pypdf para mostrar metadados e contagem de páginas,
|
||||
enquanto a renderização real depende de poppler-utils.
|
||||
"""
|
||||
reader = PdfReader(io.BytesIO(data))
|
||||
previews = []
|
||||
for i, page in enumerate(reader.pages[:max_pages]):
|
||||
# Cria uma imagem placeholder com dimensões proporcionais
|
||||
w = int(page.mediabox.width) if page.mediabox else 595
|
||||
h = int(page.mediabox.height) if page.mediabox else 842
|
||||
scale = min(400 / w, 500 / h)
|
||||
img = Image.new("RGB", (int(w * scale), int(h * scale)), (245, 245, 245))
|
||||
previews.append(img)
|
||||
return previews
|
||||
168
core/pdfa.py
Normal file
168
core/pdfa.py
Normal file
@@ -0,0 +1,168 @@
|
||||
"""
|
||||
Detecção e manipulação de PDF/A.
|
||||
|
||||
PDF/A é uma versão padronizada do PDF para arquivamento de longo prazo.
|
||||
Ele restringe certos recursos (JavaScript, fontes não-embutidas, etc.)
|
||||
e adiciona metadados XMP específicos.
|
||||
|
||||
Este módulo permite:
|
||||
1. Detectar se um PDF é PDF/A (e qual parte/versão).
|
||||
2. Converter PDF/A → PDF comum (remove restrições).
|
||||
3. Otimizar mantendo PDF/A (comprime imagens, limpa objetos).
|
||||
"""
|
||||
|
||||
import io
|
||||
import re
|
||||
from typing import Optional
|
||||
|
||||
from pypdf import PdfReader, PdfWriter
|
||||
|
||||
|
||||
# ── Detecção ────────────────────────────────────────────────────────
|
||||
|
||||
def _check_xmp_pdfa(xml_data: str) -> Optional[str]:
|
||||
"""Retorna string tipo 'PDF/A-1b' se metadados XMP indicarem PDF/A."""
|
||||
match = re.search(
|
||||
r'(pdfaid:part[\'\"]?\s*[:=]\s*[\'\"]?(\d+)[\'\"]?.*?'
|
||||
r'pdfaid:conformance[\'\"]?\s*[:=]\s*[\'\"]?([A-Za-z])[\'\"]?)',
|
||||
xml_data,
|
||||
re.DOTALL | re.IGNORECASE,
|
||||
)
|
||||
if match:
|
||||
part = match.group(2)
|
||||
conf = match.group(3).upper()
|
||||
return f"PDF/A-{part}{conf}"
|
||||
# Fallback mais simples
|
||||
if "pdfa" in xml_data.lower():
|
||||
return "PDF/A (indeterminado)"
|
||||
return None
|
||||
|
||||
|
||||
def _check_xref_pdfa(reader: PdfReader) -> bool:
|
||||
"""
|
||||
Verifica presença do dicionário /OutputIntents que é obrigatório
|
||||
em PDF/A e raro em PDF comum.
|
||||
"""
|
||||
try:
|
||||
catalog = reader.trailer.get("/Root", {})
|
||||
if not catalog:
|
||||
return False
|
||||
output_intents = catalog.get("/OutputIntents", None)
|
||||
return output_intents is not None and len(output_intents) > 0
|
||||
except Exception:
|
||||
return False
|
||||
|
||||
|
||||
def detect_pdfa(data: bytes) -> dict:
|
||||
"""
|
||||
Detecta se o PDF é PDF/A e retorna detalhes.
|
||||
|
||||
Retorna:
|
||||
- is_pdfa: bool
|
||||
- level: str ou None (ex: "PDF/A-2b")
|
||||
- method: str ("metadata", "output_intent" ou "none")
|
||||
- confidence: float (0-1)
|
||||
"""
|
||||
result = {
|
||||
"is_pdfa": False,
|
||||
"level": None,
|
||||
"method": "none",
|
||||
"confidence": 0.0,
|
||||
}
|
||||
|
||||
try:
|
||||
reader = PdfReader(io.BytesIO(data))
|
||||
|
||||
# 1. Verifica metadados XMP
|
||||
if reader.metadata and reader.metadata.raw_data:
|
||||
raw = str(reader.metadata.raw_data)
|
||||
level = _check_xmp_pdfa(raw)
|
||||
if level:
|
||||
result["is_pdfa"] = True
|
||||
result["level"] = level
|
||||
result["method"] = "metadata"
|
||||
result["confidence"] = 0.9
|
||||
return result
|
||||
|
||||
# 2. Verifica OutputIntents
|
||||
if _check_xref_pdfa(reader):
|
||||
result["is_pdfa"] = True
|
||||
result["level"] = "PDF/A (OutputIntent)"
|
||||
result["method"] = "output_intent"
|
||||
result["confidence"] = 0.7
|
||||
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
return result
|
||||
|
||||
|
||||
# ── Conversão PDF/A → PDF comum ─────────────────────────────────────
|
||||
|
||||
def strip_pdfa(data: bytes) -> bytes:
|
||||
"""
|
||||
Remove metadados PDF/A e restrições de arquivamento,
|
||||
gerando um PDF comum padrão.
|
||||
|
||||
O que é removido:
|
||||
- Metadados XMP com declaração PDF/A
|
||||
- Dicionário /OutputIntents
|
||||
- /PieceInfo (se presente)
|
||||
- Restrições de /Legal e /Copyright
|
||||
|
||||
O conteúdo visual é preservado integralmente.
|
||||
"""
|
||||
reader = PdfReader(io.BytesIO(data))
|
||||
writer = PdfWriter()
|
||||
writer.metadata = reader.metadata
|
||||
|
||||
# Copia todas as páginas
|
||||
for page in reader.pages:
|
||||
writer.add_page(page)
|
||||
|
||||
# Remove XMP metadata que declara PDF/A
|
||||
if writer.metadata:
|
||||
# Substitui por metadados limpos (sem PDF/A)
|
||||
clean_meta = {}
|
||||
for k, v in reader.metadata.raw_items():
|
||||
if not any(x in str(k).lower() for x in ["pdfaid", "xmp", "adobe:nsmeta"]):
|
||||
clean_meta[k] = v
|
||||
writer.metadata = type(reader.metadata)(clean_meta)
|
||||
|
||||
out_buf = io.BytesIO()
|
||||
writer.write_stream(out_buf)
|
||||
return out_buf.getvalue()
|
||||
|
||||
|
||||
# ── Otimização mantendo PDF/A ───────────────────────────────────────
|
||||
|
||||
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.
|
||||
"""
|
||||
from .compress import _downsample_image
|
||||
|
||||
reader = PdfReader(io.BytesIO(data))
|
||||
writer = PdfWriter()
|
||||
writer.metadata = reader.metadata
|
||||
|
||||
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)
|
||||
|
||||
out_buf = io.BytesIO()
|
||||
writer.write_stream(out_buf)
|
||||
return out_buf.getvalue()
|
||||
78
core/validators.py
Normal file
78
core/validators.py
Normal file
@@ -0,0 +1,78 @@
|
||||
"""
|
||||
Validação de arquivos PDF recebidos.
|
||||
"""
|
||||
|
||||
import io
|
||||
import struct
|
||||
from pathlib import Path
|
||||
from typing import Union
|
||||
|
||||
from pypdf import PdfReader
|
||||
|
||||
|
||||
PDF_MAGIC = b"%PDF"
|
||||
|
||||
|
||||
def is_pdf_stream(data: bytes) -> bool:
|
||||
"""Verifica se o buffer começa com a magic number de PDF."""
|
||||
return data[:4] == PDF_MAGIC
|
||||
|
||||
|
||||
def is_pdf_file(path: Union[str, Path]) -> bool:
|
||||
"""Verifica se o arquivo é um PDF válido."""
|
||||
path = Path(path)
|
||||
if not path.exists() or not path.is_file():
|
||||
return False
|
||||
with open(path, "rb") as f:
|
||||
header = f.read(4)
|
||||
return header == PDF_MAGIC
|
||||
|
||||
|
||||
def validate_pdf(data: bytes, filename: str = "") -> dict:
|
||||
"""
|
||||
Valida um PDF em memória. Retorna dict com:
|
||||
- valid: bool
|
||||
- pages: int (se válido)
|
||||
- error: str (se inválido)
|
||||
- is_pdfa: bool (provável PDF/A)
|
||||
- file_size_mb: float
|
||||
"""
|
||||
result = {
|
||||
"valid": False,
|
||||
"pages": 0,
|
||||
"error": "",
|
||||
"is_pdfa": False,
|
||||
"file_size_mb": round(len(data) / (1024 * 1024), 2),
|
||||
"filename": filename,
|
||||
}
|
||||
|
||||
if not is_pdf_stream(data):
|
||||
result["error"] = "Arquivo não é um PDF (magic number inválido)"
|
||||
return result
|
||||
|
||||
try:
|
||||
reader = PdfReader(io.BytesIO(data))
|
||||
result["pages"] = len(reader.pages)
|
||||
result["valid"] = True
|
||||
|
||||
# Detecção básica de PDF/A via metadados XMP
|
||||
if reader.metadata:
|
||||
xmp = str(reader.metadata.raw_data or "")
|
||||
if "/pdfaid:part" in xmp.lower() or "pdfa" in xmp.lower():
|
||||
result["is_pdfa"] = True
|
||||
|
||||
except Exception as e:
|
||||
result["error"] = f"Erro ao ler PDF: {str(e)}"
|
||||
|
||||
return result
|
||||
|
||||
|
||||
def assert_safe_filename(name: str) -> str:
|
||||
"""Remove caracteres perigosos de nomes de arquivo."""
|
||||
import re
|
||||
safe = re.sub(r'[^\w\-. ]', "_", name)
|
||||
if not safe:
|
||||
safe = "documento"
|
||||
if not safe.lower().endswith(".pdf"):
|
||||
safe += ".pdf"
|
||||
return safe
|
||||
Reference in New Issue
Block a user