Merge utils

src/documents/barcodes.py

@@ -17,11 +17,11 @@ from documents.converters import convert_from_tiff_to_pdf
 from documents.plugins.base import ConsumeTaskPlugin
 from documents.plugins.base import StopConsumeTaskError
 from documents.plugins.helpers import ProgressStatusOptions
-from documents.utils import copy_basic_file_stats
-from documents.utils import copy_file_with_basic_stats
-from documents.utils import maybe_override_pixel_limit
 from paperless.data_models import ConsumableDocument
 from paperless.models import Tag
+from paperless.utils import copy_basic_file_stats
+from paperless.utils import copy_file_with_basic_stats
+from paperless.utils import maybe_override_pixel_limit
 
 if TYPE_CHECKING:
     from collections.abc import Callable

src/documents/converters.py

@@ -4,9 +4,9 @@ import img2pdf
 from django.conf import settings
 from PIL import Image
 
-from documents.utils import copy_basic_file_stats
-from documents.utils import maybe_override_pixel_limit
-from documents.utils import run_subprocess
+from paperless.utils import copy_basic_file_stats
+from paperless.utils import maybe_override_pixel_limit
+from paperless.utils import run_subprocess
 
 
 def convert_from_tiff_to_pdf(tiff_path: Path, target_directory: Path) -> Path:

src/documents/management/commands/document_exporter.py

@@ -38,7 +38,6 @@ from documents.management.commands.mixins import CryptMixin
 from documents.settings import EXPORTER_ARCHIVE_NAME
 from documents.settings import EXPORTER_FILE_NAME
 from documents.settings import EXPORTER_THUMBNAIL_NAME
-from documents.utils import copy_file_with_basic_stats
 from paperless import version
 from paperless.db import GnuPG
 from paperless.models import ApplicationConfiguration
@@ -58,6 +57,7 @@ from paperless.models import WorkflowAction
 from paperless.models import WorkflowActionEmail
 from paperless.models import WorkflowActionWebhook
 from paperless.models import WorkflowTrigger
+from paperless.utils import copy_file_with_basic_stats
 from paperless_mail.models import MailAccount
 from paperless_mail.models import MailRule
 

src/documents/management/commands/document_importer.py

@@ -30,7 +30,6 @@ from documents.settings import EXPORTER_FILE_NAME
 from documents.settings import EXPORTER_THUMBNAIL_NAME
 from documents.signals.handlers import check_paths_and_prune_custom_fields
 from documents.signals.handlers import update_filename_and_move_files
-from documents.utils import copy_file_with_basic_stats
 from paperless import version
 from paperless.models import Correspondent
 from paperless.models import CustomField
@@ -39,6 +38,7 @@ from paperless.models import Document
 from paperless.models import DocumentType
 from paperless.models import Note
 from paperless.models import Tag
+from paperless.utils import copy_file_with_basic_stats
 
 if settings.AUDIT_LOG_ENABLED:
     from auditlog.registry import auditlog

src/documents/parsers.py

@@ -17,8 +17,8 @@ from django.utils import timezone
 
 from documents.loggers import LoggingMixin
 from documents.signals import document_consumer_declaration
-from documents.utils import copy_file_with_basic_stats
-from documents.utils import run_subprocess
+from paperless.utils import copy_file_with_basic_stats
+from paperless.utils import run_subprocess
 
 if TYPE_CHECKING:
     import datetime

src/documents/utils.py

@@ -1,110 +0,0 @@
-import logging
-import shutil
-from os import utime
-from pathlib import Path
-from subprocess import CompletedProcess
-from subprocess import run
-
-from django.conf import settings
-from PIL import Image
-
-
-def _coerce_to_path(
-    source: Path | str,
-    dest: Path | str,
-) -> tuple[Path, Path]:
-    return Path(source).resolve(), Path(dest).resolve()
-
-
-def copy_basic_file_stats(source: Path | str, dest: Path | str) -> None:
-    """
-    Copies only the m_time and a_time attributes from source to destination.
-    Both are expected to exist.
-
-    The extended attribute copy does weird things with SELinux and files
-    copied from temporary directories and copystat doesn't allow disabling
-    these copies
-    """
-    source, dest = _coerce_to_path(source, dest)
-    src_stat = source.stat()
-    utime(dest, ns=(src_stat.st_atime_ns, src_stat.st_mtime_ns))
-
-
-def copy_file_with_basic_stats(
-    source: Path | str,
-    dest: Path | str,
-) -> None:
-    """
-    A sort of simpler copy2 that doesn't copy extended file attributes,
-    only the access time and modified times from source to dest.
-
-    The extended attribute copy does weird things with SELinux and files
-    copied from temporary directories.
-    """
-    source, dest = _coerce_to_path(source, dest)
-
-    shutil.copy(source, dest)
-    copy_basic_file_stats(source, dest)
-
-
-def maybe_override_pixel_limit() -> None:
-    """
-    Maybe overrides the PIL limit on pixel count, if configured to allow it
-    """
-    limit: float | int | None = settings.MAX_IMAGE_PIXELS
-    if limit is not None and limit >= 0:
-        pixel_count = limit
-        if pixel_count == 0:
-            pixel_count = None
-        Image.MAX_IMAGE_PIXELS = pixel_count
-
-
-def run_subprocess(
-    arguments: list[str],
-    env: dict[str, str] | None = None,
-    logger: logging.Logger | None = None,
-    *,
-    check_exit_code: bool = True,
-    log_stdout: bool = True,
-    log_stderr: bool = True,
-) -> CompletedProcess:
-    """
-    Runs a subprocess and logs its output, checking return code if requested
-    """
-
-    proc_name = arguments[0]
-
-    completed_proc = run(args=arguments, env=env, capture_output=True, check=False)
-
-    if logger:
-        logger.info(f"{proc_name} exited {completed_proc.returncode}")
-
-    if log_stdout and logger and completed_proc.stdout:
-        stdout_str = (
-            completed_proc.stdout.decode("utf8", errors="ignore")
-            .strip()
-            .split(
-                "\n",
-            )
-        )
-        logger.info(f"{proc_name} stdout:")
-        for line in stdout_str:
-            logger.info(line)
-
-    if log_stderr and logger and completed_proc.stderr:
-        stderr_str = (
-            completed_proc.stderr.decode("utf8", errors="ignore")
-            .strip()
-            .split(
-                "\n",
-            )
-        )
-        logger.info(f"{proc_name} stderr:")
-        for line in stderr_str:
-            logger.warning(line)
-
-    # Last, if requested, after logging outputs
-    if check_exit_code:
-        completed_proc.check_returncode()
-
-    return completed_proc