Add verapdf integration for speculative PDF/A conversion
Introduce a fast path for PDF/A conversion that uses pikepdf to add PDF/A structures directly (sRGB ICC profile and XMP metadata), then validates with verapdf. If validation passes, skip Ghostscript entirely. If validation fails or verapdf is unavailable, fall back to the existing Ghostscript conversion path. New files: - src/ocrmypdf/_exec/verapdf.py: CLI wrapper for verapdf validator - tests/test_verapdf.py: Test suite for new functionality Modified: - pdfa.py: Add speculative_pdfa_conversion() and helpers - _pipeline.py: Add try_speculative_pdfa() function - _pipelines/_common.py: Integrate speculative path into postprocess()
This commit is contained in:
@@ -0,0 +1,108 @@
|
||||
# SPDX-FileCopyrightText: 2024 James R. Barlow
|
||||
# SPDX-License-Identifier: MPL-2.0
|
||||
|
||||
"""Interface to verapdf executable."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import logging
|
||||
from pathlib import Path
|
||||
from subprocess import PIPE
|
||||
from typing import NamedTuple
|
||||
|
||||
from packaging.version import Version
|
||||
|
||||
from ocrmypdf.exceptions import MissingDependencyError
|
||||
from ocrmypdf.subprocess import get_version, run
|
||||
|
||||
log = logging.getLogger(__name__)
|
||||
|
||||
|
||||
class ValidationResult(NamedTuple):
|
||||
"""Result of PDF/A validation."""
|
||||
|
||||
valid: bool
|
||||
failed_rules: int
|
||||
message: str
|
||||
|
||||
|
||||
def version() -> Version:
|
||||
"""Get verapdf version."""
|
||||
return Version(get_version('verapdf', regex=r'veraPDF (\d+(\.\d+)*)'))
|
||||
|
||||
|
||||
def available() -> bool:
|
||||
"""Check if verapdf is available."""
|
||||
try:
|
||||
version()
|
||||
except MissingDependencyError:
|
||||
return False
|
||||
return True
|
||||
|
||||
|
||||
def output_type_to_flavour(output_type: str) -> str:
|
||||
"""Map OCRmyPDF output_type to verapdf flavour.
|
||||
|
||||
Args:
|
||||
output_type: One of 'pdfa', 'pdfa-1', 'pdfa-2', 'pdfa-3'
|
||||
|
||||
Returns:
|
||||
verapdf flavour string like '1b', '2b', '3b'
|
||||
"""
|
||||
mapping = {
|
||||
'pdfa': '2b',
|
||||
'pdfa-1': '1b',
|
||||
'pdfa-2': '2b',
|
||||
'pdfa-3': '3b',
|
||||
}
|
||||
return mapping.get(output_type, '2b')
|
||||
|
||||
|
||||
def validate(input_file: Path, flavour: str) -> ValidationResult:
|
||||
"""Validate a PDF against a PDF/A profile.
|
||||
|
||||
Args:
|
||||
input_file: Path to PDF file to validate
|
||||
flavour: verapdf flavour (1a, 1b, 2a, 2b, 2u, 3a, 3b, 3u)
|
||||
|
||||
Returns:
|
||||
ValidationResult with validation status
|
||||
"""
|
||||
args = [
|
||||
'verapdf',
|
||||
'--format',
|
||||
'json',
|
||||
'--flavour',
|
||||
flavour,
|
||||
str(input_file),
|
||||
]
|
||||
|
||||
try:
|
||||
proc = run(args, stdout=PIPE, stderr=PIPE, check=False)
|
||||
except FileNotFoundError as e:
|
||||
raise MissingDependencyError('verapdf') from e
|
||||
|
||||
try:
|
||||
result = json.loads(proc.stdout)
|
||||
jobs = result.get('report', {}).get('jobs', [])
|
||||
if not jobs:
|
||||
return ValidationResult(False, -1, 'No validation jobs in result')
|
||||
validation_results = jobs[0].get('validationResult', [])
|
||||
if not validation_results:
|
||||
return ValidationResult(False, -1, 'No validation result in output')
|
||||
validation_result = validation_results[0]
|
||||
details = validation_result.get('details', {})
|
||||
failed_rules = details.get('failedRules', 0)
|
||||
|
||||
if failed_rules == 0:
|
||||
return ValidationResult(True, 0, 'PDF/A validation passed')
|
||||
else:
|
||||
return ValidationResult(
|
||||
False,
|
||||
failed_rules,
|
||||
f'PDF/A validation failed with {failed_rules} rule violations',
|
||||
)
|
||||
except (json.JSONDecodeError, KeyError, TypeError) as e:
|
||||
log.debug('Failed to parse verapdf output: %s', e)
|
||||
return ValidationResult(False, -1, f'Failed to parse verapdf output: {e}')
|
||||
@@ -36,7 +36,7 @@ from ocrmypdf.exceptions import (
|
||||
UnsupportedImageFormatError,
|
||||
)
|
||||
from ocrmypdf.helpers import IMG2PDF_KWARGS, Resolution, safe_symlink
|
||||
from ocrmypdf.pdfa import generate_pdfa_ps
|
||||
from ocrmypdf.pdfa import generate_pdfa_ps, speculative_pdfa_conversion
|
||||
from ocrmypdf.pdfinfo import Colorspace, Encoding, FloatRect, PageInfo, PdfInfo
|
||||
from ocrmypdf.pluginspec import OrientationConfidence
|
||||
|
||||
@@ -932,6 +932,52 @@ def convert_to_pdfa(input_pdf: Path, input_ps_stub: Path, context: PdfContext) -
|
||||
return output_file
|
||||
|
||||
|
||||
def try_speculative_pdfa(input_pdf: Path, context: PdfContext) -> Path | None:
|
||||
"""Try speculative PDF/A conversion with verapdf validation.
|
||||
|
||||
This attempts a fast PDF/A conversion by adding PDF/A structures
|
||||
directly with pikepdf, then validating with verapdf. If validation
|
||||
passes, returns the converted file. If it fails or verapdf is not
|
||||
available, returns None to signal that Ghostscript should be used.
|
||||
|
||||
Args:
|
||||
input_pdf: Path to the PDF to convert
|
||||
context: The PDF context
|
||||
|
||||
Returns:
|
||||
Path to valid PDF/A file, or None if speculative conversion failed
|
||||
"""
|
||||
from ocrmypdf._exec import verapdf
|
||||
|
||||
if not verapdf.available():
|
||||
log.debug('verapdf not available, skipping speculative PDF/A conversion')
|
||||
return None
|
||||
|
||||
options = context.options
|
||||
output_file = context.get_path('speculative_pdfa.pdf')
|
||||
|
||||
try:
|
||||
speculative_pdfa_conversion(input_pdf, output_file, options.output_type)
|
||||
|
||||
flavour = verapdf.output_type_to_flavour(options.output_type)
|
||||
result = verapdf.validate(output_file, flavour)
|
||||
|
||||
if result.valid:
|
||||
log.info('Speculative PDF/A conversion succeeded - skipping Ghostscript')
|
||||
return output_file
|
||||
else:
|
||||
log.debug(
|
||||
'Speculative PDF/A validation failed (%d rule violations), '
|
||||
'falling back to Ghostscript',
|
||||
result.failed_rules,
|
||||
)
|
||||
return None
|
||||
|
||||
except Exception as e:
|
||||
log.debug('Speculative PDF/A conversion failed: %s', e)
|
||||
return None
|
||||
|
||||
|
||||
def should_linearize(working_file: Path, context: PdfContext) -> bool:
|
||||
"""Determine whether the PDF should be linearized.
|
||||
|
||||
|
||||
@@ -45,6 +45,7 @@ from ocrmypdf._pipeline import (
|
||||
rasterize_preview,
|
||||
should_linearize,
|
||||
should_visible_page_image_use_jpg,
|
||||
try_speculative_pdfa,
|
||||
)
|
||||
from ocrmypdf._plugin_manager import OcrmypdfPluginManager
|
||||
from ocrmypdf._validation import (
|
||||
@@ -469,8 +470,14 @@ def postprocess(
|
||||
else:
|
||||
pdf_out = pdf_file
|
||||
if context.options.output_type.startswith('pdfa'):
|
||||
ps_stub_out = generate_postscript_stub(context)
|
||||
pdf_out = convert_to_pdfa(pdf_out, ps_stub_out, context)
|
||||
# Try speculative PDF/A conversion first (fast path using pikepdf + verapdf)
|
||||
speculative_result = try_speculative_pdfa(pdf_out, context)
|
||||
if speculative_result is not None:
|
||||
pdf_out = speculative_result
|
||||
else:
|
||||
# Fall back to Ghostscript conversion
|
||||
ps_stub_out = generate_postscript_stub(context)
|
||||
pdf_out = convert_to_pdfa(pdf_out, ps_stub_out, context)
|
||||
|
||||
optimizing = context.plugin_manager.is_optimization_enabled(context=context)
|
||||
save_settings = get_pdf_save_settings(context.options.output_type)
|
||||
|
||||
+117
-1
@@ -1,16 +1,20 @@
|
||||
# SPDX-FileCopyrightText: 2022 James R. Barlow
|
||||
# SPDX-License-Identifier: MPL-2.0
|
||||
|
||||
"""Utilities for PDF/A production and confirmation with Ghostspcript."""
|
||||
"""Utilities for PDF/A production and confirmation with Ghostscript."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import base64
|
||||
import logging
|
||||
from collections.abc import Iterator
|
||||
from importlib.resources import files as package_files
|
||||
from pathlib import Path
|
||||
|
||||
import pikepdf
|
||||
from pikepdf import Array, Dictionary, Name, Pdf, Stream
|
||||
|
||||
log = logging.getLogger(__name__)
|
||||
|
||||
SRGB_ICC_PROFILE_NAME = 'sRGB.icc'
|
||||
|
||||
@@ -131,3 +135,115 @@ def file_claims_pdfa(filename: Path):
|
||||
pdfa_dict['output'] = 'pdfa'
|
||||
pdfa_dict['conformance'] = conformance
|
||||
return pdfa_dict
|
||||
|
||||
|
||||
def _load_srgb_icc_profile() -> bytes:
|
||||
"""Load the sRGB ICC profile from package data."""
|
||||
return (package_files('ocrmypdf.data') / SRGB_ICC_PROFILE_NAME).read_bytes()
|
||||
|
||||
|
||||
def _pdfa_part_conformance(output_type: str) -> tuple[str, str]:
|
||||
"""Extract PDF/A part and conformance from output_type.
|
||||
|
||||
Args:
|
||||
output_type: One of 'pdfa', 'pdfa-1', 'pdfa-2', 'pdfa-3'
|
||||
|
||||
Returns:
|
||||
Tuple of (part, conformance) e.g., ('2', 'B')
|
||||
"""
|
||||
mapping = {
|
||||
'pdfa': ('2', 'B'),
|
||||
'pdfa-1': ('1', 'B'),
|
||||
'pdfa-2': ('2', 'B'),
|
||||
'pdfa-3': ('3', 'B'),
|
||||
}
|
||||
return mapping.get(output_type, ('2', 'B'))
|
||||
|
||||
|
||||
def add_pdfa_metadata(pdf: Pdf, part: str, conformance: str) -> None:
|
||||
"""Add PDF/A XMP metadata declaration to a PDF.
|
||||
|
||||
Args:
|
||||
pdf: An open pikepdf.Pdf object
|
||||
part: PDF/A part number ('1', '2', or '3')
|
||||
conformance: Conformance level ('A', 'B', or 'U')
|
||||
"""
|
||||
with pdf.open_metadata() as meta:
|
||||
meta['pdfaid:part'] = part
|
||||
meta['pdfaid:conformance'] = conformance
|
||||
|
||||
|
||||
def add_srgb_output_intent(pdf: Pdf) -> None:
|
||||
"""Add sRGB ICC profile as OutputIntent to PDF catalog.
|
||||
|
||||
This creates the required PDF/A OutputIntent structure with:
|
||||
- An ICC profile stream containing sRGB profile
|
||||
- An OutputIntent dictionary pointing to that profile
|
||||
- Updates the Catalog's OutputIntents array
|
||||
|
||||
Args:
|
||||
pdf: An open pikepdf.Pdf object
|
||||
"""
|
||||
icc_data = _load_srgb_icc_profile()
|
||||
|
||||
# Create ICC profile stream
|
||||
icc_stream = Stream(pdf, icc_data)
|
||||
icc_stream[Name.N] = 3 # RGB has 3 components
|
||||
|
||||
# Create OutputIntent dictionary
|
||||
output_intent = Dictionary({
|
||||
'/Type': Name.OutputIntent,
|
||||
'/S': Name('/GTS_PDFA1'),
|
||||
'/OutputConditionIdentifier': 'sRGB',
|
||||
'/DestOutputProfile': icc_stream,
|
||||
})
|
||||
|
||||
# Add to catalog's OutputIntents array
|
||||
if Name.OutputIntents not in pdf.Root:
|
||||
pdf.Root[Name.OutputIntents] = Array([])
|
||||
|
||||
# Check if sRGB OutputIntent already exists
|
||||
for intent in pdf.Root.OutputIntents: # type: ignore[attr-defined]
|
||||
if str(intent.get(Name.OutputConditionIdentifier)) == 'sRGB':
|
||||
log.debug('sRGB OutputIntent already exists, skipping')
|
||||
return
|
||||
|
||||
pdf.Root.OutputIntents.append(output_intent)
|
||||
|
||||
|
||||
def speculative_pdfa_conversion(
|
||||
input_file: Path,
|
||||
output_file: Path,
|
||||
output_type: str,
|
||||
) -> Path:
|
||||
"""Attempt to convert a PDF to PDF/A by adding required structures.
|
||||
|
||||
This function creates a copy of the input PDF and adds:
|
||||
1. sRGB ICC profile as OutputIntent
|
||||
2. XMP metadata declaring PDF/A conformance
|
||||
|
||||
This approach works for PDFs that are already mostly PDF/A compliant
|
||||
but lack the formal declarations. It does NOT perform color conversion,
|
||||
font embedding, or other transformations that Ghostscript does.
|
||||
|
||||
Args:
|
||||
input_file: Path to input PDF
|
||||
output_file: Path where output PDF should be written
|
||||
output_type: One of 'pdfa', 'pdfa-1', 'pdfa-2', 'pdfa-3'
|
||||
|
||||
Returns:
|
||||
Path to the output file
|
||||
|
||||
Raises:
|
||||
pikepdf.PdfError: If the PDF cannot be opened or modified
|
||||
"""
|
||||
part, conformance = _pdfa_part_conformance(output_type)
|
||||
|
||||
with Pdf.open(input_file) as pdf:
|
||||
add_srgb_output_intent(pdf)
|
||||
add_pdfa_metadata(pdf, part, conformance)
|
||||
|
||||
pdf.save(output_file)
|
||||
|
||||
log.debug('Speculative PDF/A conversion complete: %s', output_file)
|
||||
return output_file
|
||||
|
||||
Reference in New Issue
Block a user