mirror of
https://github.com/paperless-ngx/paperless-ngx.git
synced 2026-10-15 20:47:14 +00:00
stumpylog
cf3077def0
get_users_with_perms includes group members by default, so PDF actions such as merge and split gave them direct grants on the new documents, which outlived their group membership.
227 lines
7.4 KiB
Python
227 lines
7.4 KiB
Python
import dataclasses
|
|
import datetime
|
|
from enum import IntEnum
|
|
from pathlib import Path
|
|
from typing import TypedDict
|
|
|
|
import magic
|
|
from guardian.shortcuts import get_groups_with_perms
|
|
from guardian.shortcuts import get_users_with_perms
|
|
|
|
|
|
class StoredBarcode(TypedDict):
|
|
"""
|
|
A detected barcode as it is stored with a document
|
|
"""
|
|
|
|
page: int # 1-indexed
|
|
value: str
|
|
format: str # a DocumentBarcode.Format value
|
|
|
|
|
|
@dataclasses.dataclass
|
|
class DocumentMetadataOverrides:
|
|
"""
|
|
Manages overrides for document fields which normally would
|
|
be set from content or matching. All fields default to None,
|
|
meaning no override is happening
|
|
"""
|
|
|
|
filename: str | None = None
|
|
title: str | None = None
|
|
correspondent_id: int | None = None
|
|
document_type_id: int | None = None
|
|
tag_ids: list[int] | None = None
|
|
storage_path_id: int | None = None
|
|
created: datetime.date | None = None
|
|
asn: int | None = None
|
|
owner_id: int | None = None
|
|
view_users: list[int] | None = None
|
|
view_groups: list[int] | None = None
|
|
change_users: list[int] | None = None
|
|
change_groups: list[int] | None = None
|
|
custom_fields: dict | None = None
|
|
skip_asn_if_exists: bool = False
|
|
version_label: str | None = None
|
|
actor_id: int | None = None
|
|
remote_ocr: bool = False
|
|
barcodes: list[StoredBarcode] | None = None
|
|
|
|
def update(self, other: "DocumentMetadataOverrides") -> "DocumentMetadataOverrides":
|
|
"""
|
|
Merges two DocumentMetadataOverrides objects such that object B's overrides
|
|
are applied to object A or merged if multiple are accepted.
|
|
|
|
The update is an in-place modification of self
|
|
"""
|
|
# only if empty
|
|
if other.title is not None:
|
|
self.title = other.title
|
|
if other.correspondent_id is not None:
|
|
self.correspondent_id = other.correspondent_id
|
|
if other.document_type_id is not None:
|
|
self.document_type_id = other.document_type_id
|
|
if other.storage_path_id is not None:
|
|
self.storage_path_id = other.storage_path_id
|
|
if other.owner_id is not None:
|
|
self.owner_id = other.owner_id
|
|
if other.actor_id is not None:
|
|
self.actor_id = other.actor_id
|
|
if other.skip_asn_if_exists:
|
|
self.skip_asn_if_exists = True
|
|
if other.remote_ocr:
|
|
self.remote_ocr = True
|
|
if other.version_label is not None:
|
|
self.version_label = other.version_label
|
|
|
|
# merge
|
|
if self.tag_ids is None:
|
|
self.tag_ids = other.tag_ids
|
|
elif other.tag_ids is not None:
|
|
self.tag_ids.extend(other.tag_ids)
|
|
self.tag_ids = list(set(self.tag_ids))
|
|
|
|
if self.view_users is None:
|
|
self.view_users = other.view_users
|
|
elif other.view_users is not None:
|
|
self.view_users.extend(other.view_users)
|
|
self.view_users = list(set(self.view_users))
|
|
|
|
if self.view_groups is None:
|
|
self.view_groups = other.view_groups
|
|
elif other.view_groups is not None:
|
|
self.view_groups.extend(other.view_groups)
|
|
self.view_groups = list(set(self.view_groups))
|
|
|
|
if self.change_users is None:
|
|
self.change_users = other.change_users
|
|
elif other.change_users is not None:
|
|
self.change_users.extend(other.change_users)
|
|
self.change_users = list(set(self.change_users))
|
|
|
|
if self.change_groups is None:
|
|
self.change_groups = other.change_groups
|
|
elif other.change_groups is not None:
|
|
self.change_groups.extend(other.change_groups)
|
|
self.change_groups = list(set(self.change_groups))
|
|
|
|
if self.custom_fields is None:
|
|
self.custom_fields = other.custom_fields
|
|
elif other.custom_fields is not None:
|
|
self.custom_fields.update(other.custom_fields)
|
|
|
|
return self
|
|
|
|
@staticmethod
|
|
def from_document(doc) -> "DocumentMetadataOverrides":
|
|
"""
|
|
Fills in the overrides from a document object
|
|
"""
|
|
overrides = DocumentMetadataOverrides()
|
|
overrides.title = doc.title
|
|
overrides.correspondent_id = doc.correspondent.id if doc.correspondent else None
|
|
overrides.document_type_id = doc.document_type.id if doc.document_type else None
|
|
overrides.storage_path_id = doc.storage_path.id if doc.storage_path else None
|
|
overrides.owner_id = doc.owner.id if doc.owner else None
|
|
overrides.tag_ids = list(doc.tags.values_list("id", flat=True))
|
|
overrides.created = doc.created
|
|
|
|
overrides.view_users = list(
|
|
get_users_with_perms(
|
|
doc,
|
|
only_with_perms_in=["view_document"],
|
|
with_group_users=False,
|
|
).values_list("id", flat=True),
|
|
)
|
|
overrides.change_users = list(
|
|
get_users_with_perms(
|
|
doc,
|
|
only_with_perms_in=["change_document"],
|
|
with_group_users=False,
|
|
).values_list("id", flat=True),
|
|
)
|
|
overrides.custom_fields = {
|
|
custom_field.field.id: custom_field.value
|
|
for custom_field in doc.custom_fields.all()
|
|
}
|
|
|
|
groups_with_perms = get_groups_with_perms(
|
|
doc,
|
|
attach_perms=True,
|
|
)
|
|
overrides.view_groups = [
|
|
group.id
|
|
for group in groups_with_perms
|
|
if "view_document" in groups_with_perms[group]
|
|
]
|
|
overrides.change_groups = [
|
|
group.id
|
|
for group in groups_with_perms
|
|
if "change_document" in groups_with_perms[group]
|
|
]
|
|
|
|
return overrides
|
|
|
|
|
|
class DocumentSource(IntEnum):
|
|
"""
|
|
The source of an incoming document. May have other uses in the future
|
|
"""
|
|
|
|
ConsumeFolder = 1
|
|
ApiUpload = 2
|
|
MailFetch = 3
|
|
WebUI = 4
|
|
|
|
|
|
@dataclasses.dataclass
|
|
class ConsumableDocument:
|
|
"""
|
|
Encapsulates an incoming document, either from consume folder, API upload
|
|
or mail fetching and certain useful operations on it.
|
|
"""
|
|
|
|
source: DocumentSource
|
|
original_file: Path
|
|
root_document_id: int | None = None
|
|
original_path: Path | None = None
|
|
mailrule_id: int | None = None
|
|
mime_type: str = dataclasses.field(init=False, default=None)
|
|
|
|
def __post_init__(self) -> None:
|
|
"""
|
|
After a dataclass is initialized, this is called to finalize some data
|
|
1. Make sure the original path is an absolute, fully qualified path
|
|
2. Get the mime type of the file
|
|
"""
|
|
# Always fully qualify the path first thing
|
|
# Just in case, convert to a path if it's a str
|
|
self.original_file = Path(self.original_file).resolve()
|
|
|
|
# Get the file type once at init
|
|
# Note this function isn't called when the object is unpickled
|
|
self.mime_type = magic.from_file(self.original_file, mime=True)
|
|
|
|
|
|
class ConsumeFileDuplicateResult(TypedDict):
|
|
"""Returned by consume_file when the file is rejected as a duplicate."""
|
|
|
|
duplicate_of: int
|
|
duplicate_in_trash: bool
|
|
|
|
|
|
class ConsumeFileSuccessResult(TypedDict):
|
|
"""Returned by consume_file when the document is created successfully."""
|
|
|
|
document_id: int
|
|
|
|
|
|
class ConsumeFileStoppedResult(TypedDict):
|
|
"""Returned by consume_file when a plugin raises StopConsumeTaskError.
|
|
|
|
Examples: barcode split dispatched child tasks, double-sided scan waiting
|
|
for the second half, workflow deleted the document during consumption.
|
|
"""
|
|
|
|
reason: str
|