Spaces:
Paused
Paused
File size: 3,566 Bytes
f66643d | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 | from __future__ import annotations
import logging
import shutil
import tempfile
from pathlib import Path
import fitz
from .config import CompressConfig
logger = logging.getLogger(__name__)
def finalize_save(
doc: fitz.Document,
output_path: Path,
cfg: CompressConfig,
) -> None:
"""Subset fonts, apply deflate/garbage optimisation, optionally re-encode images."""
if cfg.subset_fonts:
doc.subset_fonts()
logger.debug("subset_fonts done")
save_kwargs: dict = {"garbage": 4, "clean": True}
if cfg.deflate:
save_kwargs.update(
deflate=True, deflate_images=True, deflate_fonts=True, use_objstms=1
)
output_path.parent.mkdir(parents=True, exist_ok=True)
doc.save(str(output_path), **save_kwargs)
logger.debug("Saved %s", output_path)
if cfg.pikepdf_image_recompress:
_recompress_images(output_path, cfg)
def _recompress_images(path: Path, cfg: CompressConfig) -> None:
"""Re-encode images in PDF via pikepdf (optional, aggressive compression)."""
from importlib.util import find_spec
if find_spec("pikepdf") is None or find_spec("PIL") is None:
logger.warning("pikepdf or Pillow not installed; skipping image recompression")
return
import pikepdf
original_size = path.stat().st_size
with tempfile.NamedTemporaryFile(suffix=".pdf", delete=False) as tmp:
tmp_path = Path(tmp.name)
try:
with pikepdf.open(str(path)) as pdf:
for page in pdf.pages:
for xobj_name in list(page.Resources.get("/XObject", {}).keys()):
xobj = page.Resources.XObject[xobj_name]
if xobj.get("/Subtype") != "/Image":
continue
_recompress_one(pdf, xobj, cfg)
pdf.save(str(tmp_path), compress_streams=True, recompress_flate=True)
new_size = tmp_path.stat().st_size
if new_size < original_size:
shutil.move(str(tmp_path), str(path))
logger.info(
"Image recompression: %d → %d bytes (%.1f%%)",
original_size,
new_size,
100 * new_size / max(1, original_size),
)
else:
tmp_path.unlink(missing_ok=True)
logger.debug("Image recompression skipped: not smaller")
except Exception as exc:
logger.warning("Image recompression failed: %s", exc)
tmp_path.unlink(missing_ok=True)
def _recompress_one(pdf, xobj, cfg: CompressConfig) -> None:
import io
import pikepdf
from PIL import Image
try:
# Skip non-standard images
cs = xobj.get("/ColorSpace")
if cs in ("/DeviceCMYK",):
return
bpc = xobj.get("/BitsPerComponent", 8)
if int(bpc) != 8:
return
if "/Mask" in xobj or "/SMask" in xobj:
return
w = int(xobj["/Width"])
h = int(xobj["/Height"])
if w * h == 0:
return
raw = xobj.read_raw_bytes()
img = Image.open(io.BytesIO(raw)).convert("RGB")
# Resize to target DPI if image is very large
# (we don't know display DPI here so skip resize — just recompress)
buf = io.BytesIO()
img.save(buf, format="JPEG", quality=cfg.jpeg_quality, optimize=True)
encoded = buf.getvalue()
if len(encoded) < len(raw):
xobj.write(encoded, filter=pikepdf.Name("/DCTDecode"))
except Exception:
pass # Leave image unchanged on any error
|