mirror of
https://github.com/paperless-ngx/paperless-ngx.git
synced 2025-09-01 01:46:16 +00:00
merge bulk download
This commit is contained in:
@@ -1,107 +0,0 @@
|
||||
from __future__ import annotations
|
||||
|
||||
from pathlib import Path
|
||||
from typing import TYPE_CHECKING
|
||||
from typing import NoReturn
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from collections.abc import Callable
|
||||
from zipfile import ZipFile
|
||||
|
||||
from paperless.models import Document
|
||||
|
||||
|
||||
class BulkArchiveStrategy:
|
||||
def __init__(self, zipf: ZipFile, *, follow_formatting: bool = False) -> None:
|
||||
self.zipf: ZipFile = zipf
|
||||
if follow_formatting:
|
||||
self.make_unique_filename: Callable[..., Path | str] = (
|
||||
self._formatted_filepath
|
||||
)
|
||||
else:
|
||||
self.make_unique_filename = self._filename_only
|
||||
|
||||
def _filename_only(
|
||||
self,
|
||||
doc: Document,
|
||||
*,
|
||||
archive: bool = False,
|
||||
folder: str = "",
|
||||
) -> str:
|
||||
"""
|
||||
Constructs a unique name for the given document to be used inside the
|
||||
zip file.
|
||||
|
||||
The filename might not be unique enough, so a counter is appended if needed
|
||||
"""
|
||||
counter = 0
|
||||
while True:
|
||||
filename: str = folder + doc.get_public_filename(
|
||||
archive=archive,
|
||||
counter=counter,
|
||||
)
|
||||
if filename in self.zipf.namelist():
|
||||
counter += 1
|
||||
else:
|
||||
return filename
|
||||
|
||||
def _formatted_filepath(
|
||||
self,
|
||||
doc: Document,
|
||||
*,
|
||||
archive: bool = False,
|
||||
folder: str = "",
|
||||
) -> Path:
|
||||
"""
|
||||
Constructs a full file path for the given document to be used inside
|
||||
the zipfile.
|
||||
|
||||
The path is already unique, as handled when a document is consumed or updated
|
||||
"""
|
||||
if archive and doc.has_archive_version:
|
||||
if TYPE_CHECKING:
|
||||
assert doc.archive_filename is not None
|
||||
in_archive_path: Path = Path(folder) / doc.archive_filename
|
||||
else:
|
||||
if TYPE_CHECKING:
|
||||
assert doc.filename is not None
|
||||
in_archive_path = Path(folder) / doc.filename
|
||||
|
||||
return in_archive_path
|
||||
|
||||
def add_document(self, doc: Document) -> NoReturn:
|
||||
raise NotImplementedError # pragma: no cover
|
||||
|
||||
|
||||
class OriginalsOnlyStrategy(BulkArchiveStrategy):
|
||||
def add_document(self, doc: Document) -> None:
|
||||
self.zipf.write(doc.source_path, self.make_unique_filename(doc))
|
||||
|
||||
|
||||
class ArchiveOnlyStrategy(BulkArchiveStrategy):
|
||||
def add_document(self, doc: Document) -> None:
|
||||
if doc.has_archive_version:
|
||||
if TYPE_CHECKING:
|
||||
assert doc.archive_path is not None
|
||||
self.zipf.write(
|
||||
doc.archive_path,
|
||||
self.make_unique_filename(doc, archive=True),
|
||||
)
|
||||
else:
|
||||
self.zipf.write(doc.source_path, self.make_unique_filename(doc))
|
||||
|
||||
|
||||
class OriginalAndArchiveStrategy(BulkArchiveStrategy):
|
||||
def add_document(self, doc: Document) -> None:
|
||||
if doc.has_archive_version:
|
||||
if TYPE_CHECKING:
|
||||
assert doc.archive_path is not None
|
||||
self.zipf.write(
|
||||
doc.archive_path,
|
||||
self.make_unique_filename(doc, archive=True, folder="archive/"),
|
||||
)
|
||||
|
||||
self.zipf.write(
|
||||
doc.source_path,
|
||||
self.make_unique_filename(doc, folder="originals/"),
|
||||
)
|
Reference in New Issue
Block a user