Enable ruff PTH (flake8-use-pathlib) and fix all findings

Replaces os.path/open()/os.stat()/os.chmod() calls with their Path
method equivalents across src, tests, misc, and bin, wrapping str
variables in Path(...) where they must stay str for other uses (e.g.
subprocess argv, CLI-arg formatting). helpers.safe_symlink() now
decodes StrOrBytesPath to a str Path via os.fsdecode() upfront, same
pattern already used elsewhere for the str|bytes union.

bin/bump_version.py

@@ -6,7 +6,6 @@
 
 from __future__ import annotations
 
-import glob
 import os
 import subprocess
 import sys
@@ -282,7 +281,7 @@ def bump_version() -> None:
     actions = []
 
     for path_pattern, version_pattern in config:
-        paths = [Path(p) for p in glob.glob(path_pattern)]
+        paths = list(Path().glob(path_pattern))
 
         if not paths:
             print(f"error: Pattern {path_pattern} didn't match any files")

misc/batch.py

@@ -16,7 +16,6 @@ from __future__ import annotations
 
 import filecmp
 import logging
-import os
 import posixpath
 import shutil
 import sys
@@ -39,7 +38,7 @@ script_dir = Path(__file__).parent
 # set archive_dir to a path for backup original documents. Leave empty if not required.
 archive_dir = "/pdfbak"
 
-start_dir = Path(sys.argv[1]) if len(sys.argv) > 1 else Path(".")
+start_dir = Path(sys.argv[1]) if len(sys.argv) > 1 else Path()
 
 if len(sys.argv) > 2:
     log_file = Path(sys.argv[2])
@@ -68,7 +67,7 @@ for filename in start_dir.glob("**/*.pdf"):
             try:
                 shutil.copy2(filename, posixpath.dirname(archive_filename))
             except OSError:
-                os.makedirs(posixpath.dirname(archive_filename))
+                Path(posixpath.dirname(archive_filename)).mkdir(parents=True)
                 shutil.copy2(filename, posixpath.dirname(archive_filename))
         try:
             result = ocrmypdf.ocr(filename, filename, deskew=True)

misc/ocrmypdf_compare.py

@@ -37,8 +37,8 @@ def do_column(label, suffix, d):
             env[k] = v
     args = shlex.split(
         cli.format(
-            in_=os.path.join(d, "input.pdf"),
+            in_=Path(d) / "input.pdf",
-            out=os.path.join(d, f"output{suffix}.pdf"),
+            out=Path(d) / f"output{suffix}.pdf",
         )
     )
     with st.expander("Environment variables", expanded=bool(env_text.strip())):
@@ -106,8 +106,8 @@ def main():
                     )
                 )
 
-            doc1 = pymupdf.open(os.path.join(d, "output1.pdf"))
+            doc1 = pymupdf.open(Path(d, "output1.pdf"))
-            doc2 = pymupdf.open(os.path.join(d, "output2.pdf"))
+            doc2 = pymupdf.open(Path(d, "output2.pdf"))
             for i, page1_2 in enumerate(zip(doc1, doc2, strict=False)):
                 st.write(f"Page {i + 1}")
                 page1, page2 = page1_2

misc/pdf_compare.py

@@ -5,7 +5,6 @@
 
 from __future__ import annotations
 
-import os
 from io import BytesIO
 from pathlib import Path
 from tempfile import TemporaryDirectory
@@ -60,8 +59,8 @@ def main():
         Path(d, "2.pdf").write_bytes(pdf_bytes2)
 
         with st.expander("Text"):
-            doc1 = pymupdf.open(os.path.join(d, "1.pdf"))
+            doc1 = pymupdf.open(Path(d, "1.pdf"))
-            doc2 = pymupdf.open(os.path.join(d, "2.pdf"))
+            doc2 = pymupdf.open(Path(d, "2.pdf"))
             for i, page1_2 in enumerate(zip(doc1, doc2, strict=False)):
                 st.write(f"Page {i + 1}")
                 page1, page2 = page1_2

misc/pdf_text_diff.py

@@ -23,7 +23,7 @@ def main(
     engine: Annotated[str, cyclopts.Parameter()] = 'pdftotext',
 ):
     """Compare text in PDFs."""
