paradizelost/paperless-ngx
1
0
Fork 0
mirror of https://github.com/paperless-ngx/paperless-ngx.git synced 2025-09-01 01:46:16 +00:00
Code Issues Packages Projects Releases Wiki Activity

Chore: Switch from os.path to pathlib.Path (#8644)

Browse Source
This commit is contained in:
Sebastian Steinbeißer
2025-01-29 19:58:53 +01:00
committed by GitHub
parent 79956d6a7b
commit fce7b03324
5 changed files with 72 additions and 137 deletions
Download Patch File Download Diff File Expand all files Collapse all files

36
src/documents/consumer.py
View File

@@ -4,6 +4,7 @@ import os
import tempfile
from enum import Enum
from pathlib import Path
from typing import TYPE_CHECKING
import magic
from django.conf import settings
@@ -154,7 +155,11 @@ class ConsumerPlugin(
"""
Confirm the input file still exists where it should
"""
if not os.path.isfile(self.input_doc.original_file):
if TYPE_CHECKING:
assert isinstance(self.input_doc.original_file, Path), (
self.input_doc.original_file
)
if not self.input_doc.original_file.is_file():
self._fail(
ConsumerStatusShortMessage.FILE_NOT_FOUND,
f"Cannot consume {self.input_doc.original_file}: File not found.",
@@ -164,7 +169,7 @@ class ConsumerPlugin(
"""
Using the MD5 of the file, check this exact file doesn't already exist
"""
with open(self.input_doc.original_file, "rb") as f:
with Path(self.input_doc.original_file).open("rb") as f:
checksum = hashlib.md5(f.read()).hexdigest()
existing_doc = Document.global_objects.filter(
Q(checksum=checksum) | Q(archive_checksum=checksum),
@@ -178,7 +183,7 @@ class ConsumerPlugin(
log_msg += " Note: existing document is in the trash."
if settings.CONSUMER_DELETE_DUPLICATES:
os.unlink(self.input_doc.original_file)
Path(self.input_doc.original_file).unlink()
self._fail(
msg,
log_msg,
@@ -237,7 +242,7 @@ class ConsumerPlugin(
if not settings.PRE_CONSUME_SCRIPT:
return
if not os.path.isfile(settings.PRE_CONSUME_SCRIPT):
if not Path(settings.PRE_CONSUME_SCRIPT).is_file():
self._fail(
ConsumerStatusShortMessage.PRE_CONSUME_SCRIPT_NOT_FOUND,
f"Configured pre-consume script "
@@ -280,7 +285,7 @@ class ConsumerPlugin(
if not settings.POST_CONSUME_SCRIPT:
return
if not os.path.isfile(settings.POST_CONSUME_SCRIPT):
if not Path(settings.POST_CONSUME_SCRIPT).is_file():
self._fail(
ConsumerStatusShortMessage.POST_CONSUME_SCRIPT_NOT_FOUND,
f"Configured post-consume script "
@@ -582,7 +587,7 @@ class ConsumerPlugin(
document.thumbnail_path,
)
if archive_path and os.path.isfile(archive_path):
if archive_path and Path(archive_path).is_file():
document.archive_filename = generate_unique_filename(
document,
archive_filename=True,
@@ -594,7 +599,7 @@ class ConsumerPlugin(
document.archive_path,
)
with open(archive_path, "rb") as f:
with Path(archive_path).open("rb") as f:
document.archive_checksum = hashlib.md5(
f.read(),
).hexdigest()
@@ -612,14 +617,14 @@ class ConsumerPlugin(
self.unmodified_original.unlink()
# https://github.com/jonaswinkler/paperless-ng/discussions/1037
shadow_file = os.path.join(
os.path.dirname(self.input_doc.original_file),
"._" + os.path.basename(self.input_doc.original_file),
shadow_file = (
Path(self.input_doc.original_file).parent
/ f"._{Path(self.input_doc.original_file).name}"
)
if os.path.isfile(shadow_file):
if Path(shadow_file).is_file():
self.log.debug(f"Deleting file {shadow_file}")
os.unlink(shadow_file)
Path(shadow_file).unlink()
except Exception as e:
self._fail(
@@ -704,7 +709,7 @@ class ConsumerPlugin(
create_date = date
self.log.debug(f"Creation date from parse_date: {create_date}")
else:
stats = os.stat(self.input_doc.original_file)
stats = Path(self.input_doc.original_file).stat()
create_date = timezone.make_aware(
datetime.datetime.fromtimestamp(stats.st_mtime),
)
@@ -800,7 +805,10 @@ class ConsumerPlugin(
) # adds to document
def _write(self, storage_type, source, target):
with open(source, "rb") as read_file, open(target, "wb") as write_file:
with (
Path(source).open("rb") as read_file,
Path(target).open("wb") as write_file,
):
write_file.write(read_file.read())
# Attempt to copy file's original stats, but it's ok if we can't