from django.contrib import admin
from guardian.admin import GuardedModelAdmin
-from documents.models import Correspondent
-from documents.models import CustomField
-from documents.models import CustomFieldInstance
-from documents.models import Document
-from documents.models import DocumentType
-from documents.models import Note
-from documents.models import PaperlessTask
-from documents.models import SavedView
-from documents.models import SavedViewFilterRule
-from documents.models import ShareLink
-from documents.models import StoragePath
-from documents.models import Tag
+from paperless.models import Correspondent
+from paperless.models import CustomField
+from paperless.models import CustomFieldInstance
+from paperless.models import Document
+from paperless.models import DocumentType
+from paperless.models import Note
+from paperless.models import PaperlessTask
+from paperless.models import SavedView
+from paperless.models import SavedViewFilterRule
+from paperless.models import ShareLink
+from paperless.models import StoragePath
+from paperless.models import Tag
if settings.AUDIT_LOG_ENABLED:
from auditlog.admin import LogEntryAdmin
from documents.converters import convert_from_tiff_to_pdf
from documents.data_models import ConsumableDocument
-from documents.models import Tag
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.models import Tag
if TYPE_CHECKING:
from collections.abc import Callable
from collections.abc import Callable
from zipfile import ZipFile
- from documents.models import Document
+ from paperless.models import Document
class BulkArchiveStrategy:
from documents.data_models import ConsumableDocument
from documents.data_models import DocumentMetadataOverrides
from documents.data_models import DocumentSource
-from documents.models import Correspondent
-from documents.models import CustomField
-from documents.models import CustomFieldInstance
-from documents.models import Document
-from documents.models import DocumentType
-from documents.models import StoragePath
from documents.permissions import set_permissions_for_object
from documents.plugins.helpers import DocumentsStatusManager
from documents.tasks import bulk_update_documents
from documents.tasks import consume_file
from documents.tasks import update_document_content_maybe_archive_file
+from paperless.models import Correspondent
+from paperless.models import CustomField
+from paperless.models import CustomFieldInstance
+from paperless.models import Document
+from paperless.models import DocumentType
+from paperless.models import StoragePath
if TYPE_CHECKING:
from django.contrib.auth.models import User
from django.core.cache import cache
-from documents.models import Document
+from paperless.models import Document
if TYPE_CHECKING:
from documents.classifier import DocumentClassifier
@register()
def changed_password_check(app_configs, **kwargs):
- from documents.models import Document
from paperless.db import GnuPG
+ from paperless.models import Document
try:
encrypted_doc = (
from documents.caching import CLASSIFIER_HASH_KEY
from documents.caching import CLASSIFIER_MODIFIED_KEY
from documents.caching import CLASSIFIER_VERSION_KEY
-from documents.models import Document
-from documents.models import MatchingModel
+from paperless.models import Document
+from paperless.models import MatchingModel
logger = logging.getLogger("paperless.classifier")
from documents.caching import CLASSIFIER_VERSION_KEY
from documents.caching import get_thumbnail_modified_key
from documents.classifier import DocumentClassifier
-from documents.models import Document
+from paperless.models import Document
def suggestions_etag(request, pk: int) -> str | None:
from documents.file_handling import create_source_path_directory
from documents.file_handling import generate_unique_filename
from documents.loggers import LoggingMixin
-from documents.models import Correspondent
-from documents.models import CustomField
-from documents.models import CustomFieldInstance
-from documents.models import Document
-from documents.models import DocumentType
-from documents.models import StoragePath
-from documents.models import Tag
-from documents.models import WorkflowTrigger
from documents.parsers import DocumentParser
from documents.parsers import ParseError
from documents.parsers import get_parser_class_for_mime_type
from documents.utils import copy_basic_file_stats
from documents.utils import copy_file_with_basic_stats
from documents.utils import run_subprocess
+from paperless.models import Correspondent
+from paperless.models import CustomField
+from paperless.models import CustomFieldInstance
+from paperless.models import Document
+from paperless.models import DocumentType
+from paperless.models import StoragePath
+from paperless.models import Tag
+from paperless.models import WorkflowTrigger
from paperless_mail.parsers import MailDocumentParser
from django.conf import settings as django_settings
from django.contrib.auth.models import User
-from documents.models import Document
from paperless.config import GeneralConfig
+from paperless.models import Document
def settings(request):
from django.conf import settings
-from documents.models import Document
from documents.templating.filepath import validate_filepath_template_and_render
from documents.templating.utils import convert_format_str_to_template_format
+from paperless.models import Document
def create_source_path_directory(source_path):
from rest_framework.filters import OrderingFilter
from rest_framework_guardian.filters import ObjectPermissionsFilter
-from documents.models import Correspondent
-from documents.models import CustomField
-from documents.models import CustomFieldInstance
-from documents.models import Document
-from documents.models import DocumentType
-from documents.models import PaperlessTask
-from documents.models import ShareLink
-from documents.models import StoragePath
-from documents.models import Tag
+from paperless.models import Correspondent
+from paperless.models import CustomField
+from paperless.models import CustomFieldInstance
+from paperless.models import Document
+from paperless.models import DocumentType
+from paperless.models import PaperlessTask
+from paperless.models import ShareLink
+from paperless.models import StoragePath
+from paperless.models import Tag
if TYPE_CHECKING:
from collections.abc import Callable
from whoosh.util.times import timespan
from whoosh.writing import AsyncWriter
-from documents.models import CustomFieldInstance
-from documents.models import Document
-from documents.models import Note
-from documents.models import User
+from paperless.models import CustomFieldInstance
+from paperless.models import Document
+from paperless.models import Note
+from paperless.models import User
if TYPE_CHECKING:
from django.db.models import QuerySet
from django.db import connection
from django.db import models
-from documents.models import Document
+from paperless.models import Document
class Command(BaseCommand):
from django.core.management.base import BaseCommand
from django.core.management.base import CommandError
-from documents.models import Document
from paperless.db import GnuPG
+from paperless.models import Document
class Command(BaseCommand):
from documents.management.commands.mixins import MultiProcessMixin
from documents.management.commands.mixins import ProgressBarMixin
-from documents.models import Document
from documents.tasks import update_document_content_maybe_archive_file
+from paperless.models import Document
logger = logging.getLogger("paperless.management.archiver")
from documents.data_models import ConsumableDocument
from documents.data_models import DocumentMetadataOverrides
from documents.data_models import DocumentSource
-from documents.models import Tag
from documents.parsers import is_file_ext_supported
from documents.tasks import consume_file
+from paperless.models import Tag
try:
from inotifyrecursive import INotify
from documents.file_handling import delete_empty_directories
from documents.file_handling import generate_filename
from documents.management.commands.mixins import CryptMixin
-from documents.models import Correspondent
-from documents.models import CustomField
-from documents.models import CustomFieldInstance
-from documents.models import Document
-from documents.models import DocumentType
-from documents.models import Note
-from documents.models import SavedView
-from documents.models import SavedViewFilterRule
-from documents.models import StoragePath
-from documents.models import Tag
-from documents.models import UiSettings
-from documents.models import Workflow
-from documents.models import WorkflowAction
-from documents.models import WorkflowActionEmail
-from documents.models import WorkflowActionWebhook
-from documents.models import WorkflowTrigger
from documents.settings import EXPORTER_ARCHIVE_NAME
from documents.settings import EXPORTER_FILE_NAME
from documents.settings import EXPORTER_THUMBNAIL_NAME
from paperless import version
from paperless.db import GnuPG
from paperless.models import ApplicationConfiguration
+from paperless.models import Correspondent
+from paperless.models import CustomField
+from paperless.models import CustomFieldInstance
+from paperless.models import Document
+from paperless.models import DocumentType
+from paperless.models import Note
+from paperless.models import SavedView
+from paperless.models import SavedViewFilterRule
+from paperless.models import StoragePath
+from paperless.models import Tag
+from paperless.models import UiSettings
+from paperless.models import Workflow
+from paperless.models import WorkflowAction
+from paperless.models import WorkflowActionEmail
+from paperless.models import WorkflowActionWebhook
+from paperless.models import WorkflowTrigger
from paperless_mail.models import MailAccount
from paperless_mail.models import MailRule
from documents.management.commands.mixins import MultiProcessMixin
from documents.management.commands.mixins import ProgressBarMixin
-from documents.models import Document
+from paperless.models import Document
@dataclasses.dataclass(frozen=True)
from documents.file_handling import create_source_path_directory
from documents.management.commands.mixins import CryptMixin
-from documents.models import Correspondent
-from documents.models import CustomField
-from documents.models import CustomFieldInstance
-from documents.models import Document
-from documents.models import DocumentType
-from documents.models import Note
-from documents.models import Tag
from documents.parsers import run_convert
from documents.settings import EXPORTER_ARCHIVE_NAME
from documents.settings import EXPORTER_CRYPTO_SETTINGS_NAME
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
+from paperless.models import CustomFieldInstance
+from paperless.models import Document
+from paperless.models import DocumentType
+from paperless.models import Note
+from paperless.models import Tag
if settings.AUDIT_LOG_ENABLED:
from auditlog.registry import auditlog
from django.db.models.signals import post_save
from documents.management.commands.mixins import ProgressBarMixin
-from documents.models import Document
+from paperless.models import Document
class Command(ProgressBarMixin, BaseCommand):
from documents.classifier import load_classifier
from documents.management.commands.mixins import ProgressBarMixin
-from documents.models import Document
from documents.signals.handlers import set_correspondent
from documents.signals.handlers import set_document_type
from documents.signals.handlers import set_storage_path
from documents.signals.handlers import set_tags
+from paperless.models import Document
logger = logging.getLogger("paperless.management.retagger")
from documents.management.commands.mixins import MultiProcessMixin
from documents.management.commands.mixins import ProgressBarMixin
-from documents.models import Document
from documents.parsers import get_parser_class_for_mime_type
+from paperless.models import Document
def _process_document(doc_id):
from documents.data_models import ConsumableDocument
from documents.data_models import DocumentSource
-from documents.models import Correspondent
-from documents.models import Document
-from documents.models import DocumentType
-from documents.models import MatchingModel
-from documents.models import StoragePath
-from documents.models import Tag
-from documents.models import Workflow
-from documents.models import WorkflowTrigger
from documents.permissions import get_objects_for_user_owner_aware
+from paperless.models import Correspondent
+from paperless.models import Document
+from paperless.models import DocumentType
+from paperless.models import MatchingModel
+from paperless.models import StoragePath
+from paperless.models import Tag
+from paperless.models import Workflow
+from paperless.models import WorkflowTrigger
if TYPE_CHECKING:
from documents.classifier import DocumentClassifier
+++ /dev/null
-import datetime
-from pathlib import Path
-from typing import Final
-
-import pathvalidate
-from celery import states
-from django.conf import settings
-from django.contrib.auth.models import Group
-from django.contrib.auth.models import User
-from django.core.validators import MaxValueValidator
-from django.core.validators import MinValueValidator
-from django.db import models
-from django.utils import timezone
-from django.utils.translation import gettext_lazy as _
-from multiselectfield import MultiSelectField
-
-if settings.AUDIT_LOG_ENABLED:
- from auditlog.registry import auditlog
-
-from django.db.models import Case
-from django.db.models.functions import Cast
-from django.db.models.functions import Substr
-from django_softdelete.models import SoftDeleteModel
-
-from documents.data_models import DocumentSource
-from documents.parsers import get_default_file_extension
-
-
-class ModelWithOwner(models.Model):
- owner = models.ForeignKey(
- User,
- blank=True,
- null=True,
- default=None,
- on_delete=models.SET_NULL,
- verbose_name=_("owner"),
- )
-
- class Meta:
- abstract = True
-
-
-class MatchingModel(ModelWithOwner):
- MATCH_NONE = 0
- MATCH_ANY = 1
- MATCH_ALL = 2
- MATCH_LITERAL = 3
- MATCH_REGEX = 4
- MATCH_FUZZY = 5
- MATCH_AUTO = 6
-
- MATCHING_ALGORITHMS = (
- (MATCH_NONE, _("None")),
- (MATCH_ANY, _("Any word")),
- (MATCH_ALL, _("All words")),
- (MATCH_LITERAL, _("Exact match")),
- (MATCH_REGEX, _("Regular expression")),
- (MATCH_FUZZY, _("Fuzzy word")),
- (MATCH_AUTO, _("Automatic")),
- )
-
- name = models.CharField(_("name"), max_length=128)
-
- match = models.CharField(_("match"), max_length=256, blank=True)
-
- matching_algorithm = models.PositiveIntegerField(
- _("matching algorithm"),
- choices=MATCHING_ALGORITHMS,
- default=MATCH_ANY,
- )
-
- is_insensitive = models.BooleanField(_("is insensitive"), default=True)
-
- class Meta:
- abstract = True
- ordering = ("name",)
- constraints = [
- models.UniqueConstraint(
- fields=["name", "owner"],
- name="%(app_label)s_%(class)s_unique_name_owner",
- ),
- models.UniqueConstraint(
- name="%(app_label)s_%(class)s_name_uniq",
- fields=["name"],
- condition=models.Q(owner__isnull=True),
- ),
- ]
-
- def __str__(self):
- return self.name
-
-
-class Correspondent(MatchingModel):
- class Meta(MatchingModel.Meta):
- verbose_name = _("correspondent")
- verbose_name_plural = _("correspondents")
-
-
-class Tag(MatchingModel):
- color = models.CharField(_("color"), max_length=7, default="#a6cee3")
-
- is_inbox_tag = models.BooleanField(
- _("is inbox tag"),
- default=False,
- help_text=_(
- "Marks this tag as an inbox tag: All newly consumed "
- "documents will be tagged with inbox tags.",
- ),
- )
-
- class Meta(MatchingModel.Meta):
- verbose_name = _("tag")
- verbose_name_plural = _("tags")
-
-
-class DocumentType(MatchingModel):
- class Meta(MatchingModel.Meta):
- verbose_name = _("document type")
- verbose_name_plural = _("document types")
-
-
-class StoragePath(MatchingModel):
- path = models.TextField(
- _("path"),
- )
-
- class Meta(MatchingModel.Meta):
- verbose_name = _("storage path")
- verbose_name_plural = _("storage paths")
-
-
-class Document(SoftDeleteModel, ModelWithOwner):
- STORAGE_TYPE_UNENCRYPTED = "unencrypted"
- STORAGE_TYPE_GPG = "gpg"
- STORAGE_TYPES = (
- (STORAGE_TYPE_UNENCRYPTED, _("Unencrypted")),
- (STORAGE_TYPE_GPG, _("Encrypted with GNU Privacy Guard")),
- )
-
- correspondent = models.ForeignKey(
- Correspondent,
- blank=True,
- null=True,
- related_name="documents",
- on_delete=models.SET_NULL,
- verbose_name=_("correspondent"),
- )
-
- storage_path = models.ForeignKey(
- StoragePath,
- blank=True,
- null=True,
- related_name="documents",
- on_delete=models.SET_NULL,
- verbose_name=_("storage path"),
- )
-
- title = models.CharField(_("title"), max_length=128, blank=True, db_index=True)
-
- document_type = models.ForeignKey(
- DocumentType,
- blank=True,
- null=True,
- related_name="documents",
- on_delete=models.SET_NULL,
- verbose_name=_("document type"),
- )
-
- content = models.TextField(
- _("content"),
- blank=True,
- help_text=_(
- "The raw, text-only data of the document. This field is "
- "primarily used for searching.",
- ),
- )
-
- mime_type = models.CharField(_("mime type"), max_length=256, editable=False)
-
- tags = models.ManyToManyField(
- Tag,
- related_name="documents",
- blank=True,
- verbose_name=_("tags"),
- )
-
- checksum = models.CharField(
- _("checksum"),
- max_length=32,
- editable=False,
- unique=True,
- help_text=_("The checksum of the original document."),
- )
-
- archive_checksum = models.CharField(
- _("archive checksum"),
- max_length=32,
- editable=False,
- blank=True,
- null=True,
- help_text=_("The checksum of the archived document."),
- )
-
- page_count = models.PositiveIntegerField(
- _("page count"),
- blank=False,
- null=True,
- unique=False,
- db_index=False,
- validators=[MinValueValidator(1)],
- help_text=_(
- "The number of pages of the document.",
- ),
- )
-
- created = models.DateTimeField(_("created"), default=timezone.now, db_index=True)
-
- modified = models.DateTimeField(
- _("modified"),
- auto_now=True,
- editable=False,
- db_index=True,
- )
-
- storage_type = models.CharField(
- _("storage type"),
- max_length=11,
- choices=STORAGE_TYPES,
- default=STORAGE_TYPE_UNENCRYPTED,
- editable=False,
- )
-
- added = models.DateTimeField(
- _("added"),
- default=timezone.now,
- editable=False,
- db_index=True,
- )
-
- filename = models.FilePathField(
- _("filename"),
- max_length=1024,
- editable=False,
- default=None,
- unique=True,
- null=True,
- help_text=_("Current filename in storage"),
- )
-
- archive_filename = models.FilePathField(
- _("archive filename"),
- max_length=1024,
- editable=False,
- default=None,
- unique=True,
- null=True,
- help_text=_("Current archive filename in storage"),
- )
-
- original_filename = models.CharField(
- _("original filename"),
- max_length=1024,
- editable=False,
- default=None,
- unique=False,
- null=True,
- help_text=_("The original name of the file when it was uploaded"),
- )
-
- ARCHIVE_SERIAL_NUMBER_MIN: Final[int] = 0
- ARCHIVE_SERIAL_NUMBER_MAX: Final[int] = 0xFF_FF_FF_FF
-
- archive_serial_number = models.PositiveIntegerField(
- _("archive serial number"),
- blank=True,
- null=True,
- unique=True,
- db_index=True,
- validators=[
- MaxValueValidator(ARCHIVE_SERIAL_NUMBER_MAX),
- MinValueValidator(ARCHIVE_SERIAL_NUMBER_MIN),
- ],
- help_text=_(
- "The position of this document in your physical document archive.",
- ),
- )
-
- class Meta:
- ordering = ("-created",)
- verbose_name = _("document")
- verbose_name_plural = _("documents")
-
- def __str__(self) -> str:
- # Convert UTC database time to local time
- created = datetime.date.isoformat(timezone.localdate(self.created))
-
- res = f"{created}"
-
- if self.correspondent:
- res += f" {self.correspondent}"
- if self.title:
- res += f" {self.title}"
- return res
-
- @property
- def source_path(self) -> Path:
- if self.filename:
- fname = str(self.filename)
- else:
- fname = f"{self.pk:07}{self.file_type}"
- if self.storage_type == self.STORAGE_TYPE_GPG:
- fname += ".gpg" # pragma: no cover
-
- return (settings.ORIGINALS_DIR / Path(fname)).resolve()
-
- @property
- def source_file(self):
- return Path(self.source_path).open("rb")
-
- @property
- def has_archive_version(self) -> bool:
- return self.archive_filename is not None
-
- @property
- def archive_path(self) -> Path | None:
- if self.has_archive_version:
- return (settings.ARCHIVE_DIR / Path(str(self.archive_filename))).resolve()
- else:
- return None
-
- @property
- def archive_file(self):
- return Path(self.archive_path).open("rb")
-
- def get_public_filename(self, *, archive=False, counter=0, suffix=None) -> str:
- """
- Returns a sanitized filename for the document, not including any paths.
- """
- result = str(self)
-
- if counter:
- result += f"_{counter:02}"
-
- if suffix:
- result += suffix
-
- if archive:
- result += ".pdf"
- else:
- result += self.file_type
-
- return pathvalidate.sanitize_filename(result, replacement_text="-")
-
- @property
- def file_type(self):
- return get_default_file_extension(self.mime_type)
-
- @property
- def thumbnail_path(self) -> Path:
- webp_file_name = f"{self.pk:07}.webp"
- if self.storage_type == self.STORAGE_TYPE_GPG:
- webp_file_name += ".gpg"
-
- webp_file_path = settings.THUMBNAIL_DIR / Path(webp_file_name)
-
- return webp_file_path.resolve()
-
- @property
- def thumbnail_file(self):
- return Path(self.thumbnail_path).open("rb")
-
- @property
- def created_date(self):
- return timezone.localdate(self.created)
-
-
-class SavedView(ModelWithOwner):
- class DisplayMode(models.TextChoices):
- TABLE = ("table", _("Table"))
- SMALL_CARDS = ("smallCards", _("Small Cards"))
- LARGE_CARDS = ("largeCards", _("Large Cards"))
-
- class DisplayFields(models.TextChoices):
- TITLE = ("title", _("Title"))
- CREATED = ("created", _("Created"))
- ADDED = ("added", _("Added"))
- TAGS = ("tag"), _("Tags")
- CORRESPONDENT = ("correspondent", _("Correspondent"))
- DOCUMENT_TYPE = ("documenttype", _("Document Type"))
- STORAGE_PATH = ("storagepath", _("Storage Path"))
- NOTES = ("note", _("Note"))
- OWNER = ("owner", _("Owner"))
- SHARED = ("shared", _("Shared"))
- ASN = ("asn", _("ASN"))
- PAGE_COUNT = ("pagecount", _("Pages"))
- CUSTOM_FIELD = ("custom_field_%d", ("Custom Field"))
-
- name = models.CharField(_("name"), max_length=128)
-
- show_on_dashboard = models.BooleanField(
- _("show on dashboard"),
- )
- show_in_sidebar = models.BooleanField(
- _("show in sidebar"),
- )
-
- sort_field = models.CharField(
- _("sort field"),
- max_length=128,
- null=True,
- blank=True,
- )
- sort_reverse = models.BooleanField(_("sort reverse"), default=False)
-
- page_size = models.PositiveIntegerField(
- _("View page size"),
- null=True,
- blank=True,
- validators=[MinValueValidator(1)],
- )
-
- display_mode = models.CharField(
- max_length=128,
- verbose_name=_("View display mode"),
- choices=DisplayMode.choices,
- null=True,
- blank=True,
- )
-
- display_fields = models.JSONField(
- verbose_name=_("Document display fields"),
- null=True,
- blank=True,
- )
-
- class Meta:
- ordering = ("name",)
- verbose_name = _("saved view")
- verbose_name_plural = _("saved views")
-
- def __str__(self):
- return f"SavedView {self.name}"
-
-
-class SavedViewFilterRule(models.Model):
- RULE_TYPES = [
- (0, _("title contains")),
- (1, _("content contains")),
- (2, _("ASN is")),
- (3, _("correspondent is")),
- (4, _("document type is")),
- (5, _("is in inbox")),
- (6, _("has tag")),
- (7, _("has any tag")),
- (8, _("created before")),
- (9, _("created after")),
- (10, _("created year is")),
- (11, _("created month is")),
- (12, _("created day is")),
- (13, _("added before")),
- (14, _("added after")),
- (15, _("modified before")),
- (16, _("modified after")),
- (17, _("does not have tag")),
- (18, _("does not have ASN")),
- (19, _("title or content contains")),
- (20, _("fulltext query")),
- (21, _("more like this")),
- (22, _("has tags in")),
- (23, _("ASN greater than")),
- (24, _("ASN less than")),
- (25, _("storage path is")),
- (26, _("has correspondent in")),
- (27, _("does not have correspondent in")),
- (28, _("has document type in")),
- (29, _("does not have document type in")),
- (30, _("has storage path in")),
- (31, _("does not have storage path in")),
- (32, _("owner is")),
- (33, _("has owner in")),
- (34, _("does not have owner")),
- (35, _("does not have owner in")),
- (36, _("has custom field value")),
- (37, _("is shared by me")),
- (38, _("has custom fields")),
- (39, _("has custom field in")),
- (40, _("does not have custom field in")),
- (41, _("does not have custom field")),
- (42, _("custom fields query")),
- (43, _("created to")),
- (44, _("created from")),
- (45, _("added to")),
- (46, _("added from")),
- (47, _("mime type is")),
- ]
-
- saved_view = models.ForeignKey(
- SavedView,
- on_delete=models.CASCADE,
- related_name="filter_rules",
- verbose_name=_("saved view"),
- )
-
- rule_type = models.PositiveIntegerField(_("rule type"), choices=RULE_TYPES)
-
- value = models.CharField(_("value"), max_length=255, blank=True, null=True)
-
- class Meta:
- verbose_name = _("filter rule")
- verbose_name_plural = _("filter rules")
-
- def __str__(self) -> str:
- return f"SavedViewFilterRule: {self.rule_type} : {self.value}"
-
-
-# Extending User Model Using a One-To-One Link
-class UiSettings(models.Model):
- user = models.OneToOneField(
- User,
- on_delete=models.CASCADE,
- related_name="ui_settings",
- )
- settings = models.JSONField(null=True)
-
- def __str__(self):
- return self.user.username
-
-
-class PaperlessTask(ModelWithOwner):
- ALL_STATES = sorted(states.ALL_STATES)
- TASK_STATE_CHOICES = sorted(zip(ALL_STATES, ALL_STATES))
-
- class TaskType(models.TextChoices):
- AUTO = ("auto_task", _("Auto Task"))
- SCHEDULED_TASK = ("scheduled_task", _("Scheduled Task"))
- MANUAL_TASK = ("manual_task", _("Manual Task"))
-
- class TaskName(models.TextChoices):
- CONSUME_FILE = ("consume_file", _("Consume File"))
- TRAIN_CLASSIFIER = ("train_classifier", _("Train Classifier"))
- CHECK_SANITY = ("check_sanity", _("Check Sanity"))
- INDEX_OPTIMIZE = ("index_optimize", _("Index Optimize"))
-
- task_id = models.CharField(
- max_length=255,
- unique=True,
- verbose_name=_("Task ID"),
- help_text=_("Celery ID for the Task that was run"),
- )
-
- acknowledged = models.BooleanField(
- default=False,
- verbose_name=_("Acknowledged"),
- help_text=_("If the task is acknowledged via the frontend or API"),
- )
-
- task_file_name = models.CharField(
- null=True,
- max_length=255,
- verbose_name=_("Task Filename"),
- help_text=_("Name of the file which the Task was run for"),
- )
-
- task_name = models.CharField(
- null=True,
- max_length=255,
- choices=TaskName.choices,
- verbose_name=_("Task Name"),
- help_text=_("Name of the task that was run"),
- )
-
- status = models.CharField(
- max_length=30,
- default=states.PENDING,
- choices=TASK_STATE_CHOICES,
- verbose_name=_("Task State"),
- help_text=_("Current state of the task being run"),
- )
-
- date_created = models.DateTimeField(
- null=True,
- default=timezone.now,
- verbose_name=_("Created DateTime"),
- help_text=_("Datetime field when the task result was created in UTC"),
- )
-
- date_started = models.DateTimeField(
- null=True,
- default=None,
- verbose_name=_("Started DateTime"),
- help_text=_("Datetime field when the task was started in UTC"),
- )
-
- date_done = models.DateTimeField(
- null=True,
- default=None,
- verbose_name=_("Completed DateTime"),
- help_text=_("Datetime field when the task was completed in UTC"),
- )
-
- result = models.TextField(
- null=True,
- default=None,
- verbose_name=_("Result Data"),
- help_text=_(
- "The data returned by the task",
- ),
- )
-
- type = models.CharField(
- max_length=30,
- choices=TaskType.choices,
- default=TaskType.AUTO,
- verbose_name=_("Task Type"),
- help_text=_("The type of task that was run"),
- )
-
- def __str__(self) -> str:
- return f"Task {self.task_id}"
-
-
-class Note(SoftDeleteModel):
- note = models.TextField(
- _("content"),
- blank=True,
- help_text=_("Note for the document"),
- )
-
- created = models.DateTimeField(
- _("created"),
- default=timezone.now,
- db_index=True,
- )
-
- document = models.ForeignKey(
- Document,
- blank=True,
- null=True,
- related_name="notes",
- on_delete=models.CASCADE,
- verbose_name=_("document"),
- )
-
- user = models.ForeignKey(
- User,
- blank=True,
- null=True,
- related_name="notes",
- on_delete=models.SET_NULL,
- verbose_name=_("user"),
- )
-
- class Meta:
- ordering = ("created",)
- verbose_name = _("note")
- verbose_name_plural = _("notes")
-
- def __str__(self):
- return self.note
-
-
-class ShareLink(SoftDeleteModel):
- class FileVersion(models.TextChoices):
- ARCHIVE = ("archive", _("Archive"))
- ORIGINAL = ("original", _("Original"))
-
- created = models.DateTimeField(
- _("created"),
- default=timezone.now,
- db_index=True,
- blank=True,
- editable=False,
- )
-
- expiration = models.DateTimeField(
- _("expiration"),
- blank=True,
- null=True,
- db_index=True,
- )
-
- slug = models.SlugField(
- _("slug"),
- db_index=True,
- unique=True,
- blank=True,
- editable=False,
- )
-
- document = models.ForeignKey(
- Document,
- blank=True,
- related_name="share_links",
- on_delete=models.CASCADE,
- verbose_name=_("document"),
- )
-
- file_version = models.CharField(
- max_length=50,
- choices=FileVersion.choices,
- default=FileVersion.ARCHIVE,
- )
-
- owner = models.ForeignKey(
- User,
- blank=True,
- null=True,
- related_name="share_links",
- on_delete=models.SET_NULL,
- verbose_name=_("owner"),
- )
-
- class Meta:
- ordering = ("created",)
- verbose_name = _("share link")
- verbose_name_plural = _("share links")
-
- def __str__(self):
- return f"Share Link for {self.document.title}"
-
-
-class CustomField(models.Model):
- """
- Defines the name and type of a custom field
- """
-
- class FieldDataType(models.TextChoices):
- STRING = ("string", _("String"))
- URL = ("url", _("URL"))
- DATE = ("date", _("Date"))
- BOOL = ("boolean"), _("Boolean")
- INT = ("integer", _("Integer"))
- FLOAT = ("float", _("Float"))
- MONETARY = ("monetary", _("Monetary"))
- DOCUMENTLINK = ("documentlink", _("Document Link"))
- SELECT = ("select", _("Select"))
-
- created = models.DateTimeField(
- _("created"),
- default=timezone.now,
- db_index=True,
- editable=False,
- )
-
- name = models.CharField(max_length=128)
-
- data_type = models.CharField(
- _("data type"),
- max_length=50,
- choices=FieldDataType.choices,
- editable=False,
- )
-
- extra_data = models.JSONField(
- _("extra data"),
- null=True,
- blank=True,
- help_text=_(
- "Extra data for the custom field, such as select options",
- ),
- )
-
- class Meta:
- ordering = ("created",)
- verbose_name = _("custom field")
- verbose_name_plural = _("custom fields")
- constraints = [
- models.UniqueConstraint(
- fields=["name"],
- name="%(app_label)s_%(class)s_unique_name",
- ),
- ]
-
- def __str__(self) -> str:
- return f"{self.name} : {self.data_type}"
-
-
-class CustomFieldInstance(SoftDeleteModel):
- """
- A single instance of a field, attached to a CustomField for the name and type
- and attached to a single Document to be metadata for it
- """
-
- TYPE_TO_DATA_STORE_NAME_MAP = {
- CustomField.FieldDataType.STRING: "value_text",
- CustomField.FieldDataType.URL: "value_url",
- CustomField.FieldDataType.DATE: "value_date",
- CustomField.FieldDataType.BOOL: "value_bool",
- CustomField.FieldDataType.INT: "value_int",
- CustomField.FieldDataType.FLOAT: "value_float",
- CustomField.FieldDataType.MONETARY: "value_monetary",
- CustomField.FieldDataType.DOCUMENTLINK: "value_document_ids",
- CustomField.FieldDataType.SELECT: "value_select",
- }
-
- created = models.DateTimeField(
- _("created"),
- default=timezone.now,
- db_index=True,
- editable=False,
- )
-
- document = models.ForeignKey(
- Document,
- blank=False,
- null=False,
- on_delete=models.CASCADE,
- related_name="custom_fields",
- editable=False,
- )
-
- field = models.ForeignKey(
- CustomField,
- blank=False,
- null=False,
- on_delete=models.CASCADE,
- related_name="fields",
- editable=False,
- )
-
- # Actual data storage
- value_text = models.CharField(max_length=128, null=True)
-
- value_bool = models.BooleanField(null=True)
-
- value_url = models.URLField(null=True)
-
- value_date = models.DateField(null=True)
-
- value_int = models.IntegerField(null=True)
-
- value_float = models.FloatField(null=True)
-
- value_monetary = models.CharField(null=True, max_length=128)
-
- value_monetary_amount = models.GeneratedField(
- expression=Case(
- # If the value starts with a number and no currency symbol, use the whole string
- models.When(
- value_monetary__regex=r"^\d+",
- then=Cast(
- Substr("value_monetary", 1),
- output_field=models.DecimalField(decimal_places=2, max_digits=65),
- ),
- ),
- # If the value starts with a 3-char currency symbol, use the rest of the string
- default=Cast(
- Substr("value_monetary", 4),
- output_field=models.DecimalField(decimal_places=2, max_digits=65),
- ),
- output_field=models.DecimalField(decimal_places=2, max_digits=65),
- ),
- output_field=models.DecimalField(decimal_places=2, max_digits=65),
- db_persist=True,
- )
-
- value_document_ids = models.JSONField(null=True)
-
- value_select = models.CharField(null=True, max_length=16)
-
- class Meta:
- ordering = ("created",)
- verbose_name = _("custom field instance")
- verbose_name_plural = _("custom field instances")
- constraints = [
- models.UniqueConstraint(
- fields=["document", "field"],
- name="%(app_label)s_%(class)s_unique_document_field",
- ),
- ]
-
- def __str__(self) -> str:
- value = (
- next(
- option.get("label")
- for option in self.field.extra_data["select_options"]
- if option.get("id") == self.value_select
- )
- if (
- self.field.data_type == CustomField.FieldDataType.SELECT
- and self.value_select is not None
- )
- else self.value
- )
- return str(self.field.name) + f" : {value}"
-
- @classmethod
- def get_value_field_name(cls, data_type: CustomField.FieldDataType):
- try:
- return cls.TYPE_TO_DATA_STORE_NAME_MAP[data_type]
- except KeyError: # pragma: no cover
- raise NotImplementedError(data_type)
-
- @property
- def value(self):
- """
- Based on the data type, access the actual value the instance stores
- A little shorthand/quick way to get what is actually here
- """
- value_field_name = self.get_value_field_name(self.field.data_type)
- return getattr(self, value_field_name)
-
-
-if settings.AUDIT_LOG_ENABLED:
- auditlog.register(
- Document,
- m2m_fields={"tags"},
- exclude_fields=["modified"],
- )
- auditlog.register(Correspondent)
- auditlog.register(Tag)
- auditlog.register(DocumentType)
- auditlog.register(Note)
- auditlog.register(CustomField)
- auditlog.register(CustomFieldInstance)
-
-
-class WorkflowTrigger(models.Model):
- class WorkflowTriggerMatching(models.IntegerChoices):
- # No auto matching
- NONE = MatchingModel.MATCH_NONE, _("None")
- ANY = MatchingModel.MATCH_ANY, _("Any word")
- ALL = MatchingModel.MATCH_ALL, _("All words")
- LITERAL = MatchingModel.MATCH_LITERAL, _("Exact match")
- REGEX = MatchingModel.MATCH_REGEX, _("Regular expression")
- FUZZY = MatchingModel.MATCH_FUZZY, _("Fuzzy word")
-
- class WorkflowTriggerType(models.IntegerChoices):
- CONSUMPTION = 1, _("Consumption Started")
- DOCUMENT_ADDED = 2, _("Document Added")
- DOCUMENT_UPDATED = 3, _("Document Updated")
- SCHEDULED = 4, _("Scheduled")
-
- class DocumentSourceChoices(models.IntegerChoices):
- CONSUME_FOLDER = DocumentSource.ConsumeFolder.value, _("Consume Folder")
- API_UPLOAD = DocumentSource.ApiUpload.value, _("Api Upload")
- MAIL_FETCH = DocumentSource.MailFetch.value, _("Mail Fetch")
- WEB_UI = DocumentSource.WebUI.value, _("Web UI")
-
- class ScheduleDateField(models.TextChoices):
- ADDED = "added", _("Added")
- CREATED = "created", _("Created")
- MODIFIED = "modified", _("Modified")
- CUSTOM_FIELD = "custom_field", _("Custom Field")
-
- type = models.PositiveIntegerField(
- _("Workflow Trigger Type"),
- choices=WorkflowTriggerType.choices,
- default=WorkflowTriggerType.CONSUMPTION,
- )
-
- sources = MultiSelectField(
- max_length=7,
- choices=DocumentSourceChoices.choices,
- default=f"{DocumentSource.ConsumeFolder},{DocumentSource.ApiUpload},{DocumentSource.MailFetch},{DocumentSource.WebUI}",
- )
-
- filter_path = models.CharField(
- _("filter path"),
- max_length=256,
- null=True,
- blank=True,
- help_text=_(
- "Only consume documents with a path that matches "
- "this if specified. Wildcards specified as * are "
- "allowed. Case insensitive.",
- ),
- )
-
- filter_filename = models.CharField(
- _("filter filename"),
- max_length=256,
- null=True,
- blank=True,
- help_text=_(
- "Only consume documents which entirely match this "
- "filename if specified. Wildcards such as *.pdf or "
- "*invoice* are allowed. Case insensitive.",
- ),
- )
-
- filter_mailrule = models.ForeignKey(
- "paperless_mail.MailRule",
- null=True,
- blank=True,
- on_delete=models.SET_NULL,
- verbose_name=_("filter documents from this mail rule"),
- )
-
- match = models.CharField(_("match"), max_length=256, blank=True)
-
- matching_algorithm = models.PositiveIntegerField(
- _("matching algorithm"),
- choices=WorkflowTriggerMatching.choices,
- default=WorkflowTriggerMatching.NONE,
- )
-
- is_insensitive = models.BooleanField(_("is insensitive"), default=True)
-
- filter_has_tags = models.ManyToManyField(
- Tag,
- blank=True,
- verbose_name=_("has these tag(s)"),
- )
-
- filter_has_document_type = models.ForeignKey(
- DocumentType,
- null=True,
- blank=True,
- on_delete=models.SET_NULL,
- verbose_name=_("has this document type"),
- )
-
- filter_has_correspondent = models.ForeignKey(
- Correspondent,
- null=True,
- blank=True,
- on_delete=models.SET_NULL,
- verbose_name=_("has this correspondent"),
- )
-
- schedule_offset_days = models.PositiveIntegerField(
- _("schedule offset days"),
- default=0,
- help_text=_(
- "The number of days to offset the schedule trigger by.",
- ),
- )
-
- schedule_is_recurring = models.BooleanField(
- _("schedule is recurring"),
- default=False,
- help_text=_(
- "If the schedule should be recurring.",
- ),
- )
-
- schedule_recurring_interval_days = models.PositiveIntegerField(
- _("schedule recurring delay in days"),
- default=1,
- validators=[MinValueValidator(1)],
- help_text=_(
- "The number of days between recurring schedule triggers.",
- ),
- )
-
- schedule_date_field = models.CharField(
- _("schedule date field"),
- max_length=20,
- choices=ScheduleDateField.choices,
- default=ScheduleDateField.ADDED,
- help_text=_(
- "The field to check for a schedule trigger.",
- ),
- )
-
- schedule_date_custom_field = models.ForeignKey(
- CustomField,
- null=True,
- blank=True,
- on_delete=models.SET_NULL,
- verbose_name=_("schedule date custom field"),
- )
-
- class Meta:
- verbose_name = _("workflow trigger")
- verbose_name_plural = _("workflow triggers")
-
- def __str__(self):
- return f"WorkflowTrigger {self.pk}"
-
-
-class WorkflowActionEmail(models.Model):
- subject = models.CharField(
- _("email subject"),
- max_length=256,
- null=False,
- help_text=_(
- "The subject of the email, can include some placeholders, "
- "see documentation.",
- ),
- )
-
- body = models.TextField(
- _("email body"),
- null=False,
- help_text=_(
- "The body (message) of the email, can include some placeholders, "
- "see documentation.",
- ),
- )
-
- to = models.TextField(
- _("emails to"),
- null=False,
- help_text=_(
- "The destination email addresses, comma separated.",
- ),
- )
-
- include_document = models.BooleanField(
- default=False,
- verbose_name=_("include document in email"),
- )
-
- def __str__(self):
- return f"Workflow Email Action {self.pk}"
-
-
-class WorkflowActionWebhook(models.Model):
- # We dont use the built-in URLField because it is not flexible enough
- # validation is handled in the serializer
- url = models.CharField(
- _("webhook url"),
- null=False,
- max_length=256,
- help_text=_("The destination URL for the notification."),
- )
-
- use_params = models.BooleanField(
- default=True,
- verbose_name=_("use parameters"),
- )
-
- as_json = models.BooleanField(
- default=False,
- verbose_name=_("send as JSON"),
- )
-
- params = models.JSONField(
- _("webhook parameters"),
- null=True,
- blank=True,
- help_text=_("The parameters to send with the webhook URL if body not used."),
- )
-
- body = models.TextField(
- _("webhook body"),
- null=True,
- blank=True,
- help_text=_("The body to send with the webhook URL if parameters not used."),
- )
-
- headers = models.JSONField(
- _("webhook headers"),
- null=True,
- blank=True,
- help_text=_("The headers to send with the webhook URL."),
- )
-
- include_document = models.BooleanField(
- default=False,
- verbose_name=_("include document in webhook"),
- )
-
- def __str__(self):
- return f"Workflow Webhook Action {self.pk}"
-
-
-class WorkflowAction(models.Model):
- class WorkflowActionType(models.IntegerChoices):
- ASSIGNMENT = (
- 1,
- _("Assignment"),
- )
- REMOVAL = (
- 2,
- _("Removal"),
- )
- EMAIL = (
- 3,
- _("Email"),
- )
- WEBHOOK = (
- 4,
- _("Webhook"),
- )
-
- type = models.PositiveIntegerField(
- _("Workflow Action Type"),
- choices=WorkflowActionType.choices,
- default=WorkflowActionType.ASSIGNMENT,
- )
-
- assign_title = models.CharField(
- _("assign title"),
- max_length=256,
- null=True,
- blank=True,
- help_text=_(
- "Assign a document title, can include some placeholders, "
- "see documentation.",
- ),
- )
-
- assign_tags = models.ManyToManyField(
- Tag,
- blank=True,
- related_name="+",
- verbose_name=_("assign this tag"),
- )
-
- assign_document_type = models.ForeignKey(
- DocumentType,
- null=True,
- blank=True,
- on_delete=models.SET_NULL,
- related_name="+",
- verbose_name=_("assign this document type"),
- )
-
- assign_correspondent = models.ForeignKey(
- Correspondent,
- null=True,
- blank=True,
- on_delete=models.SET_NULL,
- related_name="+",
- verbose_name=_("assign this correspondent"),
- )
-
- assign_storage_path = models.ForeignKey(
- StoragePath,
- null=True,
- blank=True,
- on_delete=models.SET_NULL,
- related_name="+",
- verbose_name=_("assign this storage path"),
- )
-
- assign_owner = models.ForeignKey(
- User,
- null=True,
- blank=True,
- on_delete=models.SET_NULL,
- related_name="+",
- verbose_name=_("assign this owner"),
- )
-
- assign_view_users = models.ManyToManyField(
- User,
- blank=True,
- related_name="+",
- verbose_name=_("grant view permissions to these users"),
- )
-
- assign_view_groups = models.ManyToManyField(
- Group,
- blank=True,
- related_name="+",
- verbose_name=_("grant view permissions to these groups"),
- )
-
- assign_change_users = models.ManyToManyField(
- User,
- blank=True,
- related_name="+",
- verbose_name=_("grant change permissions to these users"),
- )
-
- assign_change_groups = models.ManyToManyField(
- Group,
- blank=True,
- related_name="+",
- verbose_name=_("grant change permissions to these groups"),
- )
-
- assign_custom_fields = models.ManyToManyField(
- CustomField,
- blank=True,
- related_name="+",
- verbose_name=_("assign these custom fields"),
- )
-
- assign_custom_fields_values = models.JSONField(
- _("custom field values"),
- null=True,
- blank=True,
- help_text=_(
- "Optional values to assign to the custom fields.",
- ),
- default=dict,
- )
-
- remove_tags = models.ManyToManyField(
- Tag,
- blank=True,
- related_name="+",
- verbose_name=_("remove these tag(s)"),
- )
-
- remove_all_tags = models.BooleanField(
- default=False,
- verbose_name=_("remove all tags"),
- )
-
- remove_document_types = models.ManyToManyField(
- DocumentType,
- blank=True,
- related_name="+",
- verbose_name=_("remove these document type(s)"),
- )
-
- remove_all_document_types = models.BooleanField(
- default=False,
- verbose_name=_("remove all document types"),
- )
-
- remove_correspondents = models.ManyToManyField(
- Correspondent,
- blank=True,
- related_name="+",
- verbose_name=_("remove these correspondent(s)"),
- )
-
- remove_all_correspondents = models.BooleanField(
- default=False,
- verbose_name=_("remove all correspondents"),
- )
-
- remove_storage_paths = models.ManyToManyField(
- StoragePath,
- blank=True,
- related_name="+",
- verbose_name=_("remove these storage path(s)"),
- )
-
- remove_all_storage_paths = models.BooleanField(
- default=False,
- verbose_name=_("remove all storage paths"),
- )
-
- remove_owners = models.ManyToManyField(
- User,
- blank=True,
- related_name="+",
- verbose_name=_("remove these owner(s)"),
- )
-
- remove_all_owners = models.BooleanField(
- default=False,
- verbose_name=_("remove all owners"),
- )
-
- remove_view_users = models.ManyToManyField(
- User,
- blank=True,
- related_name="+",
- verbose_name=_("remove view permissions for these users"),
- )
-
- remove_view_groups = models.ManyToManyField(
- Group,
- blank=True,
- related_name="+",
- verbose_name=_("remove view permissions for these groups"),
- )
-
- remove_change_users = models.ManyToManyField(
- User,
- blank=True,
- related_name="+",
- verbose_name=_("remove change permissions for these users"),
- )
-
- remove_change_groups = models.ManyToManyField(
- Group,
- blank=True,
- related_name="+",
- verbose_name=_("remove change permissions for these groups"),
- )
-
- remove_all_permissions = models.BooleanField(
- default=False,
- verbose_name=_("remove all permissions"),
- )
-
- remove_custom_fields = models.ManyToManyField(
- CustomField,
- blank=True,
- related_name="+",
- verbose_name=_("remove these custom fields"),
- )
-
- remove_all_custom_fields = models.BooleanField(
- default=False,
- verbose_name=_("remove all custom fields"),
- )
-
- email = models.ForeignKey(
- WorkflowActionEmail,
- null=True,
- blank=True,
- on_delete=models.SET_NULL,
- related_name="action",
- verbose_name=_("email"),
- )
-
- webhook = models.ForeignKey(
- WorkflowActionWebhook,
- null=True,
- blank=True,
- on_delete=models.SET_NULL,
- related_name="action",
- verbose_name=_("webhook"),
- )
-
- class Meta:
- verbose_name = _("workflow action")
- verbose_name_plural = _("workflow actions")
-
- def __str__(self):
- return f"WorkflowAction {self.pk}"
-
-
-class Workflow(models.Model):
- name = models.CharField(_("name"), max_length=256, unique=True)
-
- order = models.IntegerField(_("order"), default=0)
-
- triggers = models.ManyToManyField(
- WorkflowTrigger,
- related_name="workflows",
- blank=False,
- verbose_name=_("triggers"),
- )
-
- actions = models.ManyToManyField(
- WorkflowAction,
- related_name="workflows",
- blank=False,
- verbose_name=_("actions"),
- )
-
- enabled = models.BooleanField(_("enabled"), default=True)
-
- def __str__(self):
- return f"Workflow: {self.name}"
-
-
-class WorkflowRun(models.Model):
- workflow = models.ForeignKey(
- Workflow,
- on_delete=models.CASCADE,
- related_name="runs",
- verbose_name=_("workflow"),
- )
-
- type = models.PositiveIntegerField(
- _("workflow trigger type"),
- choices=WorkflowTrigger.WorkflowTriggerType.choices,
- null=True,
- )
-
- document = models.ForeignKey(
- Document,
- null=True,
- on_delete=models.CASCADE,
- related_name="workflow_runs",
- verbose_name=_("document"),
- )
-
- run_at = models.DateTimeField(
- _("date run"),
- default=timezone.now,
- db_index=True,
- )
-
- class Meta:
- verbose_name = _("workflow run")
- verbose_name_plural = _("workflow runs")
-
- def __str__(self):
- return f"WorkflowRun of {self.workflow} at {self.run_at} on {self.document}"
from django.utils import timezone
from tqdm import tqdm
-from documents.models import Document
-from documents.models import PaperlessTask
+from paperless.models import Document
+from paperless.models import PaperlessTask
class SanityCheckMessages:
from documents import bulk_edit
from documents.data_models import DocumentSource
-from documents.models import Correspondent
-from documents.models import CustomField
-from documents.models import CustomFieldInstance
-from documents.models import Document
-from documents.models import DocumentType
-from documents.models import MatchingModel
-from documents.models import Note
-from documents.models import PaperlessTask
-from documents.models import SavedView
-from documents.models import SavedViewFilterRule
-from documents.models import ShareLink
-from documents.models import StoragePath
-from documents.models import Tag
-from documents.models import UiSettings
-from documents.models import Workflow
-from documents.models import WorkflowAction
-from documents.models import WorkflowActionEmail
-from documents.models import WorkflowActionWebhook
-from documents.models import WorkflowTrigger
from documents.parsers import is_mime_type_supported
from documents.permissions import get_groups_with_only_permission
from documents.permissions import set_permissions_for_object
from documents.templating.utils import convert_format_str_to_template_format
from documents.validators import uri_validator
from documents.validators import url_validator
+from paperless.models import Correspondent
+from paperless.models import CustomField
+from paperless.models import CustomFieldInstance
+from paperless.models import Document
+from paperless.models import DocumentType
+from paperless.models import MatchingModel
+from paperless.models import Note
+from paperless.models import PaperlessTask
+from paperless.models import SavedView
+from paperless.models import SavedViewFilterRule
+from paperless.models import ShareLink
+from paperless.models import StoragePath
+from paperless.models import Tag
+from paperless.models import UiSettings
+from paperless.models import Workflow
+from paperless.models import WorkflowAction
+from paperless.models import WorkflowActionEmail
+from paperless.models import WorkflowActionWebhook
+from paperless.models import WorkflowTrigger
if TYPE_CHECKING:
from collections.abc import Iterable
from documents.file_handling import delete_empty_directories
from documents.file_handling import generate_unique_filename
from documents.mail import send_email
-from documents.models import Correspondent
-from documents.models import CustomField
-from documents.models import CustomFieldInstance
-from documents.models import Document
-from documents.models import DocumentType
-from documents.models import MatchingModel
-from documents.models import PaperlessTask
-from documents.models import SavedView
-from documents.models import Tag
-from documents.models import Workflow
-from documents.models import WorkflowAction
-from documents.models import WorkflowRun
-from documents.models import WorkflowTrigger
from documents.permissions import get_objects_for_user_owner_aware
from documents.permissions import set_permissions_for_object
from documents.templating.workflows import parse_w_workflow_placeholders
+from paperless.models import Correspondent
+from paperless.models import CustomField
+from paperless.models import CustomFieldInstance
+from paperless.models import Document
+from paperless.models import DocumentType
+from paperless.models import MatchingModel
+from paperless.models import PaperlessTask
+from paperless.models import SavedView
+from paperless.models import Tag
+from paperless.models import Workflow
+from paperless.models import WorkflowAction
+from paperless.models import WorkflowRun
+from paperless.models import WorkflowTrigger
if TYPE_CHECKING:
from pathlib import Path
from documents.double_sided import CollatePlugin
from documents.file_handling import create_source_path_directory
from documents.file_handling import generate_unique_filename
-from documents.models import Correspondent
-from documents.models import CustomFieldInstance
-from documents.models import Document
-from documents.models import DocumentType
-from documents.models import PaperlessTask
-from documents.models import StoragePath
-from documents.models import Tag
-from documents.models import Workflow
-from documents.models import WorkflowRun
-from documents.models import WorkflowTrigger
from documents.parsers import DocumentParser
from documents.parsers import get_parser_class_for_mime_type
from documents.plugins.base import ConsumeTaskPlugin
from documents.signals import document_updated
from documents.signals.handlers import cleanup_document_deletion
from documents.signals.handlers import run_workflows
+from paperless.models import Correspondent
+from paperless.models import CustomFieldInstance
+from paperless.models import Document
+from paperless.models import DocumentType
+from paperless.models import PaperlessTask
+from paperless.models import StoragePath
+from paperless.models import Tag
+from paperless.models import Workflow
+from paperless.models import WorkflowRun
+from paperless.models import WorkflowTrigger
if settings.AUDIT_LOG_ENABLED:
from auditlog.models import LogEntry
from jinja2.sandbox import SandboxedEnvironment
from jinja2.sandbox import SecurityError
-from documents.models import Correspondent
-from documents.models import CustomField
-from documents.models import CustomFieldInstance
-from documents.models import Document
-from documents.models import DocumentType
-from documents.models import StoragePath
-from documents.models import Tag
+from paperless.models import Correspondent
+from paperless.models import CustomField
+from paperless.models import CustomFieldInstance
+from paperless.models import Document
+from paperless.models import DocumentType
+from paperless.models import StoragePath
+from paperless.models import Tag
logger = logging.getLogger("paperless.templating")
from factory import Faker
from factory.django import DjangoModelFactory
-from documents.models import Correspondent
-from documents.models import Document
+from paperless.models import Correspondent
+from paperless.models import Document
class CorrespondentFactory(DjangoModelFactory):
from documents import index
from documents.admin import DocumentAdmin
-from documents.models import Document
from documents.tests.utils import DirectoriesMixin
from paperless.admin import PaperlessUserAdmin
+from paperless.models import Document
class TestDocumentAdmin(DirectoriesMixin, TestCase):
from rest_framework import status
from rest_framework.test import APITestCase
-from documents.models import Correspondent
-from documents.models import Document
-from documents.models import DocumentType
from documents.tests.utils import DirectoriesMixin
from documents.tests.utils import SampleDirMixin
+from paperless.models import Correspondent
+from paperless.models import Document
+from paperless.models import DocumentType
class TestBulkDownload(DirectoriesMixin, SampleDirMixin, APITestCase):
from rest_framework import status
from rest_framework.test import APITestCase
-from documents.models import Correspondent
-from documents.models import CustomField
-from documents.models import Document
-from documents.models import DocumentType
-from documents.models import StoragePath
-from documents.models import Tag
from documents.tests.utils import DirectoriesMixin
+from paperless.models import Correspondent
+from paperless.models import CustomField
+from paperless.models import Document
+from paperless.models import DocumentType
+from paperless.models import StoragePath
+from paperless.models import Tag
class TestBulkEditAPI(DirectoriesMixin, APITestCase):
from rest_framework import status
from rest_framework.test import APITestCase
-from documents.models import CustomField
-from documents.models import CustomFieldInstance
-from documents.models import Document
from documents.tests.utils import DirectoriesMixin
+from paperless.models import CustomField
+from paperless.models import CustomFieldInstance
+from paperless.models import Document
class TestCustomFieldsAPI(DirectoriesMixin, APITestCase):
from documents.caching import CLASSIFIER_MODIFIED_KEY
from documents.caching import CLASSIFIER_VERSION_KEY
from documents.data_models import DocumentSource
-from documents.models import Correspondent
-from documents.models import CustomField
-from documents.models import CustomFieldInstance
-from documents.models import Document
-from documents.models import DocumentType
-from documents.models import MatchingModel
-from documents.models import Note
-from documents.models import SavedView
-from documents.models import ShareLink
-from documents.models import StoragePath
-from documents.models import Tag
-from documents.models import Workflow
-from documents.models import WorkflowAction
-from documents.models import WorkflowTrigger
from documents.signals.handlers import run_workflows
from documents.tests.utils import DirectoriesMixin
from documents.tests.utils import DocumentConsumeDelayMixin
+from paperless.models import Correspondent
+from paperless.models import CustomField
+from paperless.models import CustomFieldInstance
+from paperless.models import Document
+from paperless.models import DocumentType
+from paperless.models import MatchingModel
+from paperless.models import Note
+from paperless.models import SavedView
+from paperless.models import ShareLink
+from paperless.models import StoragePath
+from paperless.models import Tag
+from paperless.models import Workflow
+from paperless.models import WorkflowAction
+from paperless.models import WorkflowTrigger
class TestDocumentApi(DirectoriesMixin, DocumentConsumeDelayMixin, APITestCase):
from django.contrib.auth.models import User
from rest_framework.test import APITestCase
-from documents.models import CustomField
-from documents.models import Document
from documents.serialisers import DocumentSerializer
from documents.tests.utils import DirectoriesMixin
+from paperless.models import CustomField
+from paperless.models import Document
class DocumentWrapper:
from rest_framework import status
from rest_framework.test import APITestCase
-from documents.models import Correspondent
-from documents.models import Document
-from documents.models import DocumentType
-from documents.models import StoragePath
-from documents.models import Tag
from documents.tests.utils import DirectoriesMixin
+from paperless.models import Correspondent
+from paperless.models import Document
+from paperless.models import DocumentType
+from paperless.models import StoragePath
+from paperless.models import Tag
class TestApiObjects(DirectoriesMixin, APITestCase):
from rest_framework import status
from rest_framework.test import APITestCase
-from documents.models import Correspondent
-from documents.models import Document
-from documents.models import DocumentType
-from documents.models import MatchingModel
-from documents.models import StoragePath
-from documents.models import Tag
from documents.tests.utils import DirectoriesMixin
+from paperless.models import Correspondent
+from paperless.models import Document
+from paperless.models import DocumentType
+from paperless.models import MatchingModel
+from paperless.models import StoragePath
+from paperless.models import Tag
class TestApiAuth(DirectoriesMixin, APITestCase):
from documents import index
from documents.bulk_edit import set_permissions
-from documents.models import Correspondent
-from documents.models import CustomField
-from documents.models import CustomFieldInstance
-from documents.models import Document
-from documents.models import DocumentType
-from documents.models import Note
-from documents.models import SavedView
-from documents.models import StoragePath
-from documents.models import Tag
-from documents.models import Workflow
from documents.tests.utils import DirectoriesMixin
+from paperless.models import Correspondent
+from paperless.models import CustomField
+from paperless.models import CustomFieldInstance
+from paperless.models import Document
+from paperless.models import DocumentType
+from paperless.models import Note
+from paperless.models import SavedView
+from paperless.models import StoragePath
+from paperless.models import Tag
+from paperless.models import Workflow
from paperless_mail.models import MailAccount
from paperless_mail.models import MailRule
from rest_framework import status
from rest_framework.test import APITestCase
-from documents.models import PaperlessTask
from paperless import version
+from paperless.models import PaperlessTask
class TestSystemStatus(APITestCase):
from rest_framework import status
from rest_framework.test import APITestCase
-from documents.models import PaperlessTask
from documents.tests.utils import DirectoriesMixin
from documents.views import TasksViewSet
+from paperless.models import PaperlessTask
class TestTasks(DirectoriesMixin, APITestCase):
from rest_framework import status
from rest_framework.test import APITestCase
-from documents.models import Document
+from paperless.models import Document
class TestTrashAPI(APITestCase):
from rest_framework.test import APITestCase
from documents.data_models import DocumentSource
-from documents.models import Correspondent
-from documents.models import CustomField
-from documents.models import DocumentType
-from documents.models import StoragePath
-from documents.models import Tag
-from documents.models import Workflow
-from documents.models import WorkflowAction
-from documents.models import WorkflowTrigger
from documents.tests.utils import DirectoriesMixin
+from paperless.models import Correspondent
+from paperless.models import CustomField
+from paperless.models import DocumentType
+from paperless.models import StoragePath
+from paperless.models import Tag
+from paperless.models import Workflow
+from paperless.models import WorkflowAction
+from paperless.models import WorkflowTrigger
class TestApiWorkflows(DirectoriesMixin, APITestCase):
from documents.data_models import ConsumableDocument
from documents.data_models import DocumentMetadataOverrides
from documents.data_models import DocumentSource
-from documents.models import Document
-from documents.models import Tag
from documents.plugins.base import StopConsumeTaskError
from documents.tests.utils import DirectoriesMixin
from documents.tests.utils import DocumentConsumeDelayMixin
from documents.tests.utils import DummyProgressManager
from documents.tests.utils import FileSystemAssertsMixin
from documents.tests.utils import SampleDirMixin
+from paperless.models import Document
+from paperless.models import Tag
try:
import zxingcpp # noqa: F401
from guardian.shortcuts import get_users_with_perms
from documents import bulk_edit
-from documents.models import Correspondent
-from documents.models import CustomField
-from documents.models import CustomFieldInstance
-from documents.models import Document
-from documents.models import DocumentType
-from documents.models import StoragePath
-from documents.models import Tag
from documents.tests.utils import DirectoriesMixin
+from paperless.models import Correspondent
+from paperless.models import CustomField
+from paperless.models import CustomFieldInstance
+from paperless.models import Document
+from paperless.models import DocumentType
+from paperless.models import StoragePath
+from paperless.models import Tag
class TestBulkEdit(DirectoriesMixin, TestCase):
from documents.checks import changed_password_check
from documents.checks import filename_format_check
from documents.checks import parser_check
-from documents.models import Document
from documents.tests.factories import DocumentFactory
+from paperless.models import Document
class TestDocumentChecks(TestCase):
from documents.classifier import DocumentClassifier
from documents.classifier import IncompatibleClassifierVersionError
from documents.classifier import load_classifier
-from documents.models import Correspondent
-from documents.models import Document
-from documents.models import DocumentType
-from documents.models import MatchingModel
-from documents.models import StoragePath
-from documents.models import Tag
from documents.tests.utils import DirectoriesMixin
+from paperless.models import Correspondent
+from paperless.models import Document
+from paperless.models import DocumentType
+from paperless.models import MatchingModel
+from paperless.models import StoragePath
+from paperless.models import Tag
def dummy_preprocess(content: str):
from documents.consumer import ConsumerError
from documents.data_models import DocumentMetadataOverrides
from documents.data_models import DocumentSource
-from documents.models import Correspondent
-from documents.models import CustomField
-from documents.models import Document
-from documents.models import DocumentType
-from documents.models import StoragePath
-from documents.models import Tag
from documents.parsers import DocumentParser
from documents.parsers import ParseError
from documents.plugins.helpers import ProgressStatusOptions
from documents.tests.utils import DirectoriesMixin
from documents.tests.utils import FileSystemAssertsMixin
from documents.tests.utils import GetConsumerMixin
+from paperless.models import Correspondent
+from paperless.models import CustomField
+from paperless.models import Document
+from paperless.models import DocumentType
+from paperless.models import StoragePath
+from paperless.models import Tag
from paperless_mail.models import MailRule
from paperless_mail.parsers import MailDocumentParser
from whoosh import query
from documents.index import get_permissions_criterias
-from documents.models import User
+from paperless.models import User
class TestDelayedQuery(TestCase):
from django.test import override_settings
from django.utils import timezone
-from documents.models import Correspondent
-from documents.models import Document
from documents.tasks import empty_trash
+from paperless.models import Correspondent
+from paperless.models import Document
class TestDocument(TestCase):
from documents.file_handling import create_source_path_directory
from documents.file_handling import delete_empty_directories
from documents.file_handling import generate_filename
-from documents.models import Correspondent
-from documents.models import CustomField
-from documents.models import CustomFieldInstance
-from documents.models import Document
-from documents.models import DocumentType
-from documents.models import StoragePath
from documents.tasks import empty_trash
from documents.tests.utils import DirectoriesMixin
from documents.tests.utils import FileSystemAssertsMixin
+from paperless.models import Correspondent
+from paperless.models import CustomField
+from paperless.models import CustomFieldInstance
+from paperless.models import Document
+from paperless.models import DocumentType
+from paperless.models import StoragePath
class TestFileHandling(DirectoriesMixin, FileSystemAssertsMixin, TestCase):
from django.test import TestCase
from documents import index
-from documents.models import Document
from documents.tests.utils import DirectoriesMixin
+from paperless.models import Document
class TestAutoComplete(DirectoriesMixin, TestCase):
from django.test import override_settings
from documents.file_handling import generate_filename
-from documents.models import Document
from documents.tasks import update_document_content_maybe_archive_file
from documents.tests.utils import DirectoriesMixin
from documents.tests.utils import FileSystemAssertsMixin
+from paperless.models import Document
sample_file = os.path.join(os.path.dirname(__file__), "samples", "simple.pdf")
from documents.consumer import ConsumerError
from documents.data_models import ConsumableDocument
from documents.management.commands import document_consumer
-from documents.models import Tag
from documents.tests.utils import DirectoriesMixin
from documents.tests.utils import DocumentConsumeDelayMixin
+from paperless.models import Tag
class ConsumerThread(Thread):
from guardian.shortcuts import assign_perm
from documents.management.commands import document_exporter
-from documents.models import Correspondent
-from documents.models import CustomField
-from documents.models import CustomFieldInstance
-from documents.models import Document
-from documents.models import DocumentType
-from documents.models import Note
-from documents.models import StoragePath
-from documents.models import Tag
-from documents.models import User
-from documents.models import Workflow
-from documents.models import WorkflowAction
-from documents.models import WorkflowTrigger
from documents.sanity_checker import check_sanity
from documents.settings import EXPORTER_FILE_NAME
from documents.tests.utils import DirectoriesMixin
from documents.tests.utils import FileSystemAssertsMixin
from documents.tests.utils import SampleDirMixin
from documents.tests.utils import paperless_environment
+from paperless.models import Correspondent
+from paperless.models import CustomField
+from paperless.models import CustomFieldInstance
+from paperless.models import Document
+from paperless.models import DocumentType
+from paperless.models import Note
+from paperless.models import StoragePath
+from paperless.models import Tag
+from paperless.models import User
+from paperless.models import Workflow
+from paperless.models import WorkflowAction
+from paperless.models import WorkflowTrigger
from paperless_mail.models import MailAccount
from django.core.management import call_command
from django.test import TestCase
-from documents.models import Document
+from paperless.models import Document
class TestFuzzyMatchCommand(TestCase):
from django.test import TestCase
from documents.management.commands.document_importer import Command
-from documents.models import Document
from documents.settings import EXPORTER_ARCHIVE_NAME
from documents.settings import EXPORTER_FILE_NAME
from documents.tests.utils import DirectoriesMixin
from documents.tests.utils import FileSystemAssertsMixin
from documents.tests.utils import SampleDirMixin
+from paperless.models import Document
class TestCommandImport(
from django.core.management.base import CommandError
from django.test import TestCase
-from documents.models import Correspondent
-from documents.models import Document
-from documents.models import DocumentType
-from documents.models import StoragePath
-from documents.models import Tag
from documents.tests.utils import DirectoriesMixin
+from paperless.models import Correspondent
+from paperless.models import Document
+from paperless.models import DocumentType
+from paperless.models import StoragePath
+from paperless.models import Tag
class TestRetagger(DirectoriesMixin, TestCase):
from django.test import TestCase
from documents.management.commands.document_thumbnails import _process_document
-from documents.models import Document
from documents.parsers import get_default_thumbnail
from documents.tests.utils import DirectoriesMixin
from documents.tests.utils import FileSystemAssertsMixin
+from paperless.models import Document
class TestMakeThumbnails(DirectoriesMixin, FileSystemAssertsMixin, TestCase):
from django.test import override_settings
from documents import matching
-from documents.models import Correspondent
-from documents.models import Document
-from documents.models import DocumentType
-from documents.models import Tag
from documents.signals import document_consumption_finished
+from paperless.models import Correspondent
+from paperless.models import Document
+from paperless.models import DocumentType
+from paperless.models import Tag
class _TestMatchingBase(TestCase):
-from documents.models import StoragePath
from documents.tests.utils import TestMigrations
+from paperless.models import StoragePath
class TestMigrateStoragePathToTemplate(TestMigrations):
from django.test import TestCase
-from documents.models import Correspondent
-from documents.models import Document
from documents.tests.factories import CorrespondentFactory
from documents.tests.factories import DocumentFactory
+from paperless.models import Correspondent
+from paperless.models import Document
class CorrespondentTestCase(TestCase):
from django.conf import settings
from django.test import TestCase
-from documents.models import Document
from documents.sanity_checker import check_sanity
from documents.tests.utils import DirectoriesMixin
+from paperless.models import Document
class TestSanityCheck(DirectoriesMixin, TestCase):
from documents.data_models import ConsumableDocument
from documents.data_models import DocumentMetadataOverrides
from documents.data_models import DocumentSource
-from documents.models import PaperlessTask
from documents.signals.handlers import before_task_publish_handler
from documents.signals.handlers import task_failure_handler
from documents.signals.handlers import task_postrun_handler
from documents.signals.handlers import task_prerun_handler
from documents.tests.test_consumer import fake_magic_from_file
from documents.tests.utils import DirectoriesMixin
+from paperless.models import PaperlessTask
@mock.patch("documents.consumer.magic.from_file", fake_magic_from_file)
from django.utils import timezone
from documents import tasks
-from documents.models import Correspondent
-from documents.models import Document
-from documents.models import DocumentType
-from documents.models import Tag
from documents.sanity_checker import SanityCheckFailedException
from documents.sanity_checker import SanityCheckMessages
from documents.tests.test_classifier import dummy_preprocess
from documents.tests.utils import DirectoriesMixin
from documents.tests.utils import FileSystemAssertsMixin
+from paperless.models import Correspondent
+from paperless.models import Document
+from paperless.models import DocumentType
+from paperless.models import Tag
class TestIndexReindex(DirectoriesMixin, TestCase):
from django.utils import timezone
from rest_framework import status
-from documents.models import Document
-from documents.models import ShareLink
from documents.tests.utils import DirectoriesMixin
from paperless.models import ApplicationConfiguration
+from paperless.models import Document
+from paperless.models import ShareLink
class TestViews(DirectoriesMixin, TestCase):
from documents.data_models import ConsumableDocument
from documents.data_models import DocumentSource
from documents.matching import document_matches_workflow
-from documents.models import Correspondent
-from documents.models import CustomField
-from documents.models import CustomFieldInstance
-from documents.models import Document
-from documents.models import DocumentType
-from documents.models import MatchingModel
-from documents.models import StoragePath
-from documents.models import Tag
-from documents.models import Workflow
-from documents.models import WorkflowAction
-from documents.models import WorkflowActionEmail
-from documents.models import WorkflowActionWebhook
-from documents.models import WorkflowRun
-from documents.models import WorkflowTrigger
from documents.signals import document_consumption_finished
from documents.tests.utils import DirectoriesMixin
from documents.tests.utils import DummyProgressManager
from documents.tests.utils import FileSystemAssertsMixin
from documents.tests.utils import SampleDirMixin
+from paperless.models import Correspondent
+from paperless.models import CustomField
+from paperless.models import CustomFieldInstance
+from paperless.models import Document
+from paperless.models import DocumentType
+from paperless.models import MatchingModel
+from paperless.models import StoragePath
+from paperless.models import Tag
+from paperless.models import Workflow
+from paperless.models import WorkflowAction
+from paperless.models import WorkflowActionEmail
+from paperless.models import WorkflowActionWebhook
+from paperless.models import WorkflowRun
+from paperless.models import WorkflowTrigger
from paperless_mail.models import MailAccount
from paperless_mail.models import MailRule
from documents.matching import match_document_types
from documents.matching import match_storage_paths
from documents.matching import match_tags
-from documents.models import Correspondent
-from documents.models import CustomField
-from documents.models import Document
-from documents.models import DocumentType
-from documents.models import Note
-from documents.models import PaperlessTask
-from documents.models import SavedView
-from documents.models import ShareLink
-from documents.models import StoragePath
-from documents.models import Tag
-from documents.models import UiSettings
-from documents.models import Workflow
-from documents.models import WorkflowAction
-from documents.models import WorkflowTrigger
from documents.parsers import get_parser_class_for_mime_type
from documents.parsers import parse_date_generator
from documents.permissions import PaperlessAdminPermissions
from paperless.celery import app as celery_app
from paperless.config import GeneralConfig
from paperless.db import GnuPG
+from paperless.models import Correspondent
+from paperless.models import CustomField
+from paperless.models import Document
+from paperless.models import DocumentType
+from paperless.models import Note
+from paperless.models import PaperlessTask
+from paperless.models import SavedView
+from paperless.models import ShareLink
+from paperless.models import StoragePath
+from paperless.models import Tag
+from paperless.models import UiSettings
+from paperless.models import Workflow
+from paperless.models import WorkflowAction
+from paperless.models import WorkflowTrigger
from paperless.serialisers import GroupSerializer
from paperless.serialisers import UserSerializer
from paperless.views import StandardPagination
from django.forms import ValidationError
from django.urls import reverse
-from documents.models import Document
+from paperless.models import Document
from paperless.signals import handle_social_account_updated
logger = logging.getLogger("paperless.auth")
-from django.core.validators import FileExtensionValidator
+import datetime
+from pathlib import Path
+from typing import Final
+
+import pathvalidate
+from celery import states
+from django.conf import settings
+from django.contrib.auth.models import Group
+from django.contrib.auth.models import User
+from django.core.validators import MaxValueValidator
from django.core.validators import MinValueValidator
from django.db import models
+from django.utils import timezone
from django.utils.translation import gettext_lazy as _
+from multiselectfield import MultiSelectField
+
+if settings.AUDIT_LOG_ENABLED:
+ from auditlog.registry import auditlog
+
+from django.core.validators import FileExtensionValidator
+from django.db.models import Case
+from django.db.models.functions import Cast
+from django.db.models.functions import Substr
+from django_softdelete.models import SoftDeleteModel
+
+from documents.data_models import DocumentSource
+from documents.parsers import get_default_file_extension
DEFAULT_SINGLETON_INSTANCE_ID = 1
+class ModelWithOwner(models.Model):
+ owner = models.ForeignKey(
+ User,
+ blank=True,
+ null=True,
+ default=None,
+ on_delete=models.SET_NULL,
+ verbose_name=_("owner"),
+ )
+
+ class Meta:
+ abstract = True
+
+
+class MatchingModel(ModelWithOwner):
+ MATCH_NONE = 0
+ MATCH_ANY = 1
+ MATCH_ALL = 2
+ MATCH_LITERAL = 3
+ MATCH_REGEX = 4
+ MATCH_FUZZY = 5
+ MATCH_AUTO = 6
+
+ MATCHING_ALGORITHMS = (
+ (MATCH_NONE, _("None")),
+ (MATCH_ANY, _("Any word")),
+ (MATCH_ALL, _("All words")),
+ (MATCH_LITERAL, _("Exact match")),
+ (MATCH_REGEX, _("Regular expression")),
+ (MATCH_FUZZY, _("Fuzzy word")),
+ (MATCH_AUTO, _("Automatic")),
+ )
+
+ name = models.CharField(_("name"), max_length=128)
+
+ match = models.CharField(_("match"), max_length=256, blank=True)
+
+ matching_algorithm = models.PositiveIntegerField(
+ _("matching algorithm"),
+ choices=MATCHING_ALGORITHMS,
+ default=MATCH_ANY,
+ )
+
+ is_insensitive = models.BooleanField(_("is insensitive"), default=True)
+
+ class Meta:
+ abstract = True
+ ordering = ("name",)
+ constraints = [
+ models.UniqueConstraint(
+ fields=["name", "owner"],
+ name="%(app_label)s_%(class)s_unique_name_owner",
+ ),
+ models.UniqueConstraint(
+ name="%(app_label)s_%(class)s_name_uniq",
+ fields=["name"],
+ condition=models.Q(owner__isnull=True),
+ ),
+ ]
+
+ def __str__(self):
+ return self.name
+
+
+class Correspondent(MatchingModel):
+ class Meta(MatchingModel.Meta):
+ app_label = "documents"
+ verbose_name = _("correspondent")
+ verbose_name_plural = _("correspondents")
+
+
+class Tag(MatchingModel):
+ color = models.CharField(_("color"), max_length=7, default="#a6cee3")
+
+ is_inbox_tag = models.BooleanField(
+ _("is inbox tag"),
+ default=False,
+ help_text=_(
+ "Marks this tag as an inbox tag: All newly consumed "
+ "documents will be tagged with inbox tags.",
+ ),
+ )
+
+ class Meta(MatchingModel.Meta):
+ app_label = "documents"
+ verbose_name = _("tag")
+ verbose_name_plural = _("tags")
+
+
+class DocumentType(MatchingModel):
+ class Meta(MatchingModel.Meta):
+ app_label = "documents"
+ verbose_name = _("document type")
+ verbose_name_plural = _("document types")
+
+
+class StoragePath(MatchingModel):
+ path = models.TextField(
+ _("path"),
+ )
+
+ class Meta(MatchingModel.Meta):
+ app_label = "documents"
+ verbose_name = _("storage path")
+ verbose_name_plural = _("storage paths")
+
+
+class Document(SoftDeleteModel, ModelWithOwner):
+ STORAGE_TYPE_UNENCRYPTED = "unencrypted"
+ STORAGE_TYPE_GPG = "gpg"
+ STORAGE_TYPES = (
+ (STORAGE_TYPE_UNENCRYPTED, _("Unencrypted")),
+ (STORAGE_TYPE_GPG, _("Encrypted with GNU Privacy Guard")),
+ )
+
+ correspondent = models.ForeignKey(
+ Correspondent,
+ blank=True,
+ null=True,
+ related_name="documents",
+ on_delete=models.SET_NULL,
+ verbose_name=_("correspondent"),
+ )
+
+ storage_path = models.ForeignKey(
+ StoragePath,
+ blank=True,
+ null=True,
+ related_name="documents",
+ on_delete=models.SET_NULL,
+ verbose_name=_("storage path"),
+ )
+
+ title = models.CharField(_("title"), max_length=128, blank=True, db_index=True)
+
+ document_type = models.ForeignKey(
+ DocumentType,
+ blank=True,
+ null=True,
+ related_name="documents",
+ on_delete=models.SET_NULL,
+ verbose_name=_("document type"),
+ )
+
+ content = models.TextField(
+ _("content"),
+ blank=True,
+ help_text=_(
+ "The raw, text-only data of the document. This field is "
+ "primarily used for searching.",
+ ),
+ )
+
+ mime_type = models.CharField(_("mime type"), max_length=256, editable=False)
+
+ tags = models.ManyToManyField(
+ Tag,
+ related_name="documents",
+ blank=True,
+ verbose_name=_("tags"),
+ )
+
+ checksum = models.CharField(
+ _("checksum"),
+ max_length=32,
+ editable=False,
+ unique=True,
+ help_text=_("The checksum of the original document."),
+ )
+
+ archive_checksum = models.CharField(
+ _("archive checksum"),
+ max_length=32,
+ editable=False,
+ blank=True,
+ null=True,
+ help_text=_("The checksum of the archived document."),
+ )
+
+ page_count = models.PositiveIntegerField(
+ _("page count"),
+ blank=False,
+ null=True,
+ unique=False,
+ db_index=False,
+ validators=[MinValueValidator(1)],
+ help_text=_(
+ "The number of pages of the document.",
+ ),
+ )
+
+ created = models.DateTimeField(_("created"), default=timezone.now, db_index=True)
+
+ modified = models.DateTimeField(
+ _("modified"),
+ auto_now=True,
+ editable=False,
+ db_index=True,
+ )
+
+ storage_type = models.CharField(
+ _("storage type"),
+ max_length=11,
+ choices=STORAGE_TYPES,
+ default=STORAGE_TYPE_UNENCRYPTED,
+ editable=False,
+ )
+
+ added = models.DateTimeField(
+ _("added"),
+ default=timezone.now,
+ editable=False,
+ db_index=True,
+ )
+
+ filename = models.FilePathField(
+ _("filename"),
+ max_length=1024,
+ editable=False,
+ default=None,
+ unique=True,
+ null=True,
+ help_text=_("Current filename in storage"),
+ )
+
+ archive_filename = models.FilePathField(
+ _("archive filename"),
+ max_length=1024,
+ editable=False,
+ default=None,
+ unique=True,
+ null=True,
+ help_text=_("Current archive filename in storage"),
+ )
+
+ original_filename = models.CharField(
+ _("original filename"),
+ max_length=1024,
+ editable=False,
+ default=None,
+ unique=False,
+ null=True,
+ help_text=_("The original name of the file when it was uploaded"),
+ )
+
+ ARCHIVE_SERIAL_NUMBER_MIN: Final[int] = 0
+ ARCHIVE_SERIAL_NUMBER_MAX: Final[int] = 0xFF_FF_FF_FF
+
+ archive_serial_number = models.PositiveIntegerField(
+ _("archive serial number"),
+ blank=True,
+ null=True,
+ unique=True,
+ db_index=True,
+ validators=[
+ MaxValueValidator(ARCHIVE_SERIAL_NUMBER_MAX),
+ MinValueValidator(ARCHIVE_SERIAL_NUMBER_MIN),
+ ],
+ help_text=_(
+ "The position of this document in your physical document archive.",
+ ),
+ )
+
+ class Meta:
+ app_label = "documents"
+ ordering = ("-created",)
+ verbose_name = _("document")
+ verbose_name_plural = _("documents")
+
+ def __str__(self) -> str:
+ # Convert UTC database time to local time
+ created = datetime.date.isoformat(timezone.localdate(self.created))
+
+ res = f"{created}"
+
+ if self.correspondent:
+ res += f" {self.correspondent}"
+ if self.title:
+ res += f" {self.title}"
+ return res
+
+ @property
+ def source_path(self) -> Path:
+ if self.filename:
+ fname = str(self.filename)
+ else:
+ fname = f"{self.pk:07}{self.file_type}"
+ if self.storage_type == self.STORAGE_TYPE_GPG:
+ fname += ".gpg" # pragma: no cover
+
+ return (settings.ORIGINALS_DIR / Path(fname)).resolve()
+
+ @property
+ def source_file(self):
+ return Path(self.source_path).open("rb")
+
+ @property
+ def has_archive_version(self) -> bool:
+ return self.archive_filename is not None
+
+ @property
+ def archive_path(self) -> Path | None:
+ if self.has_archive_version:
+ return (settings.ARCHIVE_DIR / Path(str(self.archive_filename))).resolve()
+ else:
+ return None
+
+ @property
+ def archive_file(self):
+ return Path(self.archive_path).open("rb")
+
+ def get_public_filename(self, *, archive=False, counter=0, suffix=None) -> str:
+ """
+ Returns a sanitized filename for the document, not including any paths.
+ """
+ result = str(self)
+
+ if counter:
+ result += f"_{counter:02}"
+
+ if suffix:
+ result += suffix
+
+ if archive:
+ result += ".pdf"
+ else:
+ result += self.file_type
+
+ return pathvalidate.sanitize_filename(result, replacement_text="-")
+
+ @property
+ def file_type(self):
+ return get_default_file_extension(self.mime_type)
+
+ @property
+ def thumbnail_path(self) -> Path:
+ webp_file_name = f"{self.pk:07}.webp"
+ if self.storage_type == self.STORAGE_TYPE_GPG:
+ webp_file_name += ".gpg"
+
+ webp_file_path = settings.THUMBNAIL_DIR / Path(webp_file_name)
+
+ return webp_file_path.resolve()
+
+ @property
+ def thumbnail_file(self):
+ return Path(self.thumbnail_path).open("rb")
+
+ @property
+ def created_date(self):
+ return timezone.localdate(self.created)
+
+
+class SavedView(ModelWithOwner):
+ class DisplayMode(models.TextChoices):
+ TABLE = ("table", _("Table"))
+ SMALL_CARDS = ("smallCards", _("Small Cards"))
+ LARGE_CARDS = ("largeCards", _("Large Cards"))
+
+ class DisplayFields(models.TextChoices):
+ TITLE = ("title", _("Title"))
+ CREATED = ("created", _("Created"))
+ ADDED = ("added", _("Added"))
+ TAGS = ("tag"), _("Tags")
+ CORRESPONDENT = ("correspondent", _("Correspondent"))
+ DOCUMENT_TYPE = ("documenttype", _("Document Type"))
+ STORAGE_PATH = ("storagepath", _("Storage Path"))
+ NOTES = ("note", _("Note"))
+ OWNER = ("owner", _("Owner"))
+ SHARED = ("shared", _("Shared"))
+ ASN = ("asn", _("ASN"))
+ PAGE_COUNT = ("pagecount", _("Pages"))
+ CUSTOM_FIELD = ("custom_field_%d", ("Custom Field"))
+
+ name = models.CharField(_("name"), max_length=128)
+
+ show_on_dashboard = models.BooleanField(
+ _("show on dashboard"),
+ )
+ show_in_sidebar = models.BooleanField(
+ _("show in sidebar"),
+ )
+
+ sort_field = models.CharField(
+ _("sort field"),
+ max_length=128,
+ null=True,
+ blank=True,
+ )
+ sort_reverse = models.BooleanField(_("sort reverse"), default=False)
+
+ page_size = models.PositiveIntegerField(
+ _("View page size"),
+ null=True,
+ blank=True,
+ validators=[MinValueValidator(1)],
+ )
+
+ display_mode = models.CharField(
+ max_length=128,
+ verbose_name=_("View display mode"),
+ choices=DisplayMode.choices,
+ null=True,
+ blank=True,
+ )
+
+ display_fields = models.JSONField(
+ verbose_name=_("Document display fields"),
+ null=True,
+ blank=True,
+ )
+
+ class Meta:
+ app_label = "documents"
+ ordering = ("name",)
+ verbose_name = _("saved view")
+ verbose_name_plural = _("saved views")
+
+ def __str__(self):
+ return f"SavedView {self.name}"
+
+
+class SavedViewFilterRule(models.Model):
+ RULE_TYPES = [
+ (0, _("title contains")),
+ (1, _("content contains")),
+ (2, _("ASN is")),
+ (3, _("correspondent is")),
+ (4, _("document type is")),
+ (5, _("is in inbox")),
+ (6, _("has tag")),
+ (7, _("has any tag")),
+ (8, _("created before")),
+ (9, _("created after")),
+ (10, _("created year is")),
+ (11, _("created month is")),
+ (12, _("created day is")),
+ (13, _("added before")),
+ (14, _("added after")),
+ (15, _("modified before")),
+ (16, _("modified after")),
+ (17, _("does not have tag")),
+ (18, _("does not have ASN")),
+ (19, _("title or content contains")),
+ (20, _("fulltext query")),
+ (21, _("more like this")),
+ (22, _("has tags in")),
+ (23, _("ASN greater than")),
+ (24, _("ASN less than")),
+ (25, _("storage path is")),
+ (26, _("has correspondent in")),
+ (27, _("does not have correspondent in")),
+ (28, _("has document type in")),
+ (29, _("does not have document type in")),
+ (30, _("has storage path in")),
+ (31, _("does not have storage path in")),
+ (32, _("owner is")),
+ (33, _("has owner in")),
+ (34, _("does not have owner")),
+ (35, _("does not have owner in")),
+ (36, _("has custom field value")),
+ (37, _("is shared by me")),
+ (38, _("has custom fields")),
+ (39, _("has custom field in")),
+ (40, _("does not have custom field in")),
+ (41, _("does not have custom field")),
+ (42, _("custom fields query")),
+ (43, _("created to")),
+ (44, _("created from")),
+ (45, _("added to")),
+ (46, _("added from")),
+ (47, _("mime type is")),
+ ]
+
+ saved_view = models.ForeignKey(
+ SavedView,
+ on_delete=models.CASCADE,
+ related_name="filter_rules",
+ verbose_name=_("saved view"),
+ )
+
+ rule_type = models.PositiveIntegerField(_("rule type"), choices=RULE_TYPES)
+
+ value = models.CharField(_("value"), max_length=255, blank=True, null=True)
+
+ class Meta:
+ app_label = "documents"
+ verbose_name = _("filter rule")
+ verbose_name_plural = _("filter rules")
+
+ def __str__(self) -> str:
+ return f"SavedViewFilterRule: {self.rule_type} : {self.value}"
+
+
+# Extending User Model Using a One-To-One Link
+class UiSettings(models.Model):
+ user = models.OneToOneField(
+ User,
+ on_delete=models.CASCADE,
+ related_name="ui_settings",
+ )
+ settings = models.JSONField(null=True)
+
+ class Meta:
+ app_label = "documents"
+
+ def __str__(self):
+ return self.user.username
+
+
+class PaperlessTask(ModelWithOwner):
+ ALL_STATES = sorted(states.ALL_STATES)
+ TASK_STATE_CHOICES = sorted(zip(ALL_STATES, ALL_STATES))
+
+ class TaskType(models.TextChoices):
+ AUTO = ("auto_task", _("Auto Task"))
+ SCHEDULED_TASK = ("scheduled_task", _("Scheduled Task"))
+ MANUAL_TASK = ("manual_task", _("Manual Task"))
+
+ class TaskName(models.TextChoices):
+ CONSUME_FILE = ("consume_file", _("Consume File"))
+ TRAIN_CLASSIFIER = ("train_classifier", _("Train Classifier"))
+ CHECK_SANITY = ("check_sanity", _("Check Sanity"))
+ INDEX_OPTIMIZE = ("index_optimize", _("Index Optimize"))
+
+ task_id = models.CharField(
+ max_length=255,
+ unique=True,
+ verbose_name=_("Task ID"),
+ help_text=_("Celery ID for the Task that was run"),
+ )
+
+ acknowledged = models.BooleanField(
+ default=False,
+ verbose_name=_("Acknowledged"),
+ help_text=_("If the task is acknowledged via the frontend or API"),
+ )
+
+ task_file_name = models.CharField(
+ null=True,
+ max_length=255,
+ verbose_name=_("Task Filename"),
+ help_text=_("Name of the file which the Task was run for"),
+ )
+
+ task_name = models.CharField(
+ null=True,
+ max_length=255,
+ choices=TaskName.choices,
+ verbose_name=_("Task Name"),
+ help_text=_("Name of the task that was run"),
+ )
+
+ status = models.CharField(
+ max_length=30,
+ default=states.PENDING,
+ choices=TASK_STATE_CHOICES,
+ verbose_name=_("Task State"),
+ help_text=_("Current state of the task being run"),
+ )
+
+ date_created = models.DateTimeField(
+ null=True,
+ default=timezone.now,
+ verbose_name=_("Created DateTime"),
+ help_text=_("Datetime field when the task result was created in UTC"),
+ )
+
+ date_started = models.DateTimeField(
+ null=True,
+ default=None,
+ verbose_name=_("Started DateTime"),
+ help_text=_("Datetime field when the task was started in UTC"),
+ )
+
+ date_done = models.DateTimeField(
+ null=True,
+ default=None,
+ verbose_name=_("Completed DateTime"),
+ help_text=_("Datetime field when the task was completed in UTC"),
+ )
+
+ result = models.TextField(
+ null=True,
+ default=None,
+ verbose_name=_("Result Data"),
+ help_text=_(
+ "The data returned by the task",
+ ),
+ )
+
+ type = models.CharField(
+ max_length=30,
+ choices=TaskType.choices,
+ default=TaskType.AUTO,
+ verbose_name=_("Task Type"),
+ help_text=_("The type of task that was run"),
+ )
+
+ class Meta:
+ app_label = "documents"
+
+ def __str__(self) -> str:
+ return f"Task {self.task_id}"
+
+
+class Note(SoftDeleteModel):
+ note = models.TextField(
+ _("content"),
+ blank=True,
+ help_text=_("Note for the document"),
+ )
+
+ created = models.DateTimeField(
+ _("created"),
+ default=timezone.now,
+ db_index=True,
+ )
+
+ document = models.ForeignKey(
+ Document,
+ blank=True,
+ null=True,
+ related_name="notes",
+ on_delete=models.CASCADE,
+ verbose_name=_("document"),
+ )
+
+ user = models.ForeignKey(
+ User,
+ blank=True,
+ null=True,
+ related_name="notes",
+ on_delete=models.SET_NULL,
+ verbose_name=_("user"),
+ )
+
+ class Meta:
+ app_label = "documents"
+ ordering = ("created",)
+ verbose_name = _("note")
+ verbose_name_plural = _("notes")
+
+ def __str__(self):
+ return self.note
+
+
+class ShareLink(SoftDeleteModel):
+ class FileVersion(models.TextChoices):
+ ARCHIVE = ("archive", _("Archive"))
+ ORIGINAL = ("original", _("Original"))
+
+ created = models.DateTimeField(
+ _("created"),
+ default=timezone.now,
+ db_index=True,
+ blank=True,
+ editable=False,
+ )
+
+ expiration = models.DateTimeField(
+ _("expiration"),
+ blank=True,
+ null=True,
+ db_index=True,
+ )
+
+ slug = models.SlugField(
+ _("slug"),
+ db_index=True,
+ unique=True,
+ blank=True,
+ editable=False,
+ )
+
+ document = models.ForeignKey(
+ Document,
+ blank=True,
+ related_name="share_links",
+ on_delete=models.CASCADE,
+ verbose_name=_("document"),
+ )
+
+ file_version = models.CharField(
+ max_length=50,
+ choices=FileVersion.choices,
+ default=FileVersion.ARCHIVE,
+ )
+
+ owner = models.ForeignKey(
+ User,
+ blank=True,
+ null=True,
+ related_name="share_links",
+ on_delete=models.SET_NULL,
+ verbose_name=_("owner"),
+ )
+
+ class Meta:
+ app_label = "documents"
+ ordering = ("created",)
+ verbose_name = _("share link")
+ verbose_name_plural = _("share links")
+
+ def __str__(self):
+ return f"Share Link for {self.document.title}"
+
+
+class CustomField(models.Model):
+ """
+ Defines the name and type of a custom field
+ """
+
+ class FieldDataType(models.TextChoices):
+ STRING = ("string", _("String"))
+ URL = ("url", _("URL"))
+ DATE = ("date", _("Date"))
+ BOOL = ("boolean"), _("Boolean")
+ INT = ("integer", _("Integer"))
+ FLOAT = ("float", _("Float"))
+ MONETARY = ("monetary", _("Monetary"))
+ DOCUMENTLINK = ("documentlink", _("Document Link"))
+ SELECT = ("select", _("Select"))
+
+ created = models.DateTimeField(
+ _("created"),
+ default=timezone.now,
+ db_index=True,
+ editable=False,
+ )
+
+ name = models.CharField(max_length=128)
+
+ data_type = models.CharField(
+ _("data type"),
+ max_length=50,
+ choices=FieldDataType.choices,
+ editable=False,
+ )
+
+ extra_data = models.JSONField(
+ _("extra data"),
+ null=True,
+ blank=True,
+ help_text=_(
+ "Extra data for the custom field, such as select options",
+ ),
+ )
+
+ class Meta:
+ app_label = "documents"
+ ordering = ("created",)
+ verbose_name = _("custom field")
+ verbose_name_plural = _("custom fields")
+ constraints = [
+ models.UniqueConstraint(
+ fields=["name"],
+ name="%(app_label)s_%(class)s_unique_name",
+ ),
+ ]
+
+ def __str__(self) -> str:
+ return f"{self.name} : {self.data_type}"
+
+
+class CustomFieldInstance(SoftDeleteModel):
+ """
+ A single instance of a field, attached to a CustomField for the name and type
+ and attached to a single Document to be metadata for it
+ """
+
+ TYPE_TO_DATA_STORE_NAME_MAP = {
+ CustomField.FieldDataType.STRING: "value_text",
+ CustomField.FieldDataType.URL: "value_url",
+ CustomField.FieldDataType.DATE: "value_date",
+ CustomField.FieldDataType.BOOL: "value_bool",
+ CustomField.FieldDataType.INT: "value_int",
+ CustomField.FieldDataType.FLOAT: "value_float",
+ CustomField.FieldDataType.MONETARY: "value_monetary",
+ CustomField.FieldDataType.DOCUMENTLINK: "value_document_ids",
+ CustomField.FieldDataType.SELECT: "value_select",
+ }
+
+ created = models.DateTimeField(
+ _("created"),
+ default=timezone.now,
+ db_index=True,
+ editable=False,
+ )
+
+ document = models.ForeignKey(
+ Document,
+ blank=False,
+ null=False,
+ on_delete=models.CASCADE,
+ related_name="custom_fields",
+ editable=False,
+ )
+
+ field = models.ForeignKey(
+ CustomField,
+ blank=False,
+ null=False,
+ on_delete=models.CASCADE,
+ related_name="fields",
+ editable=False,
+ )
+
+ # Actual data storage
+ value_text = models.CharField(max_length=128, null=True)
+
+ value_bool = models.BooleanField(null=True)
+
+ value_url = models.URLField(null=True)
+
+ value_date = models.DateField(null=True)
+
+ value_int = models.IntegerField(null=True)
+
+ value_float = models.FloatField(null=True)
+
+ value_monetary = models.CharField(null=True, max_length=128)
+
+ value_monetary_amount = models.GeneratedField(
+ expression=Case(
+ # If the value starts with a number and no currency symbol, use the whole string
+ models.When(
+ value_monetary__regex=r"^\d+",
+ then=Cast(
+ Substr("value_monetary", 1),
+ output_field=models.DecimalField(decimal_places=2, max_digits=65),
+ ),
+ ),
+ # If the value starts with a 3-char currency symbol, use the rest of the string
+ default=Cast(
+ Substr("value_monetary", 4),
+ output_field=models.DecimalField(decimal_places=2, max_digits=65),
+ ),
+ output_field=models.DecimalField(decimal_places=2, max_digits=65),
+ ),
+ output_field=models.DecimalField(decimal_places=2, max_digits=65),
+ db_persist=True,
+ )
+
+ value_document_ids = models.JSONField(null=True)
+
+ value_select = models.CharField(null=True, max_length=16)
+
+ class Meta:
+ app_label = "documents"
+ ordering = ("created",)
+ verbose_name = _("custom field instance")
+ verbose_name_plural = _("custom field instances")
+ constraints = [
+ models.UniqueConstraint(
+ fields=["document", "field"],
+ name="%(app_label)s_%(class)s_unique_document_field",
+ ),
+ ]
+
+ def __str__(self) -> str:
+ value = (
+ next(
+ option.get("label")
+ for option in self.field.extra_data["select_options"]
+ if option.get("id") == self.value_select
+ )
+ if (
+ self.field.data_type == CustomField.FieldDataType.SELECT
+ and self.value_select is not None
+ )
+ else self.value
+ )
+ return str(self.field.name) + f" : {value}"
+
+ @classmethod
+ def get_value_field_name(cls, data_type: CustomField.FieldDataType):
+ try:
+ return cls.TYPE_TO_DATA_STORE_NAME_MAP[data_type]
+ except KeyError: # pragma: no cover
+ raise NotImplementedError(data_type)
+
+ @property
+ def value(self):
+ """
+ Based on the data type, access the actual value the instance stores
+ A little shorthand/quick way to get what is actually here
+ """
+ value_field_name = self.get_value_field_name(self.field.data_type)
+ return getattr(self, value_field_name)
+
+
+if settings.AUDIT_LOG_ENABLED:
+ auditlog.register(
+ Document,
+ m2m_fields={"tags"},
+ exclude_fields=["modified"],
+ )
+ auditlog.register(Correspondent)
+ auditlog.register(Tag)
+ auditlog.register(DocumentType)
+ auditlog.register(Note)
+ auditlog.register(CustomField)
+ auditlog.register(CustomFieldInstance)
+
+
+class WorkflowTrigger(models.Model):
+ class WorkflowTriggerMatching(models.IntegerChoices):
+ # No auto matching
+ NONE = MatchingModel.MATCH_NONE, _("None")
+ ANY = MatchingModel.MATCH_ANY, _("Any word")
+ ALL = MatchingModel.MATCH_ALL, _("All words")
+ LITERAL = MatchingModel.MATCH_LITERAL, _("Exact match")
+ REGEX = MatchingModel.MATCH_REGEX, _("Regular expression")
+ FUZZY = MatchingModel.MATCH_FUZZY, _("Fuzzy word")
+
+ class WorkflowTriggerType(models.IntegerChoices):
+ CONSUMPTION = 1, _("Consumption Started")
+ DOCUMENT_ADDED = 2, _("Document Added")
+ DOCUMENT_UPDATED = 3, _("Document Updated")
+ SCHEDULED = 4, _("Scheduled")
+
+ class DocumentSourceChoices(models.IntegerChoices):
+ CONSUME_FOLDER = DocumentSource.ConsumeFolder.value, _("Consume Folder")
+ API_UPLOAD = DocumentSource.ApiUpload.value, _("Api Upload")
+ MAIL_FETCH = DocumentSource.MailFetch.value, _("Mail Fetch")
+ WEB_UI = DocumentSource.WebUI.value, _("Web UI")
+
+ class ScheduleDateField(models.TextChoices):
+ ADDED = "added", _("Added")
+ CREATED = "created", _("Created")
+ MODIFIED = "modified", _("Modified")
+ CUSTOM_FIELD = "custom_field", _("Custom Field")
+
+ type = models.PositiveIntegerField(
+ _("Workflow Trigger Type"),
+ choices=WorkflowTriggerType.choices,
+ default=WorkflowTriggerType.CONSUMPTION,
+ )
+
+ sources = MultiSelectField(
+ max_length=7,
+ choices=DocumentSourceChoices.choices,
+ default=f"{DocumentSource.ConsumeFolder},{DocumentSource.ApiUpload},{DocumentSource.MailFetch},{DocumentSource.WebUI}",
+ )
+
+ filter_path = models.CharField(
+ _("filter path"),
+ max_length=256,
+ null=True,
+ blank=True,
+ help_text=_(
+ "Only consume documents with a path that matches "
+ "this if specified. Wildcards specified as * are "
+ "allowed. Case insensitive.",
+ ),
+ )
+
+ filter_filename = models.CharField(
+ _("filter filename"),
+ max_length=256,
+ null=True,
+ blank=True,
+ help_text=_(
+ "Only consume documents which entirely match this "
+ "filename if specified. Wildcards such as *.pdf or "
+ "*invoice* are allowed. Case insensitive.",
+ ),
+ )
+
+ filter_mailrule = models.ForeignKey(
+ "paperless_mail.MailRule",
+ null=True,
+ blank=True,
+ on_delete=models.SET_NULL,
+ verbose_name=_("filter documents from this mail rule"),
+ )
+
+ match = models.CharField(_("match"), max_length=256, blank=True)
+
+ matching_algorithm = models.PositiveIntegerField(
+ _("matching algorithm"),
+ choices=WorkflowTriggerMatching.choices,
+ default=WorkflowTriggerMatching.NONE,
+ )
+
+ is_insensitive = models.BooleanField(_("is insensitive"), default=True)
+
+ filter_has_tags = models.ManyToManyField(
+ Tag,
+ blank=True,
+ verbose_name=_("has these tag(s)"),
+ )
+
+ filter_has_document_type = models.ForeignKey(
+ DocumentType,
+ null=True,
+ blank=True,
+ on_delete=models.SET_NULL,
+ verbose_name=_("has this document type"),
+ )
+
+ filter_has_correspondent = models.ForeignKey(
+ Correspondent,
+ null=True,
+ blank=True,
+ on_delete=models.SET_NULL,
+ verbose_name=_("has this correspondent"),
+ )
+
+ schedule_offset_days = models.PositiveIntegerField(
+ _("schedule offset days"),
+ default=0,
+ help_text=_(
+ "The number of days to offset the schedule trigger by.",
+ ),
+ )
+
+ schedule_is_recurring = models.BooleanField(
+ _("schedule is recurring"),
+ default=False,
+ help_text=_(
+ "If the schedule should be recurring.",
+ ),
+ )
+
+ schedule_recurring_interval_days = models.PositiveIntegerField(
+ _("schedule recurring delay in days"),
+ default=1,
+ validators=[MinValueValidator(1)],
+ help_text=_(
+ "The number of days between recurring schedule triggers.",
+ ),
+ )
+
+ schedule_date_field = models.CharField(
+ _("schedule date field"),
+ max_length=20,
+ choices=ScheduleDateField.choices,
+ default=ScheduleDateField.ADDED,
+ help_text=_(
+ "The field to check for a schedule trigger.",
+ ),
+ )
+
+ schedule_date_custom_field = models.ForeignKey(
+ CustomField,
+ null=True,
+ blank=True,
+ on_delete=models.SET_NULL,
+ verbose_name=_("schedule date custom field"),
+ )
+
+ class Meta:
+ app_label = "documents"
+ verbose_name = _("workflow trigger")
+ verbose_name_plural = _("workflow triggers")
+
+ def __str__(self):
+ return f"WorkflowTrigger {self.pk}"
+
+
+class WorkflowActionEmail(models.Model):
+ subject = models.CharField(
+ _("email subject"),
+ max_length=256,
+ null=False,
+ help_text=_(
+ "The subject of the email, can include some placeholders, "
+ "see documentation.",
+ ),
+ )
+
+ body = models.TextField(
+ _("email body"),
+ null=False,
+ help_text=_(
+ "The body (message) of the email, can include some placeholders, "
+ "see documentation.",
+ ),
+ )
+
+ to = models.TextField(
+ _("emails to"),
+ null=False,
+ help_text=_(
+ "The destination email addresses, comma separated.",
+ ),
+ )
+
+ include_document = models.BooleanField(
+ default=False,
+ verbose_name=_("include document in email"),
+ )
+
+ class Meta:
+ app_label = "documents"
+
+ def __str__(self):
+ return f"Workflow Email Action {self.pk}"
+
+
+class WorkflowActionWebhook(models.Model):
+ # We dont use the built-in URLField because it is not flexible enough
+ # validation is handled in the serializer
+ url = models.CharField(
+ _("webhook url"),
+ null=False,
+ max_length=256,
+ help_text=_("The destination URL for the notification."),
+ )
+
+ use_params = models.BooleanField(
+ default=True,
+ verbose_name=_("use parameters"),
+ )
+
+ as_json = models.BooleanField(
+ default=False,
+ verbose_name=_("send as JSON"),
+ )
+
+ params = models.JSONField(
+ _("webhook parameters"),
+ null=True,
+ blank=True,
+ help_text=_("The parameters to send with the webhook URL if body not used."),
+ )
+
+ body = models.TextField(
+ _("webhook body"),
+ null=True,
+ blank=True,
+ help_text=_("The body to send with the webhook URL if parameters not used."),
+ )
+
+ headers = models.JSONField(
+ _("webhook headers"),
+ null=True,
+ blank=True,
+ help_text=_("The headers to send with the webhook URL."),
+ )
+
+ include_document = models.BooleanField(
+ default=False,
+ verbose_name=_("include document in webhook"),
+ )
+
+ class Meta:
+ app_label = "documents"
+
+ def __str__(self):
+ return f"Workflow Webhook Action {self.pk}"
+
+
+class WorkflowAction(models.Model):
+ class WorkflowActionType(models.IntegerChoices):
+ ASSIGNMENT = (
+ 1,
+ _("Assignment"),
+ )
+ REMOVAL = (
+ 2,
+ _("Removal"),
+ )
+ EMAIL = (
+ 3,
+ _("Email"),
+ )
+ WEBHOOK = (
+ 4,
+ _("Webhook"),
+ )
+
+ type = models.PositiveIntegerField(
+ _("Workflow Action Type"),
+ choices=WorkflowActionType.choices,
+ default=WorkflowActionType.ASSIGNMENT,
+ )
+
+ assign_title = models.CharField(
+ _("assign title"),
+ max_length=256,
+ null=True,
+ blank=True,
+ help_text=_(
+ "Assign a document title, can include some placeholders, "
+ "see documentation.",
+ ),
+ )
+
+ assign_tags = models.ManyToManyField(
+ Tag,
+ blank=True,
+ related_name="+",
+ verbose_name=_("assign this tag"),
+ )
+
+ assign_document_type = models.ForeignKey(
+ DocumentType,
+ null=True,
+ blank=True,
+ on_delete=models.SET_NULL,
+ related_name="+",
+ verbose_name=_("assign this document type"),
+ )
+
+ assign_correspondent = models.ForeignKey(
+ Correspondent,
+ null=True,
+ blank=True,
+ on_delete=models.SET_NULL,
+ related_name="+",
+ verbose_name=_("assign this correspondent"),
+ )
+
+ assign_storage_path = models.ForeignKey(
+ StoragePath,
+ null=True,
+ blank=True,
+ on_delete=models.SET_NULL,
+ related_name="+",
+ verbose_name=_("assign this storage path"),
+ )
+
+ assign_owner = models.ForeignKey(
+ User,
+ null=True,
+ blank=True,
+ on_delete=models.SET_NULL,
+ related_name="+",
+ verbose_name=_("assign this owner"),
+ )
+
+ assign_view_users = models.ManyToManyField(
+ User,
+ blank=True,
+ related_name="+",
+ verbose_name=_("grant view permissions to these users"),
+ )
+
+ assign_view_groups = models.ManyToManyField(
+ Group,
+ blank=True,
+ related_name="+",
+ verbose_name=_("grant view permissions to these groups"),
+ )
+
+ assign_change_users = models.ManyToManyField(
+ User,
+ blank=True,
+ related_name="+",
+ verbose_name=_("grant change permissions to these users"),
+ )
+
+ assign_change_groups = models.ManyToManyField(
+ Group,
+ blank=True,
+ related_name="+",
+ verbose_name=_("grant change permissions to these groups"),
+ )
+
+ assign_custom_fields = models.ManyToManyField(
+ CustomField,
+ blank=True,
+ related_name="+",
+ verbose_name=_("assign these custom fields"),
+ )
+
+ assign_custom_fields_values = models.JSONField(
+ _("custom field values"),
+ null=True,
+ blank=True,
+ help_text=_(
+ "Optional values to assign to the custom fields.",
+ ),
+ default=dict,
+ )
+
+ remove_tags = models.ManyToManyField(
+ Tag,
+ blank=True,
+ related_name="+",
+ verbose_name=_("remove these tag(s)"),
+ )
+
+ remove_all_tags = models.BooleanField(
+ default=False,
+ verbose_name=_("remove all tags"),
+ )
+
+ remove_document_types = models.ManyToManyField(
+ DocumentType,
+ blank=True,
+ related_name="+",
+ verbose_name=_("remove these document type(s)"),
+ )
+
+ remove_all_document_types = models.BooleanField(
+ default=False,
+ verbose_name=_("remove all document types"),
+ )
+
+ remove_correspondents = models.ManyToManyField(
+ Correspondent,
+ blank=True,
+ related_name="+",
+ verbose_name=_("remove these correspondent(s)"),
+ )
+
+ remove_all_correspondents = models.BooleanField(
+ default=False,
+ verbose_name=_("remove all correspondents"),
+ )
+
+ remove_storage_paths = models.ManyToManyField(
+ StoragePath,
+ blank=True,
+ related_name="+",
+ verbose_name=_("remove these storage path(s)"),
+ )
+
+ remove_all_storage_paths = models.BooleanField(
+ default=False,
+ verbose_name=_("remove all storage paths"),
+ )
+
+ remove_owners = models.ManyToManyField(
+ User,
+ blank=True,
+ related_name="+",
+ verbose_name=_("remove these owner(s)"),
+ )
+
+ remove_all_owners = models.BooleanField(
+ default=False,
+ verbose_name=_("remove all owners"),
+ )
+
+ remove_view_users = models.ManyToManyField(
+ User,
+ blank=True,
+ related_name="+",
+ verbose_name=_("remove view permissions for these users"),
+ )
+
+ remove_view_groups = models.ManyToManyField(
+ Group,
+ blank=True,
+ related_name="+",
+ verbose_name=_("remove view permissions for these groups"),
+ )
+
+ remove_change_users = models.ManyToManyField(
+ User,
+ blank=True,
+ related_name="+",
+ verbose_name=_("remove change permissions for these users"),
+ )
+
+ remove_change_groups = models.ManyToManyField(
+ Group,
+ blank=True,
+ related_name="+",
+ verbose_name=_("remove change permissions for these groups"),
+ )
+
+ remove_all_permissions = models.BooleanField(
+ default=False,
+ verbose_name=_("remove all permissions"),
+ )
+
+ remove_custom_fields = models.ManyToManyField(
+ CustomField,
+ blank=True,
+ related_name="+",
+ verbose_name=_("remove these custom fields"),
+ )
+
+ remove_all_custom_fields = models.BooleanField(
+ default=False,
+ verbose_name=_("remove all custom fields"),
+ )
+
+ email = models.ForeignKey(
+ WorkflowActionEmail,
+ null=True,
+ blank=True,
+ on_delete=models.SET_NULL,
+ related_name="action",
+ verbose_name=_("email"),
+ )
+
+ webhook = models.ForeignKey(
+ WorkflowActionWebhook,
+ null=True,
+ blank=True,
+ on_delete=models.SET_NULL,
+ related_name="action",
+ verbose_name=_("webhook"),
+ )
+
+ class Meta:
+ app_label = "documents"
+ verbose_name = _("workflow action")
+ verbose_name_plural = _("workflow actions")
+
+ def __str__(self):
+ return f"WorkflowAction {self.pk}"
+
+
+class Workflow(models.Model):
+ name = models.CharField(_("name"), max_length=256, unique=True)
+
+ order = models.IntegerField(_("order"), default=0)
+
+ triggers = models.ManyToManyField(
+ WorkflowTrigger,
+ related_name="workflows",
+ blank=False,
+ verbose_name=_("triggers"),
+ )
+
+ actions = models.ManyToManyField(
+ WorkflowAction,
+ related_name="workflows",
+ blank=False,
+ verbose_name=_("actions"),
+ )
+
+ enabled = models.BooleanField(_("enabled"), default=True)
+
+ class Meta:
+ app_label = "documents"
+
+ def __str__(self):
+ return f"Workflow: {self.name}"
+
+
+class WorkflowRun(models.Model):
+ workflow = models.ForeignKey(
+ Workflow,
+ on_delete=models.CASCADE,
+ related_name="runs",
+ verbose_name=_("workflow"),
+ )
+
+ type = models.PositiveIntegerField(
+ _("workflow trigger type"),
+ choices=WorkflowTrigger.WorkflowTriggerType.choices,
+ null=True,
+ )
+
+ document = models.ForeignKey(
+ Document,
+ null=True,
+ on_delete=models.CASCADE,
+ related_name="workflow_runs",
+ verbose_name=_("document"),
+ )
+
+ run_at = models.DateTimeField(
+ _("date run"),
+ default=timezone.now,
+ db_index=True,
+ )
+
+ class Meta:
+ app_label = "documents"
+ verbose_name = _("workflow run")
+ verbose_name_plural = _("workflow runs")
+
+ def __str__(self):
+ return f"WorkflowRun of {self.workflow} at {self.run_at} on {self.document}"
+
+
class AbstractSingletonModel(models.Model):
class Meta:
abstract = True
from documents.data_models import DocumentMetadataOverrides
from documents.data_models import DocumentSource
from documents.loggers import LoggingMixin
-from documents.models import Correspondent
from documents.parsers import is_mime_type_supported
from documents.tasks import consume_file
+from paperless.models import Correspondent
from paperless_mail.models import MailAccount
from paperless_mail.models import MailRule
from paperless_mail.models import ProcessedMail
from django.utils import timezone
from django.utils.translation import gettext_lazy as _
-import documents.models as document_models
+import paperless.models as document_models
class MailAccount(document_models.ModelWithOwner):
from rest_framework import status
from rest_framework.test import APITestCase
-from documents.models import Correspondent
-from documents.models import DocumentType
-from documents.models import Tag
from documents.tests.utils import DirectoriesMixin
+from paperless.models import Correspondent
+from paperless.models import DocumentType
+from paperless.models import Tag
from paperless_mail.models import MailAccount
from paperless_mail.models import MailRule
from paperless_mail.tests.test_mail import BogusMailBox
from rest_framework import status
from rest_framework.test import APITestCase
-from documents.models import Correspondent
from documents.tests.utils import DirectoriesMixin
from documents.tests.utils import FileSystemAssertsMixin
+from paperless.models import Correspondent
from paperless_mail import tasks
from paperless_mail.mail import MailAccountHandler
from paperless_mail.mail import MailError
projects / thirdparty / paperless-ngx.git / commitdiff