-    with open(pdf1, 'rb') as f1, open(pdf2, 'rb') as f2:
+    with pdf1.open('rb') as f1, pdf2.open('rb') as f2:
         text1 = run(
             ['pdftotext', '-layout', '-', '-'],
             stdin=f1,

misc/synology.py

@@ -13,13 +13,14 @@ import shutil
 import subprocess
 import sys
 import time
+from pathlib import Path
 
 # pylint: disable=logging-format-interpolation
 # pylint: disable=logging-not-lazy
 
-script_dir = os.path.dirname(os.path.realpath(__file__))
+script_dir = Path(os.path.realpath(__file__)).parent
 timestamp = time.strftime("%Y-%m-%d-%H%M_")
-log_file = script_dir + '/' + timestamp + 'ocrmypdf.log'
+log_file = script_dir / (timestamp + 'ocrmypdf.log')
 logging.basicConfig(
     level=logging.INFO,
     format='%(asctime)s %(message)s',
@@ -33,10 +34,10 @@ for dir_name, _subdirs, file_list in os.walk(start_dir):
     logging.info(dir_name)
     os.chdir(dir_name)
     for filename in file_list:
-        file_stem, file_ext = os.path.splitext(filename)
+        file_stem, file_ext = Path(filename).stem, Path(filename).suffix
         if file_ext != '.pdf':
             continue
-        full_path = os.path.join(dir_name, filename)
+        full_path = Path(dir_name, filename)
         timestamp_ocr = time.strftime("%Y-%m-%d-%H%M_OCR_")
         filename_ocr = timestamp_ocr + file_stem + '.pdf'
         # create string for pdf processing
@@ -52,10 +53,10 @@ for dir_name, _subdirs, file_list in os.walk(start_dir):
             '-',
         ]
         logging.info(cmd)
-        full_path_ocr = os.path.join(dir_name, filename_ocr)
+        full_path_ocr = Path(dir_name, filename_ocr)
         with (
-            open(filename, 'rb') as input_file,
+            Path(filename).open('rb') as input_file,
-            open(full_path_ocr, 'wb') as output_file,
+            full_path_ocr.open('wb') as output_file,
         ):
             proc = subprocess.run(
                 cmd,
@@ -67,8 +68,8 @@ for dir_name, _subdirs, file_list in os.walk(start_dir):
                 errors='ignore',
             )
         logging.info(proc.stderr)
-        os.chmod(full_path_ocr, 0o664)
+        full_path_ocr.chmod(0o664)
-        os.chmod(full_path, 0o664)
+        full_path.chmod(0o664)
         full_path_ocr_archive = sys.argv[2]
         full_path_archive = sys.argv[2] + '/no_ocr'
         shutil.move(full_path_ocr, full_path_ocr_archive)

pyproject.toml

@@ -135,6 +135,7 @@ exclude = ["src/ocrmypdf/_version.py"] # Autogenerated
   "SIM", # simplify
   "B",   # flake8-bugbear
   "ICN", # flake8-import-conventions
+  "PTH", # flake8-use-pathlib
 ]
 ignore = [
   "B028", # warning with no explicit stacklevel

src/ocrmypdf/_exec/jbig2enc.py

@@ -35,7 +35,7 @@ def available() -> bool:
 
 def convert_single(cwd, infile, outfile, threshold):
     args = ['jbig2', '--pdf', '-t', str(threshold), infile]
-    with open(outfile, 'wb') as fstdout:
+    with outfile.open('wb') as fstdout:
         proc = run(args, cwd=cwd, stdout=fstdout, stderr=PIPE)
     proc.check_returncode()
     return proc

src/ocrmypdf/_exec/pngquant.py

@@ -25,7 +25,7 @@ def quantize(input_file: Path, output_file: Path, quality_min: int, quality_max:
         quality_min: Minimum quality to use
         quality_max: Maximum quality to use
     """
-    with open(input_file, 'rb') as input_stream:
+    with input_file.open('rb') as input_stream:
         args = [
             'pngquant',
             '--force',

src/ocrmypdf/_metadata.py

@@ -7,7 +7,6 @@ from __future__ import annotations
 
 import datetime as dt
 import logging
-import os
 from pathlib import Path
 from typing import Any
 
@@ -103,7 +102,7 @@ def should_linearize(working_file: Path, context: PdfContext) -> bool:
 
     For smaller files, linearization is not worth the effort.
     """
-    filesize = os.stat(working_file).st_size
+    filesize = working_file.stat().st_size
     return filesize > (context.options.fast_web_view * 1_000_000)
 
 

src/ocrmypdf/_pipeline.py

@@ -140,7 +140,7 @@ def triage_image_file(input_file: Path, output_file: Path, options: OcrOptions)
             layout_fun = img2pdf.get_fixed_dpi_layout_fun(
                 Resolution(options.image_dpi, options.image_dpi)
             )
-        with open(output_file, 'wb') as outf:
+        with output_file.open('wb') as outf:
             img2pdf.convert(
                 os.fspath(input_file),
                 layout_fun=layout_fun,
@@ -159,7 +159,7 @@ def _pdf_guess_version(input_file: Path, search_window=1024) -> str:
 
     Returns empty string if not found, indicating file is probably not PDF.
     """
-    with open(input_file, 'rb') as f:
+    with input_file.open('rb') as f:
         signature = f.read(search_window)
     m = re.search(rb'%PDF-(\d\.\d)', signature)
     if m:
@@ -833,7 +833,7 @@ def create_pdf_page_from_image(
 
     # Create a new single page PDF to hold
     bio = BytesIO()
-    with open(image, 'rb') as imfile:
+    with image.open('rb') as imfile:
         log.debug('convert')
 
         layout_fun = img2pdf.get_layout_fun(pagesize)
@@ -1199,7 +1199,7 @@ def should_linearize(working_file: Path, context: PdfContext) -> bool:
 
     For smaller files, linearization is not worth the effort.
     """
-    filesize = os.stat(working_file).st_size
+    filesize = working_file.stat().st_size
     return filesize > (context.options.fast_web_view * 1_000_000)
 
 
@@ -1312,7 +1312,7 @@ def merge_sidecars(txt_files: Iterable[Path | None], context: PdfContext) -> Pat
     and returns the path to the merged file.
     """
     output_file = context.get_path('sidecar.txt')
-    with open(output_file, 'w', encoding="utf-8") as stream:
+    with output_file.open('w', encoding="utf-8") as stream:
         for (from_, to_), txt_file in enumerate_compress_ranges(txt_files):
             if from_ != 1:
                 stream.write('\f')  # Form feed between pages for all pages after first
@@ -1360,5 +1360,5 @@ def copy_final(input_file: Path, output_file: PathOrIO) -> None:
             # get the appropriate umask, ownership, etc.
             # The `hasattr` check above already ruled out stream-like objects.
             assert isinstance(output_file, str | bytes | os.PathLike)
-            with open(output_file, 'w+b') as output_stream:
+            with Path(os.fsdecode(output_file)).open('w+b') as output_stream:
                 copyfileobj(input_stream, output_stream)

src/ocrmypdf/_validation.py

@@ -200,7 +200,7 @@ def create_input_file(options: OcrOptions, work_folder: Path) -> tuple[Path, str
         # stdin
         log.info('reading file from standard input')
         target = work_folder / 'stdin'
-        with open(target, 'wb') as stream_buffer:
+        with target.open('wb') as stream_buffer:
             copyfileobj(sys.stdin.buffer, stream_buffer)
         return target, "stdin"
     elif hasattr(options.input_file, 'readable'):
@@ -209,7 +209,7 @@ def create_input_file(options: OcrOptions, work_folder: Path) -> tuple[Path, str
             raise InputFileError("Input file stream is not readable")
         log.info('reading file from input stream')
         target = work_folder / 'stream'
-        with open(target, 'wb') as stream_buffer:
+        with target.open('wb') as stream_buffer:
             copyfileobj(input_stream, stream_buffer)
         return target, "stream"
     else:

src/ocrmypdf/helpers.py

@@ -148,11 +148,11 @@ def safe_symlink(input_file: StrOrBytesPath, soft_link_name: StrOrBytesPath) ->
     used since symlinks may require administrator privileges. An existing link at the
     destination is removed.
     """
-    input_file = os.fspath(input_file)
+    input_path = Path(os.fsdecode(input_file))
-    soft_link_name = os.fspath(soft_link_name)
+    soft_link_path = Path(os.fsdecode(soft_link_name))
 
     # Guard against soft linking to oneself
-    if input_file == soft_link_name:
+    if input_path == soft_link_path:
         log.warning(
             "No symbolic link created. You are using the original data directory "
             "as the working directory."
@@ -160,28 +160,24 @@ def safe_symlink(input_file: StrOrBytesPath, soft_link_name: StrOrBytesPath) ->
         return
 
     # Soft link already exists: delete for relink?
-    if os.path.lexists(soft_link_name):
+    if os.path.lexists(soft_link_path):
         # do not delete or overwrite real (non-soft link) file
-        if not os.path.islink(soft_link_name):
+        if not soft_link_path.is_symlink():
-            raise FileExistsError(
+            raise FileExistsError(f"{soft_link_path} exists and is not a link")
-                f"{os.fsdecode(soft_link_name)} exists and is not a link"
+        soft_link_path.unlink()
-            )
-        os.unlink(soft_link_name)
 
-    if not os.path.exists(input_file):
+    if not input_path.exists():
-        raise FileNotFoundError(
+        raise FileNotFoundError(f"trying to create a broken symlink to {input_path}")
-            f"trying to create a broken symlink to {os.fsdecode(input_file)}"
-        )
 
     if os.name == 'nt':
         # Don't actually use symlinks on Windows due to permission issues
-        shutil.copyfile(input_file, soft_link_name)
+        shutil.copyfile(input_path, soft_link_path)
         return
 
-    log.debug("os.symlink(%s, %s)", input_file, soft_link_name)
+    log.debug("os.symlink(%s, %s)", input_path, soft_link_path)
 
     # Create symbolic link using absolute path
-    os.symlink(os.path.abspath(input_file), soft_link_name)
+    soft_link_path.symlink_to(input_path.resolve())
 
 
 def samefile(file1: os.PathLike, file2: os.PathLike) -> bool:
@@ -192,7 +188,7 @@ def samefile(file1: os.PathLike, file2: os.PathLike) -> bool:
     if os.name == 'nt':
         return file1 == file2
     else:
-        return os.path.samefile(file1, file2)
+        return Path(file1).samefile(file2)
 
 
 def is_iterable_notstr(thing: Any) -> bool:
@@ -207,7 +203,7 @@ def monotonic(seq: Sequence) -> bool:
 
 def page_number(input_file: os.PathLike) -> int:
     """Get one-based page number implied by filename (000002.pdf -> 2)."""
-    return int(os.path.basename(os.fspath(input_file))[0:6])
+    return int(Path(input_file).name[0:6])
 
 
 def available_cpu_count() -> int:

src/ocrmypdf/subprocess/_run.py

@@ -131,7 +131,7 @@ def _fix_process_args(
         args = fix_windows_args(program, args, env)
 
     log.debug("Running: %s", args)
-    process_log = log.getChild(os.path.basename(program))
+    process_log = log.getChild(Path(program).name)
     text = bool(kwargs.get('text', False))
 
     return args, env, process_log, text

tests/plugins/tesseract_debug_rotate.py

@@ -74,11 +74,11 @@ class FixedRotateNoopOcrEngine(OcrEngine):
     def generate_hocr(input_file, output_hocr, output_text, options):
         with (
             Image.open(input_file) as im,
-            open(output_hocr, 'w', encoding='utf-8') as f,
+            output_hocr.open('w', encoding='utf-8') as f,
         ):
             w, h = im.size
             f.write(HOCR_TEMPLATE.format(str(w), str(h)))
-        with open(output_text, 'w') as f:
+        with output_text.open('w') as f:
             f.write('')
 
     @staticmethod

tests/plugins/tesseract_noop.py

@@ -72,11 +72,11 @@ class NoopOcrEngine(OcrEngine):
     def generate_hocr(input_file, output_hocr, output_text, options):
         with (
             Image.open(input_file) as im,
-            open(output_hocr, 'w', encoding='utf-8') as f,
+            output_hocr.open('w', encoding='utf-8') as f,
         ):
             w, h = im.size
             f.write(HOCR_TEMPLATE.format(str(w), str(h)))
-        with open(output_text, 'w') as f:
+        with output_text.open('w') as f:
             f.write('')
 
     @staticmethod

tests/test_helpers.py

@@ -121,8 +121,8 @@ def test_shim_paths(tmp_path):
     results = result_str.split(os.pathsep)
     assert results[0] == str(syspath), results
     assert results[-3].endswith('tesseract-ocr'), results
-    assert results[-2].endswith(os.path.join('gs9.52.3', 'bin')), results
+    assert results[-2].endswith(str(Path('gs9.52.3', 'bin'))), results
-    assert results[-1].endswith(os.path.join('gs', '9.51', 'bin')), results
+    assert results[-1].endswith(str(Path('gs', '9.51', 'bin'))), results
 
 
 def test_resolution():

tests/test_hocrtransform.py

@@ -27,7 +27,7 @@ from .conftest import check_ocrmypdf
 
 def text_from_pdf(filename):
     output_string = StringIO()
-    with open(filename, 'rb') as in_file:
+    with filename.open('rb') as in_file:
         parser = PDFParser(in_file)
         doc = PDFDocument(parser)
         rsrcmgr = PDFResourceManager()

tests/test_main.py

@@ -107,7 +107,7 @@ def test_redo_ocr(resources, outpdf):
 
 def test_argsfile(resources, outdir):
     path_argsfile = outdir / 'test_argsfile.txt'
-    with open(str(path_argsfile), 'w') as argsfile:
+    with path_argsfile.open('w') as argsfile:
         print(
             '--title',
             'ArgsFile Test',
@@ -646,7 +646,7 @@ def test_compression_preserved(ocrmypdf_exec, resources, image, outpdf):
 
     im = Image.open(input_file)
     # Runs: ocrmypdf - output.pdf < testfile
-    with open(input_file, 'rb') as input_stream:
+    with Path(input_file).open('rb') as input_stream:
         p_args = ocrmypdf_exec + [
             '--optimize',
             '0',
@@ -704,7 +704,7 @@ def test_compression_changed(ocrmypdf_exec, resources, image, compression, outpd
     im = Image.open(input_file)
 
     # Runs: ocrmypdf - output.pdf < testfile
-    with open(input_file, 'rb') as input_stream:
+    with Path(input_file).open('rb') as input_stream:
         p_args = ocrmypdf_exec + [
             '--image-dpi',
             '150',
@@ -763,7 +763,7 @@ def test_sidecar_pagecount(resources, outpdf):
     pdfinfo = PdfInfo(resources / '3small.pdf')
     num_pages = len(pdfinfo)
 
-    with open(sidecar, encoding='utf-8') as f:
+    with sidecar.open(encoding='utf-8') as f:
         ocr_text = f.read()
 
     # There should a formfeed between each pair of pages, so the count of
@@ -784,7 +784,7 @@ def test_sidecar_nonempty(resources, outpdf):
         'tests/plugins/tesseract_cache.py',
     )
 
-    with open(sidecar, encoding='utf-8') as f:
+    with sidecar.open(encoding='utf-8') as f:
         ocr_text = f.read()
     assert 'the' in ocr_text
 

tests/test_metadata.py

@@ -137,7 +137,7 @@ def test_unset_metadata(output_type, field, resources, outpdf, caplog):
     # isn't contained anywhere in the output pdf. We'll also check to ensure
     # it's in the input pdf and that any values not unset are still in the
     # output pdf.
-    with open(input_file, 'rb') as before, open(outpdf, 'rb') as after:
+    with input_file.open('rb') as before, outpdf.open('rb') as after:
         before_data = before.read()
         after_data = after.read()
 

tests/test_pdf_renderer.py

@@ -32,7 +32,7 @@ from ocrmypdf.hocrtransform import (
 def text_from_pdf(filename: Path) -> str:
     """Extract text from a PDF file using pdfminer."""
     output_string = StringIO()
-    with open(filename, 'rb') as in_file:
+    with filename.open('rb') as in_file:
         parser = PDFParser(in_file)
         doc = PDFDocument(parser)
         rsrcmgr = PDFResourceManager()

tests/test_stdio.py

@@ -4,6 +4,7 @@
 from __future__ import annotations
 
 import os
+from pathlib import Path
 from subprocess import DEVNULL, PIPE, run
 
 import pytest
@@ -18,7 +19,7 @@ def test_stdin(ocrmypdf_exec, resources, outpdf):
     output_file = str(outpdf)
 
     # Runs: ocrmypdf - output.pdf < testfile.pdf
-    with open(input_file, 'rb') as input_stream:
+    with Path(input_file).open('rb') as input_stream:
         p_args = ocrmypdf_exec + [
             '-',
             output_file,
@@ -36,7 +37,7 @@ def test_stdout(ocrmypdf_exec, resources, outpdf):
     output_file = str(outpdf)
 
     # Runs: ocrmypdf francais.pdf - > test_stdout.pdf
-    with open(output_file, 'wb') as output_stream:
+    with Path(output_file).open('wb') as output_stream:
         p_args = ocrmypdf_exec + [
             input_file,
             '-',
@@ -58,7 +59,7 @@ def test_stdout_protected_from_pollution(ocrmypdf_exec, resources, outpdf):
     # A plugin deliberately writes garbage to stdout during the run. With stdout
     # protection active, that garbage must be diverted to stderr and never reach
     # the PDF we are writing to stdout.
-    with open(output_file, 'wb') as output_stream:
+    with Path(output_file).open('wb') as output_stream:
         p_args = ocrmypdf_exec + [
             input_file,
             '-',
@@ -70,7 +71,7 @@ def test_stdout_protected_from_pollution(ocrmypdf_exec, resources, outpdf):
         p = run(p_args, stdout=output_stream, stderr=PIPE, stdin=DEVNULL, check=True)
 
     assert check_pdf(output_file), "PDF on stdout was corrupted"
-    with open(output_file, 'rb') as f:
+    with Path(output_file).open('rb') as f:
         assert b'POLLUTION' not in f.read(), "pollution leaked into the PDF"
     assert b'POLLUTION' in p.stderr, "pollution was not diverted to stderr"