# SPDX-FileCopyrightText: 2022 James R. Barlow # SPDX-License-Identifier: MPL-2.0 """Built-in plugin to implement PDF page rasterization and PDF/A production.""" from __future__ import annotations import logging from enum import StrEnum from pathlib import Path from typing import Annotated from packaging.version import Version from pikepdf import Name, Pdf, Stream from pydantic import BaseModel, Field from ocrmypdf import hookimpl from ocrmypdf._exec import ghostscript from ocrmypdf._options import ProcessingMode from ocrmypdf.exceptions import MissingDependencyError from ocrmypdf.subprocess import check_external_program log = logging.getLogger(__name__) # Currently all blacklisted versions are lower than 9.55, so none need to # be added here. If a future version is blacklisted, add it here. BLACKLISTED_GS_VERSIONS: frozenset[Version] = frozenset() class ColorConversionStrategy(StrEnum): """Ghostscript color conversion strategies.""" CMYK = 'CMYK' GRAY = 'Gray' LEAVE_COLOR_UNCHANGED = 'LeaveColorUnchanged' RGB = 'RGB' USE_DEVICE_INDEPENDENT_COLOR = 'UseDeviceIndependentColor' class PdfaImageCompression(StrEnum): """PDF/A image compression methods.""" AUTO = 'auto' JPEG = 'jpeg' LOSSLESS = 'lossless' def _resolve_auto_compression( compression: PdfaImageCompression, optimize_level: int ) -> PdfaImageCompression: """Resolve 'auto' image compression based on the optimization level. At ``-O0`` (no optimization) ``auto`` maps to ``lossless`` so Ghostscript will not transcode lossless images to JPEG during PDF/A generation. At all other levels ``auto`` defers to Ghostscript's heuristic, which may recompress images lossily. ``-O1`` is a historical exception: although it is otherwise a lossless-only optimization level, coercing ``auto`` to ``lossless`` there can bloat output substantially (Ghostscript's heuristic often picks JPEG for photographic content), so the default is left alone for backwards compatibility. Users who want guaranteed lossless image handling at any level can pass ``--pdfa-image-compression=lossless`` explicitly. Explicit ``jpeg`` and ``lossless`` choices are always respected. """ if compression == PdfaImageCompression.AUTO and optimize_level == 0: return PdfaImageCompression.LOSSLESS return compression class GhostscriptOptions(BaseModel): """Options specific to Ghostscript operations.""" color_conversion_strategy: Annotated[ ColorConversionStrategy, Field(description="Ghostscript color conversion strategy"), ] = ColorConversionStrategy.LEAVE_COLOR_UNCHANGED pdfa_image_compression: Annotated[ PdfaImageCompression, Field(description="PDF/A image compression method") ] = PdfaImageCompression.AUTO jpeg_quality: Annotated[ int | None, Field( ge=0, le=100, description=( "JPEG quality (0-100) for Ghostscript image recompression during " "PDF/A generation; None uses Ghostscript's default." ), ), ] = None jpeg_maxdpi: Annotated[ int | None, Field( ge=1, description=( "Maximum DPI for Ghostscript image downsampling during PDF/A " "generation." ), ), ] = None @classmethod def add_arguments_to_parser(cls, parser, namespace: str = 'ghostscript'): """Add Ghostscript-specific arguments to the argument parser. Args: parser: The argument parser to add arguments to namespace: The namespace prefix for argument names (not used for ghostscript for backward compatibility) """ gs = parser.add_argument_group("Ghostscript", "Advanced control of Ghostscript") gs.add_argument( '--color-conversion-strategy', action='store', type=str, choices=[ccs.value for ccs in ColorConversionStrategy], default=ColorConversionStrategy.LEAVE_COLOR_UNCHANGED.value, help="Set Ghostscript color conversion strategy", ) gs.add_argument( '--pdfa-image-compression', choices=[pc.value for pc in PdfaImageCompression], default=PdfaImageCompression.AUTO.value, help="Specify how to compress images in the output PDF/A. 'auto' lets " "OCRmyPDF decide: at -O0 it uses lossless image compression so " "Ghostscript does not transcode lossless images to JPEG; at -O1 and " "above it defers to Ghostscript's heuristic, which may recompress " "images lossily. 'jpeg' changes all grayscale and color images to " "JPEG compression. 'lossless' uses PNG-style lossless compression " "for non-JPEG images and passes existing JPEGs through unchanged " "(re-encoding them losslessly would only inflate them). Monochrome " "images are always compressed using a " "lossless codec. Compression settings " "are applied to all pages, including those for which OCR was " "skipped. Not supported for --output-type=pdf ; that setting " "preserves the original compression of all images.", ) gs.add_argument( '--ghostscript-jpeg-quality', type=int, metavar='Q', default=None, dest=f'{namespace}_jpeg_quality', help=( "Advanced: Set Ghostscript's -dJPEGQ for images that Ghostscript " "transcodes to JPEG during PDF/A generation. 0 is maximum " "compression; 100 is best quality. If omitted, Ghostscript's " "default is used. This only affects images Ghostscript chooses " "to recompress; for general JPEG quality tuning prefer " "--jpeg-quality, which is applied by the OCRmyPDF optimizer." ), ) gs.add_argument( '--ghostscript-jpeg-maxdpi', type=int, metavar='DPI', default=None, dest=f'{namespace}_jpeg_maxdpi', help=( "Advanced: Force Ghostscript to downsample color, grayscale, " "and monochrome images in PDF/A output to the given maximum DPI. " "Reducing JPEG quality usually gives better results than " "downsampling at the same file size, and can degrade quality " "of high-resolution monochrome masks." ), ) @hookimpl def register_options(): """Register Ghostscript option model.""" return {'ghostscript': GhostscriptOptions} @hookimpl def add_options(parser): # Use the model's CLI generation method GhostscriptOptions.add_arguments_to_parser(parser) @hookimpl def check_options(options): """Check that the options are valid for this plugin.""" # Only require Ghostscript for pdfa* output types (not 'auto' or 'pdf') # 'auto' mode uses best-effort PDF/A without Ghostscript fallback if options.output_type.startswith('pdfa'): check_external_program( program='gs', package='ghostscript', version_checker=ghostscript.version, need_version='9.54', # RHEL 9's version; Ubuntu 22.04 has 9.55 ) gs_version = ghostscript.version() if gs_version in BLACKLISTED_GS_VERSIONS: raise MissingDependencyError( f"Ghostscript {gs_version} contains serious regressions and is not " "supported. Please upgrade to a newer version." ) if Version('10.0.0') <= gs_version < Version('10.02.1') and ( options.mode in (ProcessingMode.skip, ProcessingMode.redo) ): raise MissingDependencyError( f"Ghostscript 10.0.0 through 10.02.0 (your version: {gs_version}) " "contain serious regressions that corrupt PDFs with existing text, " "such as those processed using --skip-text or --redo-ocr " "(or --mode skip/redo). Please upgrade to a newer version, or use " "--output-type pdf to avoid Ghostscript, or use --force-ocr " "(or --mode force) to discard existing text." ) if gs_version >= Version('10.6.0'): log.warning( "Ghostscript %s contains JPEG encoding errors that may corrupt " "images. OCRmyPDF will attempt to mitigate, but versions 10.6.0+ " "are strongly not recommended until this is fixed upstream.", gs_version, ) if options.output_type == 'pdfa': options.output_type = 'pdfa-2' if ( options.ghostscript.color_conversion_strategy not in ghostscript.COLOR_CONVERSION_STRATEGIES ): raise ValueError( f"Invalid color conversion strategy: " f"{options.ghostscript.color_conversion_strategy}" ) if ( options.ghostscript.pdfa_image_compression != 'auto' and options.output_type not in ('auto', 'pdfa', 'pdfa-1', 'pdfa-2', 'pdfa-3') ): log.warning( "--pdfa-image-compression argument only applies when " "--output-type is 'auto' or one of 'pdfa', 'pdfa-1', 'pdfa-2', 'pdfa-3'" ) @hookimpl def rasterize_pdf_page( input_file, output_file, raster_device, raster_dpi, pageno, page_dpi, rotation, filter_vector, stop_on_soft_error, options, use_cropbox, ): """Rasterize a single page of a PDF file using Ghostscript.""" # Check if user explicitly requested a different rasterizer if options is not None and options.rasterizer == 'pypdfium': # Let pypdfium handle it (it will error in check_options if unavailable) return None log.debug("Rasterizing page %d with the Ghostscript rasterizer", pageno) ghostscript.rasterize_pdf( input_file, output_file, raster_device=raster_device, raster_dpi=raster_dpi, pageno=pageno, page_dpi=page_dpi, rotation=rotation, filter_vector=filter_vector, stop_on_error=stop_on_soft_error, use_cropbox=use_cropbox, ) return output_file def _collect_dctdecode_images(pdf: Pdf) -> dict[tuple, list[tuple[Stream, bytes]]]: """Collect all DCTDecode (JPEG) images from a PDF. Returns a dict mapping image signatures to a list of (stream, raw_bytes) tuples. The signature is (Width, Height, Filter, BitsPerComponent, ColorSpace). """ images: dict[tuple, list[tuple[Stream, bytes]]] = {} def get_colorspace_key(obj): """Get a hashable key for the colorspace.""" cs = obj.get(Name.ColorSpace) if cs is None: return None if isinstance(cs, Name): return str(cs) # For array colorspaces like [/ICCBased ...], use the first element try: return str(cs[0]) if len(cs) > 0 else str(cs) except (TypeError, KeyError): return str(cs) def process_xobject_dict(xobjects, depth=0): """Process an XObject dictionary for DCTDecode images.""" if xobjects is None: return if depth > 10: log.warning("Recursion depth exceeded in _collect_dctdecode_images") return for key in xobjects.keys(): obj = xobjects[key] if obj is None: continue # Check if it's an image with DCTDecode if obj.get(Name.Subtype) == Name.Image: filt = obj.get(Name.Filter) if filt == Name.DCTDecode: sig = ( int(obj.get(Name.Width, 0)), int(obj.get(Name.Height, 0)), str(filt), int(obj.get(Name.BitsPerComponent, 0)), get_colorspace_key(obj), ) raw_bytes = obj.read_raw_bytes() if sig not in images: images[sig] = [] images[sig].append((obj, raw_bytes)) # Recurse into Form XObjects elif obj.get(Name.Subtype) == Name.Form: if Name.Resources in obj: res = obj[Name.Resources] if Name.XObject in res: process_xobject_dict(res[Name.XObject], depth=depth + 1) for page in pdf.pages: if Name.Resources not in page: continue resources = page[Name.Resources] if Name.XObject not in resources: continue process_xobject_dict(resources[Name.XObject]) return images def _repair_gs106_jpeg_corruption( input_pdf_path: Path, output_pdf_path: Path, ) -> bool: """Repair JPEG corruption caused by Ghostscript 10.6. Ghostscript 10.6 has a bug that truncates JPEG data by 1-15 bytes. This function detects and repairs such corruption by copying the original JPEG bytes from the input PDF. Returns True if any repairs were made. """ repaired_count = 0 first_error_logged = False with ( Pdf.open(input_pdf_path) as input_pdf, Pdf.open(output_pdf_path, allow_overwriting_input=True) as output_pdf, ): # Collect all DCTDecode images from both PDFs input_images = _collect_dctdecode_images(input_pdf) output_images = _collect_dctdecode_images(output_pdf) # For each output image, try to find a corresponding input image for sig, output_list in output_images.items(): if sig not in input_images: continue input_list = input_images[sig] for output_stream, output_bytes in output_list: # Try to find a matching input image for _input_stream, input_bytes in input_list: input_len = len(input_bytes) output_len = len(output_bytes) # Check if output is 1-15 bytes shorter diff = input_len - output_len if not (1 <= diff <= 15): continue # Check if the bytes are identical up to the truncation point if output_bytes != input_bytes[:output_len]: continue # This is a corrupt image - repair it if not first_error_logged: log.error( "Ghostscript 10.6 JPEG corruption detected. " "Repairing damaged images from original PDF." ) first_error_logged = True log.warning( f"Replacing corrupt JPEG image " f"({sig[0]}x{sig[1]}, {diff} bytes truncated)" ) # Write the original bytes back to the output stream output_stream.write( input_bytes, filter=Name.DCTDecode, ) repaired_count += 1 break # Move to next output image if repaired_count > 0: output_pdf.save(output_pdf_path) log.info( f"Repaired {repaired_count} JPEG image(s) corrupted by Ghostscript" ) return repaired_count > 0 @hookimpl def generate_pdfa( pdf_pages, pdfmark, output_file, context, pdf_version, pdfa_part, progressbar_class, stop_on_soft_error, ): """Generate a PDF/A from the list of PDF pages and PDF/A metadata.""" # Normalize output_type at point of use output_type = context.options.output_type if output_type == 'pdfa': output_type = 'pdfa-2' compression = _resolve_auto_compression( context.options.ghostscript.pdfa_image_compression, context.options.optimize, ) ghostscript.generate_pdfa( pdf_pages=[pdfmark, *pdf_pages], output_file=output_file, compression=compression, color_conversion_strategy=context.options.ghostscript.color_conversion_strategy, jpeg_quality=context.options.ghostscript.jpeg_quality, jpeg_maxdpi=context.options.ghostscript.jpeg_maxdpi, pdf_version=pdf_version, pdfa_part=pdfa_part, progressbar_class=progressbar_class, stop_on_error=stop_on_soft_error, ) # Repair JPEG corruption caused by Ghostscript 10.6.x gs_version = ghostscript.version() if gs_version >= Version('10.6.0') and len(pdf_pages) == 1: input_pdf = Path(pdf_pages[0]) _repair_gs106_jpeg_corruption(input_pdf, Path(output_file)) return output_file