Convert many uses of str paths to Path
This commit is contained in:
@@ -47,7 +47,7 @@ def log_listener(queue):
|
||||
logger = logging.getLogger(record.name)
|
||||
logger.handle(record)
|
||||
except Exception: # pylint: disable=broad-except
|
||||
import traceback
|
||||
import traceback # pylint: disable=import-outside-toplevel
|
||||
|
||||
print("Logging problem", file=sys.stderr)
|
||||
traceback.print_exc(file=sys.stderr)
|
||||
|
||||
@@ -16,7 +16,6 @@
|
||||
# along with OCRmyPDF. If not, see <http://www.gnu.org/licenses/>.
|
||||
|
||||
import logging
|
||||
import os
|
||||
from contextlib import suppress
|
||||
from pathlib import Path
|
||||
|
||||
@@ -181,7 +180,7 @@ def _find_font(text, pdf_base):
|
||||
class OcrGrafter:
|
||||
def __init__(self, context):
|
||||
self.context = context
|
||||
self.path_base = Path(context.origin).resolve()
|
||||
self.path_base = context.origin
|
||||
|
||||
self.pdf_base = pikepdf.open(self.path_base)
|
||||
self.font, self.font_key = None, None
|
||||
@@ -264,12 +263,14 @@ class OcrGrafter:
|
||||
# {interim_count} is the opened file we were updateing
|
||||
# {interim_count - 1} can be deleted
|
||||
# {interim_count + 1} is the new file will produce and open
|
||||
old_file = self.output_file + f'_working{self.interim_count - 1}.pdf'
|
||||
old_file = self.output_file.with_suffix(f'.working{self.interim_count - 1}.pdf')
|
||||
if not self.context.options.keep_temporary_files:
|
||||
with suppress(FileNotFoundError):
|
||||
os.unlink(old_file)
|
||||
old_file.unlink()
|
||||
|
||||
next_file = self.output_file + f'_working{self.interim_count + 1}.pdf'
|
||||
next_file = self.output_file.with_suffix(
|
||||
f'.working{self.interim_count + 1}.pdf'
|
||||
)
|
||||
self.pdf_base.save(next_file)
|
||||
self.pdf_base.close()
|
||||
|
||||
|
||||
@@ -19,6 +19,7 @@ import os
|
||||
import shutil
|
||||
import sys
|
||||
from functools import partial
|
||||
from pathlib import Path
|
||||
|
||||
from ocrmypdf._plugin_manager import get_plugin_manager
|
||||
|
||||
@@ -26,10 +27,12 @@ from ocrmypdf._plugin_manager import get_plugin_manager
|
||||
class PdfContext:
|
||||
"""Holds our context for a particular run of the pipeline"""
|
||||
|
||||
def __init__(self, options, work_folder, origin, pdfinfo, plugin_manager):
|
||||
def __init__(
|
||||
self, options, work_folder: Path, origin: Path, pdfinfo, plugin_manager
|
||||
):
|
||||
self.options = options
|
||||
self.work_folder = work_folder
|
||||
self.origin = origin
|
||||
self.work_folder = Path(work_folder)
|
||||
self.origin = Path(origin)
|
||||
self.pdfinfo = pdfinfo
|
||||
self.plugin_manager = plugin_manager
|
||||
if options:
|
||||
@@ -39,8 +42,8 @@ class PdfContext:
|
||||
if self.name == '-':
|
||||
self.name = 'stdin'
|
||||
|
||||
def get_path(self, name):
|
||||
return os.path.join(self.work_folder, name)
|
||||
def get_path(self, name: str) -> Path:
|
||||
return self.work_folder / name
|
||||
|
||||
def get_page_contexts(self):
|
||||
npages = len(self.pdfinfo)
|
||||
@@ -54,7 +57,7 @@ class PageContext:
|
||||
Must be pickable, so only store intrinsic/simple data elements
|
||||
"""
|
||||
|
||||
def __init__(self, pdf_context, pageno):
|
||||
def __init__(self, pdf_context: PdfContext, pageno):
|
||||
self.work_folder = pdf_context.work_folder
|
||||
self.origin = pdf_context.origin
|
||||
self.options = pdf_context.options
|
||||
@@ -63,8 +66,8 @@ class PageContext:
|
||||
self.pageinfo = pdf_context.pdfinfo[pageno]
|
||||
self.plugin_manager = pdf_context.plugin_manager
|
||||
|
||||
def get_path(self, name):
|
||||
return os.path.join(self.work_folder, "%06d_%s" % (self.pageno + 1, name))
|
||||
def get_path(self, name: str) -> Path:
|
||||
return self.work_folder / ("%06d_%s" % (self.pageno + 1, name))
|
||||
|
||||
def __getstate__(self):
|
||||
state = self.__dict__.copy()
|
||||
|
||||
@@ -55,7 +55,7 @@ def triage_image_file(input_file, output_file, options):
|
||||
im = Image.open(input_file)
|
||||
except EnvironmentError as e:
|
||||
# Recover the original filename
|
||||
log.error(str(e).replace(input_file, options.input_file))
|
||||
log.error(str(e).replace(str(input_file), str(options.input_file)))
|
||||
raise UnsupportedImageFormatError() from e
|
||||
|
||||
with im:
|
||||
@@ -102,7 +102,10 @@ def triage_image_file(input_file, output_file, options):
|
||||
)
|
||||
with open(output_file, 'wb') as outf:
|
||||
img2pdf.convert(
|
||||
input_file, layout_fun=layout_fun, with_pdfrw=False, outputstream=outf
|
||||
os.fspath(input_file),
|
||||
layout_fun=layout_fun,
|
||||
with_pdfrw=False,
|
||||
outputstream=outf,
|
||||
)
|
||||
log.info("Successfully converted to PDF, processing...")
|
||||
except img2pdf.ImageOpenError as e:
|
||||
@@ -139,7 +142,7 @@ def triage(original_filename, input_file, output_file, options):
|
||||
return output_file
|
||||
except EnvironmentError as e:
|
||||
log.debug(f"Temporary file was at: {input_file}")
|
||||
msg = str(e).replace(input_file, original_filename)
|
||||
msg = str(e).replace(str(input_file), original_filename)
|
||||
raise InputFileError(msg) from e
|
||||
|
||||
triage_image_file(input_file, output_file, options)
|
||||
|
||||
@@ -304,7 +304,7 @@ def run_pipeline(options, *, plugin_manager, api=False):
|
||||
if not plugin_manager:
|
||||
plugin_manager = get_plugin_manager([])
|
||||
|
||||
work_folder = mkdtemp(prefix="com.github.ocrmypdf.")
|
||||
work_folder = Path(mkdtemp(prefix="com.github.ocrmypdf."))
|
||||
debug_log_handler = None
|
||||
if (options.keep_temporary_files or options.verbose >= 1) and not os.environ.get(
|
||||
'PYTEST_CURRENT_TEST', ''
|
||||
@@ -317,10 +317,7 @@ def run_pipeline(options, *, plugin_manager, api=False):
|
||||
|
||||
# Triage image or pdf
|
||||
origin_pdf = triage(
|
||||
original_filename,
|
||||
start_input_file,
|
||||
os.path.join(work_folder, 'origin.pdf'),
|
||||
options,
|
||||
original_filename, start_input_file, work_folder / 'origin.pdf', options
|
||||
)
|
||||
|
||||
plugin_manager.hook.prepare(options=options)
|
||||
|
||||
@@ -376,17 +376,17 @@ def log_page_orientations(pdfinfo):
|
||||
log.info('Page orientations detected: %s', ' '.join(orientations))
|
||||
|
||||
|
||||
def create_input_file(options, work_folder):
|
||||
def create_input_file(options, work_folder: Path) -> (Path, str):
|
||||
if options.input_file == '-':
|
||||
# stdin
|
||||
log.info('reading file from standard input')
|
||||
target = os.path.join(work_folder, 'stdin')
|
||||
target = work_folder / 'stdin'
|
||||
with open(target, 'wb') as stream_buffer:
|
||||
copyfileobj(sys.stdin.buffer, stream_buffer)
|
||||
return target, "<stdin>"
|
||||
else:
|
||||
try:
|
||||
target = os.path.join(work_folder, 'origin')
|
||||
target = work_folder / 'origin'
|
||||
safe_symlink(options.input_file, target)
|
||||
return target, os.fspath(options.input_file)
|
||||
except FileNotFoundError:
|
||||
|
||||
@@ -24,6 +24,7 @@ import logging
|
||||
import os
|
||||
import shlex
|
||||
from functools import lru_cache
|
||||
from pathlib import Path
|
||||
from subprocess import PIPE, STDOUT, CalledProcessError
|
||||
from tempfile import TemporaryDirectory
|
||||
|
||||
@@ -67,8 +68,8 @@ def run(input_file, output_file, dpi, mode_args):
|
||||
"Failed to convert image to a supported format."
|
||||
) from e
|
||||
|
||||
input_pnm = os.path.join(tmpdir, f'input{suffix}')
|
||||
output_pnm = os.path.join(tmpdir, f'output{suffix}')
|
||||
input_pnm = Path(tmpdir) / f'input{suffix}'
|
||||
output_pnm = Path(tmpdir) / f'output{suffix}'
|
||||
im.save(input_pnm, format='PPM')
|
||||
|
||||
# To prevent any shenanigans from accepting arbitrary parameters in
|
||||
@@ -78,7 +79,7 @@ def run(input_file, output_file, dpi, mode_args):
|
||||
# 3) append absolute paths for the input and output file
|
||||
# This should ensure that a user cannot clobber some other file with
|
||||
# their unpaper arguments (whether intentionally or otherwise)
|
||||
args_unpaper.extend([input_pnm, output_pnm])
|
||||
args_unpaper.extend([os.fspath(input_pnm), os.fspath(output_pnm)])
|
||||
try:
|
||||
proc = external_run(
|
||||
args_unpaper,
|
||||
|
||||
@@ -29,9 +29,11 @@
|
||||
# SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
|
||||
|
||||
import argparse
|
||||
import os
|
||||
import re
|
||||
from collections import namedtuple
|
||||
from math import atan, cos, sin
|
||||
from pathlib import Path
|
||||
from xml.etree import ElementTree
|
||||
|
||||
from reportlab.lib.units import inch
|
||||
@@ -155,8 +157,8 @@ class HocrTransform:
|
||||
|
||||
def to_pdf(
|
||||
self,
|
||||
out_filename: str,
|
||||
image_filename: str = None,
|
||||
out_filename: Path,
|
||||
image_filename: Path = None,
|
||||
show_bounding_boxes: bool = False,
|
||||
fontname: str = "Helvetica",
|
||||
invisible_text: bool = False,
|
||||
@@ -173,7 +175,9 @@ class HocrTransform:
|
||||
# create the PDF file
|
||||
# page size in points (1/72 in.)
|
||||
pdf = Canvas(
|
||||
out_filename, pagesize=(self.width, self.height), pageCompression=1
|
||||
os.fspath(out_filename),
|
||||
pagesize=(self.width, self.height),
|
||||
pageCompression=1,
|
||||
)
|
||||
|
||||
# draw bounding box for each paragraph
|
||||
@@ -226,7 +230,9 @@ class HocrTransform:
|
||||
)
|
||||
# put the image on the page, scaled to fill the page
|
||||
if image_filename is not None:
|
||||
pdf.drawImage(image_filename, 0, 0, width=self.width, height=self.height)
|
||||
pdf.drawImage(
|
||||
os.fspath(image_filename), 0, 0, width=self.width, height=self.height
|
||||
)
|
||||
|
||||
# finish up the page and save it
|
||||
pdf.showPage()
|
||||
|
||||
@@ -820,9 +820,8 @@ class PdfInfo:
|
||||
|
||||
|
||||
def main():
|
||||
# pylint: disable=import-outside-toplevel
|
||||
import argparse
|
||||
from pprint import pprint
|
||||
import argparse # pylint: disable=import-outside-toplevel
|
||||
from pprint import pprint # pylint: disable=import-outside-toplevel
|
||||
|
||||
parser = argparse.ArgumentParser()
|
||||
parser.add_argument('infile')
|
||||
|
||||
+7
-7
@@ -52,7 +52,7 @@ def is_macos():
|
||||
def running_in_docker():
|
||||
# Docker creates a file named /.dockerenv (newer versions) or
|
||||
# /.dockerinit (older) -- this is undocumented, not an offical test
|
||||
return os.path.exists('/.dockerenv') or os.path.exists('/.dockerinit')
|
||||
return Path('/.dockerenv').exists() or Path('/.dockerinit').exists()
|
||||
|
||||
|
||||
@pytest.helpers.register
|
||||
@@ -69,9 +69,9 @@ def have_unpaper():
|
||||
return True
|
||||
|
||||
|
||||
TESTS_ROOT = os.path.abspath(os.path.dirname(__file__))
|
||||
SPOOF_PATH = os.path.join(TESTS_ROOT, 'spoof')
|
||||
PROJECT_ROOT = os.path.dirname(TESTS_ROOT)
|
||||
TESTS_ROOT = Path(__file__).parent.resolve()
|
||||
SPOOF_PATH = TESTS_ROOT / 'spoof'
|
||||
PROJECT_ROOT = TESTS_ROOT
|
||||
OCRMYPDF = [sys.executable, '-m', 'ocrmypdf']
|
||||
|
||||
|
||||
@@ -146,7 +146,7 @@ def spoof(tmp_path_factory, **kwargs):
|
||||
tmpdir.mkdir(parents=True)
|
||||
|
||||
for replace_program, with_spoof in kwargs.items():
|
||||
spoofer = Path(SPOOF_PATH) / with_spoof
|
||||
spoofer = SPOOF_PATH / with_spoof
|
||||
if os.name != 'nt':
|
||||
spoofer.chmod(0o755)
|
||||
(tmpdir / replace_program).symlink_to(spoofer)
|
||||
@@ -224,8 +224,8 @@ def check_ocrmypdf(input_file, output_file, *args, env=None):
|
||||
result = api.run_pipeline(options, plugin_manager=None, api=True)
|
||||
|
||||
assert result == 0
|
||||
assert os.path.exists(str(output_file)), "Output file not created"
|
||||
assert os.stat(str(output_file)).st_size > 100, "PDF too small or empty"
|
||||
assert output_file.exists(), "Output file not created"
|
||||
assert output_file.stat().st_size > 100, "PDF too small or empty"
|
||||
|
||||
return output_file
|
||||
|
||||
|
||||
+3
-3
@@ -210,7 +210,7 @@ def test_force_ocr_on_pdf_with_no_images(spoof_tesseract_crash, resources, no_ou
|
||||
resources / 'blank.pdf', no_outpdf, '--force-ocr', env=spoof_tesseract_crash
|
||||
)
|
||||
assert p.returncode == ExitCode.child_process_error
|
||||
assert not os.path.exists(no_outpdf)
|
||||
assert not no_outpdf.exists()
|
||||
|
||||
|
||||
@pytest.mark.skipif(
|
||||
@@ -321,7 +321,7 @@ def test_tesseract_crash(renderer, spoof_tesseract_crash, resources, no_outpdf):
|
||||
env=spoof_tesseract_crash,
|
||||
)
|
||||
assert p.returncode == ExitCode.child_process_error
|
||||
assert not os.path.exists(no_outpdf)
|
||||
assert not no_outpdf.exists()
|
||||
assert "SubprocessOutputError" in err
|
||||
|
||||
|
||||
@@ -330,7 +330,7 @@ def test_tesseract_crash_autorotate(spoof_tesseract_crash, resources, no_outpdf)
|
||||
resources / 'ccitt.pdf', no_outpdf, '-r', env=spoof_tesseract_crash
|
||||
)
|
||||
assert p.returncode == ExitCode.child_process_error
|
||||
assert not os.path.exists(no_outpdf)
|
||||
assert not no_outpdf.exists()
|
||||
assert "uncaught exception" in err
|
||||
print(out)
|
||||
print(err)
|
||||
|
||||
Reference in New Issue
Block a user