menekis/OCRmyPDF
Public Access
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Separate probing from execution in _exec and subprocess modules

Browse Source 
Split ocrmypdf.subprocess/__init__.py into three private submodules by
concern (_run, _version, _check) and reduce __init__ to re-exports.
Introduce ocrmypdf._exec._probe.ToolProbe to centralize the version()/
available() pattern each tool module was reimplementing, so the "is this
tool installed and suitable?" question is cleanly distinct from the
pure, picklable functions that do the work.

Also replace the ghostscript module-import log.addFilter() side effect
with an idempotent _ensure_log_filter_installed() called at the top of
each work function, so the DuplicateFilter is present in subprocess
workers without relying on import-time ordering.

Public API of ocrmypdf.subprocess is unchanged.
This commit is contained in:
James R. Barlow
2026-04-24 13:33:34 -07:00
parent 3336d67e77
commit 84f5fe9ee0
11 changed files with 528 additions and 387 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/ocrmypdf/_exec/_probe.py
+72
View File
@@ -0,0 +1,72 @@
# SPDX-FileCopyrightText: 2026 James R. Barlow
# SPDX-License-Identifier: MPL-2.0
"""Probe helper for external executables.
Each ``ocrmypdf._exec.<tool>`` module describes its external program with a
module-level :class:`ToolProbe` and delegates ``version()`` / ``available()``
to it. This separates the "is the tool installed and suitable?" question
(probing) from the "run the tool" question (execution). Work functions stay
as pure module-level functions so they are trivially picklable for use in
subprocess workers.
"""
from __future__ import annotations
from collections.abc import Mapping
from dataclasses import dataclass
from packaging.version import Version
from ocrmypdf.exceptions import MissingDependencyError
from ocrmypdf.subprocess import get_version
@dataclass(frozen=True)
class ToolProbe:
"""Describes how to detect an external executable and its version.
Attributes:
program: The program name as it appears on PATH (or a full path).
version_arg: The argument that elicits a version string.
version_regex: A regex with a capturing group that extracts the
version from the program's output.
version_cls: A :class:`packaging.version.Version` subclass, used for
tools with non-standard version strings (e.g. Tesseract).
env: Optional environment overrides applied when probing the version.
also_catch: Additional exception types that should be treated as
"not available" by :meth:`available`. :class:`OSError` is useful
for tools like verapdf whose launcher may fail with non-standard
errors when the JVM is missing.
"""
program: str
version_arg: str = '--version'
version_regex: str = r'(\d+(\.\d+)*)'
version_cls: type[Version] = Version
env: Mapping[str, str] | None = None
also_catch: tuple[type[BaseException], ...] = ()
def version(self) -> Version:
"""Return the installed version of the program.
Raises:
MissingDependencyError: if the program cannot be found or its
version string cannot be parsed.
"""
raw = get_version(
self.program,
version_arg=self.version_arg,
regex=self.version_regex,
env=self.env,
)
return self.version_cls(raw)
def available(self) -> bool:
"""Return whether a usable version of the program is installed."""
try:
self.version()
except MissingDependencyError:
return False
except self.also_catch:
return False
return True
src/ocrmypdf/_exec/ghostscript.py
+15 -4
View File
@@ -16,6 +16,7 @@ from subprocess import PIPE, CalledProcessError
from packaging.version import Version
from PIL import Image, UnidentifiedImageError
from ocrmypdf._exec._probe import ToolProbe
from ocrmypdf.exceptions import (
ColorConversionNeededError,
InputFileError,
@@ -23,7 +24,7 @@ from ocrmypdf.exceptions import (
)
from ocrmypdf.helpers import Resolution
from ocrmypdf.pluginspec import GhostscriptRasterDevice
from ocrmypdf.subprocess import get_version, run, run_polling_stderr
from ocrmypdf.subprocess import run, run_polling_stderr
COLOR_CONVERSION_STRATEGIES = frozenset(
[
@@ -69,11 +70,19 @@ class DuplicateFilter(logging.Filter):
return True
log.addFilter(DuplicateFilter(log))
PROBE = ToolProbe(program=GS)
version = PROBE.version
available = PROBE.available
def version() -> Version:
return Version(get_version(GS))
def _ensure_log_filter_installed() -> None:
"""Idempotently attach the duplicate-suppressing filter to the GS logger.
Called at the top of each work function so the filter is present in the
main process *and* in any subprocess worker that calls Ghostscript.
"""
if not any(isinstance(f, DuplicateFilter) for f in log.filters):
log.addFilter(DuplicateFilter(log))
def _gs_error_reported(stream) -> bool:
@@ -123,6 +132,7 @@ def rasterize_pdf(
use_cropbox: If True, rasterize the CropBox instead of MediaBox.
Default is False (use MediaBox).
"""
_ensure_log_filter_installed()
raster_dpi = raster_dpi.round(6)
if not page_dpi:
page_dpi = raster_dpi
@@ -273,6 +283,7 @@ def generate_pdfa(
progressbar_class=None,
stop_on_error: bool = False,
):
_ensure_log_filter_installed()
# Ghostscript's compression is all or nothing. We can either force all images
# to JPEG, force all to Flate/PNG, or let it decide how to encode the images.
# In most case it's best to let it decide.
src/ocrmypdf/_exec/jbig2enc.py
+6 -4
View File
@@ -9,21 +9,23 @@ from subprocess import PIPE, CalledProcessError
from packaging.version import Version
from ocrmypdf._exec._probe import ToolProbe
from ocrmypdf.exceptions import MissingDependencyError
from ocrmypdf.subprocess import get_version, run
from ocrmypdf.subprocess import run
_PROBE = ToolProbe(program='jbig2', version_regex=r'jbig2enc (\d+(\.\d+)*).*')
def version() -> Version:
try:
version = get_version('jbig2', regex=r'jbig2enc (\d+(\.\d+)*).*')
return _PROBE.version()
except CalledProcessError as e:
# TeX Live for Windows provides an incompatible jbig2.EXE which may
# be on the PATH.
raise MissingDependencyError('jbig2enc') from e
return Version(version)
def available():
def available() -> bool:
try:
version()
except MissingDependencyError:
src/ocrmypdf/_exec/pngquant.py
+5 -15
View File
@@ -8,22 +8,12 @@ from __future__ import annotations
from pathlib import Path
from subprocess import PIPE
from packaging.version import Version
from ocrmypdf._exec._probe import ToolProbe
from ocrmypdf.subprocess import run
from ocrmypdf.exceptions import MissingDependencyError
from ocrmypdf.subprocess import get_version, run
def version() -> Version:
return Version(get_version('pngquant', regex=r'(\d+(\.\d+)*).*'))
def available():
try:
version()
except MissingDependencyError:
return False
return True
PROBE = ToolProbe(program='pngquant', version_regex=r'(\d+(\.\d+)*).*')
version = PROBE.version
available = PROBE.available
def quantize(input_file: Path, output_file: Path, quality_min: int, quality_max: int):
src/ocrmypdf/_exec/tesseract.py
+10 -6
View File
@@ -17,13 +17,14 @@ from subprocess import PIPE, STDOUT, CalledProcessError, TimeoutExpired
from packaging.version import Version
from ocrmypdf._exec._probe import ToolProbe
from ocrmypdf.exceptions import (
MissingDependencyError,
SubprocessOutputError,
TesseractConfigError,
)
from ocrmypdf.pluginspec import OrientationConfidence
from ocrmypdf.subprocess import get_version, run
from ocrmypdf.subprocess import run
log = logging.getLogger(__name__)
@@ -115,8 +116,13 @@ class TesseractVersion(Version):
)
def version() -> Version:
return TesseractVersion(get_version('tesseract', regex=r'tesseract\s(.+)'))
PROBE = ToolProbe(
program='tesseract',
version_regex=r'tesseract\s(.+)',
version_cls=TesseractVersion,
)
version = PROBE.version
available = PROBE.available
def has_thresholding() -> bool:
@@ -287,9 +293,7 @@ def tesseract_log_output(stream: bytes) -> None:
lines = text.splitlines()
for line in lines:
if line.startswith(
("Tesseract Open Source", "Warning in pixReadMem")
):
if line.startswith(("Tesseract Open Source", "Warning in pixReadMem")):
continue
elif 'diacritics' in line:
tlog.warning("lots of diacritics - possibly poor OCR")
src/ocrmypdf/_exec/unpaper.py
+5 -4
View File
@@ -14,11 +14,11 @@ from pathlib import Path
from subprocess import PIPE, STDOUT
from tempfile import TemporaryDirectory
from packaging.version import Version
from PIL import Image
from ocrmypdf._exec._probe import ToolProbe
from ocrmypdf.exceptions import SubprocessOutputError
from ocrmypdf.subprocess import get_version, run
from ocrmypdf.subprocess import run
# unpaper documentation:
# https://github.com/Flameeyes/unpaper/blob/main/doc/basic-concepts.md
@@ -46,8 +46,9 @@ class UnpaperImageTooLargeError(Exception):
super().__init__(self.message)
def version() -> Version:
return Version(get_version('unpaper', regex=r'(?m).*?(\d+(\.\d+)(\.\d+)?)'))
PROBE = ToolProbe(program='unpaper', version_regex=r'(?m).*?(\d+(\.\d+)(\.\d+)?)')
version = PROBE.version
available = PROBE.available
@contextmanager
src/ocrmypdf/_exec/verapdf.py
+9 -15
View File
@@ -11,10 +11,9 @@ from pathlib import Path
from subprocess import PIPE
from typing import NamedTuple
from packaging.version import Version
from ocrmypdf._exec._probe import ToolProbe
from ocrmypdf.exceptions import MissingDependencyError
from ocrmypdf.subprocess import get_version, run
from ocrmypdf.subprocess import run
log = logging.getLogger(__name__)
@@ -27,18 +26,13 @@ class ValidationResult(NamedTuple):
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, OSError):
return False
return True
PROBE = ToolProbe(
program='verapdf',
version_regex=r'veraPDF (\d+(\.\d+)*)',
also_catch=(OSError,),
)
version = PROBE.version
available = PROBE.available
def output_type_to_flavour(output_type: str) -> str:
src/ocrmypdf/subprocess/__init__.py
+25 -339
View File
@@ -1,345 +1,31 @@
# SPDX-FileCopyrightText: 2022 James R. Barlow
# SPDX-License-Identifier: MPL-2.0
"""Wrappers to manage subprocess calls."""
"""Wrappers to manage subprocess calls.
This package is split into three private submodules by concern:
- :mod:`ocrmypdf.subprocess._run` - low-level execution wrappers (``run``,
``run_polling_stderr``) that add OCRmyPDF-aware logging and Windows PATH
resolution. Useful as drop-in replacements for :func:`subprocess.run`.
- :mod:`ocrmypdf.subprocess._version` - version probing (``get_version``).
- :mod:`ocrmypdf.subprocess._check` - startup validation
(``check_external_program``) with platform-aware error messages.
The names below are the stable public API. Importing from the private
submodules directly is not supported for external code.
"""
from __future__ import annotations
import logging
import os
import re
import sys
from collections.abc import Callable, Mapping, Sequence
from contextlib import suppress
from pathlib import Path
from subprocess import PIPE, STDOUT, CalledProcessError, CompletedProcess, Popen
from subprocess import run as subprocess_run
from ocrmypdf.subprocess._check import check_external_program
from ocrmypdf.subprocess._run import Args, Environ, run, run_polling_stderr
from ocrmypdf.subprocess._version import get_version
from packaging.version import Version
from ocrmypdf.exceptions import MissingDependencyError
# pylint: disable=logging-format-interpolation
log = logging.getLogger(__name__)
Args = Sequence[Path | str]
Environ = Mapping[str, str] | os._Environ # pylint: disable=protected-access
def run(
args: Args,
*,
env: Environ | None = None,
logs_errors_to_stdout: bool = False,
check: bool = False,
**kwargs,
) -> CompletedProcess:
"""Wrapper around :py:func:`subprocess.run`.
The main purpose of this wrapper is to log subprocess output in an orderly
fashion that identifies the responsible subprocess. An additional
task is that this function goes to greater lengths to find possible Windows
locations of our dependencies when they are not on the system PATH.
Arguments should be identical to ``subprocess.run``, except for following:
Args:
args: Positional arguments to pass to ``subprocess.run``.
env: A set of environment variables. If None, the OS environment is used.
logs_errors_to_stdout: If True, indicates that the process writes its error
messages to stdout rather than stderr, so stdout should be logged
if there is an error. If False, stderr is logged. Could be used with
stderr=STDOUT, stdout=PIPE for example.
check: If True, raise an exception if the process exits with a non-zero
status code. If False, the return value will indicate success or failure.
kwargs: Additional arguments to pass to ``subprocess.run``.
"""
args, env, process_log, _text = _fix_process_args(args, env, kwargs)
stderr = None
stderr_name = 'stderr' if not logs_errors_to_stdout else 'stdout'
try:
proc = subprocess_run(args, env=env, check=check, **kwargs)
except CalledProcessError as e:
stderr = getattr(e, stderr_name, None)
raise
else:
stderr = getattr(proc, stderr_name, None)
finally:
if process_log.isEnabledFor(logging.DEBUG) and stderr:
with suppress(AttributeError, UnicodeDecodeError):
stderr = stderr.decode('utf-8', 'replace')
if logs_errors_to_stdout:
process_log.debug("stdout/stderr = %s", stderr)
else:
process_log.debug("stderr = %s", stderr)
return proc
def run_polling_stderr(
args: Args,
*,
callback: Callable[[str], None],
check: bool = False,
env: Environ | None = None,
**kwargs,
) -> CompletedProcess:
"""Run a process like ``ocrmypdf.subprocess.run``, and poll stderr.
Every line of produced by stderr will be forwarded to the callback function.
The intended use is monitoring progress of subprocesses that output their
own progress indicators. In addition, each line will be logged if debug
logging is enabled.
Requires stderr to be opened in text mode for ease of handling errors. In
addition the expected encoding= and errors= arguments should be set. Note
that if stdout is already set up, it need not be binary.
"""
args, env, process_log, text = _fix_process_args(args, env, kwargs)
assert text, "Must use text=True"
with Popen(args, env=env, **kwargs) as proc:
lines = []
while proc.poll() is None:
if proc.stderr is None:
continue
for msg in iter(proc.stderr.readline, ''):
if process_log.isEnabledFor(logging.DEBUG):
process_log.debug(msg.strip())
callback(msg)
lines.append(msg)
stderr = ''.join(lines)
if check and proc.returncode != 0:
raise CalledProcessError(proc.returncode, args, output=None, stderr=stderr)
return CompletedProcess(args, proc.returncode, None, stderr=stderr)
def _fix_process_args(
args: Args, env: Environ | None, kwargs
) -> tuple[Args, Environ, logging.Logger, bool]:
if not env:
env = os.environ
# Search in spoof path if necessary
program = str(args[0])
if sys.platform == 'win32':
# pylint: disable=import-outside-toplevel
from ocrmypdf.subprocess._windows import fix_windows_args
args = fix_windows_args(program, args, env)
log.debug("Running: %s", args)
process_log = log.getChild(os.path.basename(program))
text = bool(kwargs.get('text', False))
return args, env, process_log, text
def get_version(
program: str,
*,
version_arg: str = '--version',
regex=r'(\d+(\.\d+)*)',
env: Environ | None = None,
) -> str:
"""Get the version of the specified program.
Arguments:
program: The program to version check.
version_arg: The argument needed to ask for its version, e.g. ``--version``.
regex: A regular expression to parse the program's output and obtain the
version.
env: Custom ``os.environ`` in which to run program.
"""
args_prog = [program, version_arg]
try:
proc = run(
args_prog,
close_fds=True,
text=True,
stdout=PIPE,
stderr=STDOUT,
check=True,
env=env,
)
output: str = proc.stdout
except FileNotFoundError as e:
raise MissingDependencyError(
f"Could not find program '{program}' on the PATH"
) from e
except CalledProcessError as e:
if e.returncode != 0:
log.exception(e)
raise MissingDependencyError(
f"Ran program '{program}' but it exited with an error:\n{e.output}"
) from e
raise MissingDependencyError(
f"Could not find program '{program}' on the PATH"
) from e
match = re.match(regex, output.strip())
if not match:
raise MissingDependencyError(
f"The program '{program}' did not report its version. "
f"Message was:\n{output}"
)
version = match.group(1)
return version
MISSING_PROGRAM = '''
The program '{program}' could not be executed or was not found on your
system PATH.
'''
MISSING_OPTIONAL_PROGRAM = '''
The program '{program}' could not be executed or was not found on your
system PATH. This program is required when you use the
{required_for} arguments. You could try omitting these arguments, or install
the package.
'''
MISSING_RECOMMEND_PROGRAM = '''
The program '{program}' could not be executed or was not found on your
system PATH. This program is recommended when using the {required_for} arguments,
but not required, so we will proceed. For best results, install the program.
'''
OLD_VERSION = '''
OCRmyPDF requires '{program}' {need_version} or higher. Your system appears
to have {found_version}. Please update this program.
'''
OLD_VERSION_REQUIRED_FOR = '''
OCRmyPDF requires '{program}' {need_version} or higher when run with the
{required_for} arguments. {program} {found_version} is installed.
If you omit these arguments, OCRmyPDF may be able to
proceed. For best results, update the program.
'''
OSX_INSTALL_ADVICE = '''
If you have homebrew installed, try these command to install the missing
package:
brew install {package}
'''
LINUX_INSTALL_ADVICE = '''
On systems with the aptitude package manager (Debian, Ubuntu), try these
commands:
sudo apt update
sudo apt install {package}
On RPM-based systems (Red Hat, Fedora), try this command:
sudo dnf install {package}
'''
WINDOWS_INSTALL_ADVICE = '''
If not already installed, install the Chocolatey package manager. Then use
a command prompt to install the missing package:
choco install {package}
'''
def _get_platform() -> str:
if sys.platform.startswith('freebsd'):
return 'freebsd'
elif sys.platform.startswith('linux'):
return 'linux'
elif sys.platform.startswith('win'):
return 'windows'
return sys.platform
def _error_trailer(program: str, package: str | Mapping[str, str], **kwargs) -> None:
del kwargs
if isinstance(package, Mapping):
package = package.get(_get_platform(), program)
if _get_platform() == 'darwin':
log.info(OSX_INSTALL_ADVICE.format(**locals()))
elif _get_platform() == 'linux':
log.info(LINUX_INSTALL_ADVICE.format(**locals()))
elif _get_platform() == 'windows':
log.info(WINDOWS_INSTALL_ADVICE.format(**locals()))
def _error_missing_program(
program: str, package: str, required_for: str | None, recommended: bool
) -> None:
# pylint: disable=unused-argument
if recommended:
log.warning(MISSING_RECOMMEND_PROGRAM.format(**locals()))
elif required_for:
log.error(MISSING_OPTIONAL_PROGRAM.format(**locals()))
else:
log.error(MISSING_PROGRAM.format(**locals()))
_error_trailer(**locals())
def _error_old_version(
program: str,
package: str,
need_version: str,
found_version: str,
required_for: str | None,
) -> None:
# pylint: disable=unused-argument
if required_for:
log.error(OLD_VERSION_REQUIRED_FOR.format(**locals()))
else:
log.error(OLD_VERSION.format(**locals()))
_error_trailer(**locals())
def check_external_program(
*,
program: str,
package: str,
version_checker: Callable[[], Version],
need_version: str | Version,
required_for: str | None = None,
recommended: bool = False,
version_parser: type[Version] = Version,
) -> None:
"""Check for required version of external program and raise exception if not.
Args:
program: The name of the program to test.
package: The name of a software package that typically supplies this program.
Usually the same as program.
version_checker: A callable without arguments that retrieves the installed
version of program.
need_version: The minimum required version.
required_for: The name of an argument of feature that requires this program.
recommended: If this external program is recommended, instead of raising
an exception, log a warning and allow execution to continue.
version_parser: A class that should be used to parse and compare version
numbers. Used when version numbers do not follow standard conventions.
"""
if not isinstance(need_version, Version):
need_version = version_parser(need_version)
try:
found_version = version_checker()
except (CalledProcessError, FileNotFoundError) as e:
_error_missing_program(program, package, required_for, recommended)
if not recommended:
raise MissingDependencyError(program) from e
return
except MissingDependencyError:
_error_missing_program(program, package, required_for, recommended)
if not recommended:
raise
return
if found_version and found_version < need_version:
_error_old_version(
program, package, str(need_version), str(found_version), required_for
)
if not recommended:
raise MissingDependencyError(program)
log.debug('Found %s %s', program, found_version)
__all__ = [
'Args',
'Environ',
'check_external_program',
'get_version',
'run',
'run_polling_stderr',
]
src/ocrmypdf/subprocess/_check.py
+171
View File
@@ -0,0 +1,171 @@
# SPDX-FileCopyrightText: 2022 James R. Barlow
# SPDX-License-Identifier: MPL-2.0
"""Validate that required external programs are installed and new enough."""
from __future__ import annotations
import logging
import sys
from collections.abc import Callable, Mapping
from subprocess import CalledProcessError
from packaging.version import Version
from ocrmypdf.exceptions import MissingDependencyError
log = logging.getLogger('ocrmypdf.subprocess')
MISSING_PROGRAM = '''
The program '{program}' could not be executed or was not found on your
system PATH.
'''
MISSING_OPTIONAL_PROGRAM = '''
The program '{program}' could not be executed or was not found on your
system PATH. This program is required when you use the
{required_for} arguments. You could try omitting these arguments, or install
the package.
'''
MISSING_RECOMMEND_PROGRAM = '''
The program '{program}' could not be executed or was not found on your
system PATH. This program is recommended when using the {required_for} arguments,
but not required, so we will proceed. For best results, install the program.
'''
OLD_VERSION = '''
OCRmyPDF requires '{program}' {need_version} or higher. Your system appears
to have {found_version}. Please update this program.
'''
OLD_VERSION_REQUIRED_FOR = '''
OCRmyPDF requires '{program}' {need_version} or higher when run with the
{required_for} arguments. {program} {found_version} is installed.
If you omit these arguments, OCRmyPDF may be able to
proceed. For best results, update the program.
'''
OSX_INSTALL_ADVICE = '''
If you have homebrew installed, try these command to install the missing
package:
brew install {package}
'''
LINUX_INSTALL_ADVICE = '''
On systems with the aptitude package manager (Debian, Ubuntu), try these
commands:
sudo apt update
sudo apt install {package}
On RPM-based systems (Red Hat, Fedora), try this command:
sudo dnf install {package}
'''
WINDOWS_INSTALL_ADVICE = '''
If not already installed, install the Chocolatey package manager. Then use
a command prompt to install the missing package:
choco install {package}
'''
def _get_platform() -> str:
if sys.platform.startswith('freebsd'):
return 'freebsd'
elif sys.platform.startswith('linux'):
return 'linux'
elif sys.platform.startswith('win'):
return 'windows'
return sys.platform
def _error_trailer(program: str, package: str | Mapping[str, str], **kwargs) -> None:
del kwargs
if isinstance(package, Mapping):
package = package.get(_get_platform(), program)
if _get_platform() == 'darwin':
log.info(OSX_INSTALL_ADVICE.format(**locals()))
elif _get_platform() == 'linux':
log.info(LINUX_INSTALL_ADVICE.format(**locals()))
elif _get_platform() == 'windows':
log.info(WINDOWS_INSTALL_ADVICE.format(**locals()))
def _error_missing_program(
program: str, package: str, required_for: str | None, recommended: bool
) -> None:
# pylint: disable=unused-argument
if recommended:
log.warning(MISSING_RECOMMEND_PROGRAM.format(**locals()))
elif required_for:
log.error(MISSING_OPTIONAL_PROGRAM.format(**locals()))
else:
log.error(MISSING_PROGRAM.format(**locals()))
_error_trailer(**locals())
def _error_old_version(
program: str,
package: str,
need_version: str,
found_version: str,
required_for: str | None,
) -> None:
# pylint: disable=unused-argument
if required_for:
log.error(OLD_VERSION_REQUIRED_FOR.format(**locals()))
else:
log.error(OLD_VERSION.format(**locals()))
_error_trailer(**locals())
def check_external_program(
*,
program: str,
package: str,
version_checker: Callable[[], Version],
need_version: str | Version,
required_for: str | None = None,
recommended: bool = False,
version_parser: type[Version] = Version,
) -> None:
"""Check for required version of external program and raise exception if not.
Args:
program: The name of the program to test.
package: The name of a software package that typically supplies this program.
Usually the same as program.
version_checker: A callable without arguments that retrieves the installed
version of program.
need_version: The minimum required version.
required_for: The name of an argument of feature that requires this program.
recommended: If this external program is recommended, instead of raising
an exception, log a warning and allow execution to continue.
version_parser: A class that should be used to parse and compare version
numbers. Used when version numbers do not follow standard conventions.
"""
if not isinstance(need_version, Version):
need_version = version_parser(need_version)
try:
found_version = version_checker()
except (CalledProcessError, FileNotFoundError) as e:
_error_missing_program(program, package, required_for, recommended)
if not recommended:
raise MissingDependencyError(program) from e
return
except MissingDependencyError:
_error_missing_program(program, package, required_for, recommended)
if not recommended:
raise
return
if found_version and found_version < need_version:
_error_old_version(
program, package, str(need_version), str(found_version), required_for
)
if not recommended:
raise MissingDependencyError(program)
log.debug('Found %s %s', program, found_version)
src/ocrmypdf/subprocess/_run.py
+137
View File
@@ -0,0 +1,137 @@
# SPDX-FileCopyrightText: 2022 James R. Barlow
# SPDX-License-Identifier: MPL-2.0
"""Low-level wrappers around :py:mod:`subprocess`.
These functions exist to give OCRmyPDF child processes uniform logging
behavior and to route through any platform-specific PATH fix-ups before
invocation. They are intended as drop-in replacements for
:py:func:`subprocess.run` in contexts where that routing is desirable
(for example, plugin-provided tools).
"""
from __future__ import annotations
import logging
import os
import sys
from collections.abc import Callable, Mapping, Sequence
from contextlib import suppress
from pathlib import Path
from subprocess import CalledProcessError, CompletedProcess, Popen
from subprocess import run as subprocess_run
log = logging.getLogger('ocrmypdf.subprocess')
Args = Sequence[Path | str]
Environ = Mapping[str, str] | os._Environ # pylint: disable=protected-access
def run(
args: Args,
*,
env: Environ | None = None,
logs_errors_to_stdout: bool = False,
check: bool = False,
**kwargs,
) -> CompletedProcess:
"""Wrapper around :py:func:`subprocess.run`.
The main purpose of this wrapper is to log subprocess output in an orderly
fashion that identifies the responsible subprocess. An additional
task is that this function goes to greater lengths to find possible Windows
locations of our dependencies when they are not on the system PATH.
Arguments should be identical to ``subprocess.run``, except for following:
Args:
args: Positional arguments to pass to ``subprocess.run``.
env: A set of environment variables. If None, the OS environment is used.
logs_errors_to_stdout: If True, indicates that the process writes its error
messages to stdout rather than stderr, so stdout should be logged
if there is an error. If False, stderr is logged. Could be used with
stderr=STDOUT, stdout=PIPE for example.
check: If True, raise an exception if the process exits with a non-zero
status code. If False, the return value will indicate success or failure.
kwargs: Additional arguments to pass to ``subprocess.run``.
"""
args, env, process_log, _text = _fix_process_args(args, env, kwargs)
stderr = None
stderr_name = 'stderr' if not logs_errors_to_stdout else 'stdout'
try:
proc = subprocess_run(args, env=env, check=check, **kwargs)
except CalledProcessError as e:
stderr = getattr(e, stderr_name, None)
raise
else:
stderr = getattr(proc, stderr_name, None)
finally:
if process_log.isEnabledFor(logging.DEBUG) and stderr:
with suppress(AttributeError, UnicodeDecodeError):
stderr = stderr.decode('utf-8', 'replace')
if logs_errors_to_stdout:
process_log.debug("stdout/stderr = %s", stderr)
else:
process_log.debug("stderr = %s", stderr)
return proc
def run_polling_stderr(
args: Args,
*,
callback: Callable[[str], None],
check: bool = False,
env: Environ | None = None,
**kwargs,
) -> CompletedProcess:
"""Run a process like ``ocrmypdf.subprocess.run``, and poll stderr.
Every line of produced by stderr will be forwarded to the callback function.
The intended use is monitoring progress of subprocesses that output their
own progress indicators. In addition, each line will be logged if debug
logging is enabled.
Requires stderr to be opened in text mode for ease of handling errors. In
addition the expected encoding= and errors= arguments should be set. Note
that if stdout is already set up, it need not be binary.
"""
args, env, process_log, text = _fix_process_args(args, env, kwargs)
assert text, "Must use text=True"
with Popen(args, env=env, **kwargs) as proc:
lines = []
while proc.poll() is None:
if proc.stderr is None:
continue
for msg in iter(proc.stderr.readline, ''):
if process_log.isEnabledFor(logging.DEBUG):
process_log.debug(msg.strip())
callback(msg)
lines.append(msg)
stderr = ''.join(lines)
if check and proc.returncode != 0:
raise CalledProcessError(proc.returncode, args, output=None, stderr=stderr)
return CompletedProcess(args, proc.returncode, None, stderr=stderr)
def _fix_process_args(
args: Args, env: Environ | None, kwargs
) -> tuple[Args, Environ, logging.Logger, bool]:
if not env:
env = os.environ
# Search in spoof path if necessary
program = str(args[0])
if sys.platform == 'win32':
# pylint: disable=import-outside-toplevel
from ocrmypdf.subprocess._windows import fix_windows_args
args = fix_windows_args(program, args, env)
log.debug("Running: %s", args)
process_log = log.getChild(os.path.basename(program))
text = bool(kwargs.get('text', False))
return args, env, process_log, text
src/ocrmypdf/subprocess/_version.py
+73
View File
@@ -0,0 +1,73 @@
# SPDX-FileCopyrightText: 2022 James R. Barlow
# SPDX-License-Identifier: MPL-2.0
"""Extract version strings from external programs."""
from __future__ import annotations
import logging
import re
from subprocess import PIPE, STDOUT, CalledProcessError
from ocrmypdf.exceptions import MissingDependencyError
from ocrmypdf.subprocess._run import Environ
log = logging.getLogger('ocrmypdf.subprocess')
def get_version(
program: str,
*,
version_arg: str = '--version',
regex=r'(\d+(\.\d+)*)',
env: Environ | None = None,
) -> str:
"""Get the version of the specified program.
Arguments:
program: The program to version check.
version_arg: The argument needed to ask for its version, e.g. ``--version``.
regex: A regular expression to parse the program's output and obtain the
version.
env: Custom ``os.environ`` in which to run program.
"""
# Late import of the public ``run`` so that tests patching
# ``ocrmypdf.subprocess.run`` affect this function. Binding ``run`` at
# module load time would capture the real implementation and bypass the
# patch.
from ocrmypdf import subprocess as _sp
args_prog = [program, version_arg]
try:
proc = _sp.run(
args_prog,
close_fds=True,
text=True,
stdout=PIPE,
stderr=STDOUT,
check=True,
env=env,
)
output: str = proc.stdout
except FileNotFoundError as e:
raise MissingDependencyError(
f"Could not find program '{program}' on the PATH"
) from e
except CalledProcessError as e:
if e.returncode != 0:
log.exception(e)
raise MissingDependencyError(
f"Ran program '{program}' but it exited with an error:\n{e.output}"
) from e
raise MissingDependencyError(
f"Could not find program '{program}' on the PATH"
) from e
match = re.match(regex, output.strip())
if not match:
raise MissingDependencyError(
f"The program '{program}' did not report its version. "
f"Message was:\n{output}"
)
version = match.group(1)
return version