mirror of
https://github.com/paperless-ngx/paperless-ngx.git
synced 2026-10-31 03:55:43 +00:00
Compare commits
3
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
c40849e9bb | ||
|
|
f4358cef53 | ||
|
|
2cb04b0b8b |
No files matched your search
@@ -3,7 +3,6 @@ from __future__ import annotations
|
|||||||
import logging
|
import logging
|
||||||
import re
|
import re
|
||||||
from fnmatch import fnmatch
|
from fnmatch import fnmatch
|
||||||
from fnmatch import translate as fnmatch_translate
|
|
||||||
from typing import TYPE_CHECKING
|
from typing import TYPE_CHECKING
|
||||||
|
|
||||||
from rest_framework import serializers
|
from rest_framework import serializers
|
||||||
@@ -548,6 +547,48 @@ def existing_document_matches_workflow(
|
|||||||
return (True, None)
|
return (True, None)
|
||||||
|
|
||||||
|
|
||||||
|
def glob_to_database_regex(pattern: str) -> str:
|
||||||
|
"""
|
||||||
|
Translates an fnmatch style glob into a regular expression which the
|
||||||
|
database can evaluate, matching the whole value.
|
||||||
|
|
||||||
|
fnmatch.translate is not usable here: it emits syntax such as (?s:...),
|
||||||
|
atomic groups and named group backreferences, which PostgreSQL rejects.
|
||||||
|
Only constructs common to Python, PostgreSQL and MariaDB are produced.
|
||||||
|
"""
|
||||||
|
parts: list[str] = []
|
||||||
|
i = 0
|
||||||
|
n = len(pattern)
|
||||||
|
while i < n:
|
||||||
|
char = pattern[i]
|
||||||
|
i += 1
|
||||||
|
if char == "*":
|
||||||
|
parts.append(".*")
|
||||||
|
elif char == "?":
|
||||||
|
parts.append(".")
|
||||||
|
elif char == "[":
|
||||||
|
j = i
|
||||||
|
if j < n and pattern[j] == "!":
|
||||||
|
j += 1
|
||||||
|
if j < n and pattern[j] == "]":
|
||||||
|
j += 1
|
||||||
|
while j < n and pattern[j] != "]":
|
||||||
|
j += 1
|
||||||
|
if j >= n:
|
||||||
|
parts.append(r"\[")
|
||||||
|
else:
|
||||||
|
stuff = pattern[i:j].replace("\\", r"\\").replace("[", r"\[")
|
||||||
|
i = j + 1
|
||||||
|
if stuff.startswith("!"):
|
||||||
|
stuff = "^" + stuff[1:]
|
||||||
|
elif stuff.startswith("^"):
|
||||||
|
stuff = "\\" + stuff
|
||||||
|
parts.append(f"[{stuff}]")
|
||||||
|
else:
|
||||||
|
parts.append(re.escape(char))
|
||||||
|
return "^" + "".join(parts) + "$"
|
||||||
|
|
||||||
|
|
||||||
def prefilter_documents_by_workflowtrigger(
|
def prefilter_documents_by_workflowtrigger(
|
||||||
documents: QuerySet[Document],
|
documents: QuerySet[Document],
|
||||||
trigger: WorkflowTrigger,
|
trigger: WorkflowTrigger,
|
||||||
@@ -628,8 +669,9 @@ def prefilter_documents_by_workflowtrigger(
|
|||||||
documents = documents.annotate(**annotations).filter(custom_field_q)
|
documents = documents.annotate(**annotations).filter(custom_field_q)
|
||||||
|
|
||||||
if trigger.filter_filename:
|
if trigger.filter_filename:
|
||||||
regex = fnmatch_translate(trigger.filter_filename).lstrip("^").rstrip("$")
|
documents = documents.filter(
|
||||||
documents = documents.filter(original_filename__iregex=regex)
|
original_filename__iregex=glob_to_database_regex(trigger.filter_filename),
|
||||||
|
)
|
||||||
|
|
||||||
return documents
|
return documents
|
||||||
|
|
||||||
|
|||||||
@@ -6,6 +6,7 @@ from unittest.mock import patch
|
|||||||
from urllib.parse import quote
|
from urllib.parse import quote
|
||||||
|
|
||||||
import time_machine
|
import time_machine
|
||||||
|
from django.db import connection
|
||||||
from django.test import override_settings
|
from django.test import override_settings
|
||||||
from django.utils import timezone
|
from django.utils import timezone
|
||||||
from rest_framework.test import APITestCase
|
from rest_framework.test import APITestCase
|
||||||
@@ -20,6 +21,13 @@ from paperless_testing.dirs import DirectoriesMixin
|
|||||||
from paperless_testing.factories import UserFactory
|
from paperless_testing.factories import UserFactory
|
||||||
|
|
||||||
|
|
||||||
|
def _exact_equal(value: str | None, expected: str) -> bool:
|
||||||
|
"""`exact` follows the column collation, which is case insensitive on MariaDB."""
|
||||||
|
if connection.vendor == "mysql" and value is not None:
|
||||||
|
return value.casefold() == expected.casefold()
|
||||||
|
return value == expected
|
||||||
|
|
||||||
|
|
||||||
class DocumentWrapper:
|
class DocumentWrapper:
|
||||||
"""
|
"""
|
||||||
Allows Pythonic access to the custom fields associated with the wrapped document.
|
Allows Pythonic access to the custom fields associated with the wrapped document.
|
||||||
@@ -287,7 +295,8 @@ class TestCustomFieldsSearch(DirectoriesMixin, APITestCase):
|
|||||||
self._assert_query_match_predicate(
|
self._assert_query_match_predicate(
|
||||||
["string_field", "exact", "paperless"],
|
["string_field", "exact", "paperless"],
|
||||||
lambda document: (
|
lambda document: (
|
||||||
"string_field" in document and document["string_field"] == "paperless"
|
"string_field" in document
|
||||||
|
and _exact_equal(document["string_field"], "paperless")
|
||||||
),
|
),
|
||||||
)
|
)
|
||||||
|
|
||||||
@@ -296,7 +305,10 @@ class TestCustomFieldsSearch(DirectoriesMixin, APITestCase):
|
|||||||
["string_field", "in", ["paperless", "Paperless"]],
|
["string_field", "in", ["paperless", "Paperless"]],
|
||||||
lambda document: (
|
lambda document: (
|
||||||
"string_field" in document
|
"string_field" in document
|
||||||
and document["string_field"] in ("paperless", "Paperless")
|
and any(
|
||||||
|
_exact_equal(document["string_field"], expected)
|
||||||
|
for expected in ("paperless", "Paperless")
|
||||||
|
)
|
||||||
),
|
),
|
||||||
)
|
)
|
||||||
|
|
||||||
@@ -761,7 +773,7 @@ class TestCustomFieldsSearch(DirectoriesMixin, APITestCase):
|
|||||||
lambda document: (
|
lambda document: (
|
||||||
not (
|
not (
|
||||||
"string_field" in document
|
"string_field" in document
|
||||||
and document["string_field"] == "paperless"
|
and _exact_equal(document["string_field"], "paperless")
|
||||||
)
|
)
|
||||||
),
|
),
|
||||||
)
|
)
|
||||||
|
|||||||
@@ -242,9 +242,7 @@ class TestDocumentListEffectiveContentAnnotation:
|
|||||||
# Only the list query itself reads a content column: no extra query
|
# Only the list query itself reads a content column: no extra query
|
||||||
# for the skipped prefetch, none for a per-instance fallback
|
# for the skipped prefetch, none for a per-instance fallback
|
||||||
content_queries = [
|
content_queries = [
|
||||||
query
|
query for query in ctx.captured_queries if _content_column() in query["sql"]
|
||||||
for query in ctx.captured_queries
|
|
||||||
if '"documents_document"."content"' in query["sql"]
|
|
||||||
]
|
]
|
||||||
assert len(content_queries) == 1
|
assert len(content_queries) == 1
|
||||||
|
|
||||||
@@ -365,6 +363,11 @@ def _get_document_queries(
|
|||||||
return [q for q in ctx.captured_queries if '"django_content_type"' not in q["sql"]]
|
return [q for q in ctx.captured_queries if '"django_content_type"' not in q["sql"]]
|
||||||
|
|
||||||
|
|
||||||
|
def _content_column() -> str:
|
||||||
|
quote = connection.ops.quote_name
|
||||||
|
return f"{quote('documents_document')}.{quote('content')}"
|
||||||
|
|
||||||
|
|
||||||
def _get_effective_content_fallback_queries(
|
def _get_effective_content_fallback_queries(
|
||||||
ctx: CaptureQueriesContext,
|
ctx: CaptureQueriesContext,
|
||||||
) -> list[dict[str, str]]:
|
) -> list[dict[str, str]]:
|
||||||
@@ -379,7 +382,7 @@ def _get_effective_content_fallback_queries(
|
|||||||
return [
|
return [
|
||||||
q
|
q
|
||||||
for q in ctx.captured_queries
|
for q in ctx.captured_queries
|
||||||
if q["sql"].startswith('SELECT "documents_document"."content" FROM')
|
if q["sql"].startswith(f"SELECT {_content_column()} FROM")
|
||||||
]
|
]
|
||||||
|
|
||||||
|
|
||||||
|
|||||||
@@ -0,0 +1,68 @@
|
|||||||
|
import re
|
||||||
|
from fnmatch import fnmatch
|
||||||
|
|
||||||
|
import pytest
|
||||||
|
|
||||||
|
from documents.matching import glob_to_database_regex
|
||||||
|
from documents.models import Document
|
||||||
|
from paperless_testing.factories import DocumentFactory
|
||||||
|
|
||||||
|
NAMES = [
|
||||||
|
"invoice.pdf",
|
||||||
|
"Invoice.PDF",
|
||||||
|
"invoice_2024.pdf",
|
||||||
|
"a.b.c",
|
||||||
|
"a-b",
|
||||||
|
"sample [1].pdf",
|
||||||
|
"sample 1.pdf",
|
||||||
|
"x^y",
|
||||||
|
"back\\slash",
|
||||||
|
"dir/file.pdf",
|
||||||
|
"",
|
||||||
|
]
|
||||||
|
|
||||||
|
PATTERNS = [
|
||||||
|
"*",
|
||||||
|
"invoice.pdf",
|
||||||
|
"*.pdf",
|
||||||
|
"*invoice*pdf",
|
||||||
|
"*a*b*c*",
|
||||||
|
"invoice_????.pdf",
|
||||||
|
"sample [[]1].pdf",
|
||||||
|
"sample [1-2].pdf",
|
||||||
|
"sample [!2].pdf",
|
||||||
|
"sample [^2].pdf",
|
||||||
|
"sample [",
|
||||||
|
"a.b.c",
|
||||||
|
"x^y",
|
||||||
|
"back\\slash",
|
||||||
|
"[]]",
|
||||||
|
]
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.mark.parametrize("pattern", PATTERNS)
|
||||||
|
def test_regex_agrees_with_fnmatch(pattern: str) -> None:
|
||||||
|
regex = glob_to_database_regex(pattern)
|
||||||
|
for name in NAMES:
|
||||||
|
expected = fnmatch(name.lower(), pattern.lower())
|
||||||
|
actual = re.search(regex, name, re.IGNORECASE) is not None
|
||||||
|
assert actual is expected, f"{pattern!r} vs {name!r}"
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.mark.parametrize("pattern", ["*", "*a*b*c*", "a*b*c*d*e"])
|
||||||
|
def test_regex_avoids_syntax_postgresql_rejects(pattern: str) -> None:
|
||||||
|
regex = glob_to_database_regex(pattern)
|
||||||
|
assert not any(marker in regex for marker in ("(?", "\\Z"))
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.mark.django_db
|
||||||
|
def test_database_filter_matches_glob() -> None:
|
||||||
|
matching = DocumentFactory(original_filename="Invoice_2024.PDF")
|
||||||
|
DocumentFactory(original_filename="invoice_2024.txt")
|
||||||
|
DocumentFactory(original_filename="receipt.pdf")
|
||||||
|
|
||||||
|
found = Document.objects.filter(
|
||||||
|
original_filename__iregex=glob_to_database_regex("*invoice*.pdf"),
|
||||||
|
)
|
||||||
|
|
||||||
|
assert list(found) == [matching]
|
||||||
Reference in new issue
Block a user