mirror of
https://github.com/paperless-ngx/paperless-ngx.git
synced 2026-10-31 00:35:19 +00:00
Compare commits
1
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
5be95f8fcc |
No files matched your search
@@ -3,6 +3,7 @@ from __future__ import annotations
|
||||
import logging
|
||||
import re
|
||||
from fnmatch import fnmatch
|
||||
from fnmatch import translate as fnmatch_translate
|
||||
from typing import TYPE_CHECKING
|
||||
|
||||
from rest_framework import serializers
|
||||
@@ -547,48 +548,6 @@ def existing_document_matches_workflow(
|
||||
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(
|
||||
documents: QuerySet[Document],
|
||||
trigger: WorkflowTrigger,
|
||||
@@ -669,9 +628,8 @@ def prefilter_documents_by_workflowtrigger(
|
||||
documents = documents.annotate(**annotations).filter(custom_field_q)
|
||||
|
||||
if trigger.filter_filename:
|
||||
documents = documents.filter(
|
||||
original_filename__iregex=glob_to_database_regex(trigger.filter_filename),
|
||||
)
|
||||
regex = fnmatch_translate(trigger.filter_filename).lstrip("^").rstrip("$")
|
||||
documents = documents.filter(original_filename__iregex=regex)
|
||||
|
||||
return documents
|
||||
|
||||
|
||||
@@ -6,7 +6,6 @@ from unittest.mock import patch
|
||||
from urllib.parse import quote
|
||||
|
||||
import time_machine
|
||||
from django.db import connection
|
||||
from django.test import override_settings
|
||||
from django.utils import timezone
|
||||
from rest_framework.test import APITestCase
|
||||
@@ -21,13 +20,6 @@ from paperless_testing.dirs import DirectoriesMixin
|
||||
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:
|
||||
"""
|
||||
Allows Pythonic access to the custom fields associated with the wrapped document.
|
||||
@@ -295,8 +287,7 @@ class TestCustomFieldsSearch(DirectoriesMixin, APITestCase):
|
||||
self._assert_query_match_predicate(
|
||||
["string_field", "exact", "paperless"],
|
||||
lambda document: (
|
||||
"string_field" in document
|
||||
and _exact_equal(document["string_field"], "paperless")
|
||||
"string_field" in document and document["string_field"] == "paperless"
|
||||
),
|
||||
)
|
||||
|
||||
@@ -305,10 +296,7 @@ class TestCustomFieldsSearch(DirectoriesMixin, APITestCase):
|
||||
["string_field", "in", ["paperless", "Paperless"]],
|
||||
lambda document: (
|
||||
"string_field" in document
|
||||
and any(
|
||||
_exact_equal(document["string_field"], expected)
|
||||
for expected in ("paperless", "Paperless")
|
||||
)
|
||||
and document["string_field"] in ("paperless", "Paperless")
|
||||
),
|
||||
)
|
||||
|
||||
@@ -773,7 +761,7 @@ class TestCustomFieldsSearch(DirectoriesMixin, APITestCase):
|
||||
lambda document: (
|
||||
not (
|
||||
"string_field" in document
|
||||
and _exact_equal(document["string_field"], "paperless")
|
||||
and document["string_field"] == "paperless"
|
||||
)
|
||||
),
|
||||
)
|
||||
|
||||
@@ -242,7 +242,9 @@ class TestDocumentListEffectiveContentAnnotation:
|
||||
# Only the list query itself reads a content column: no extra query
|
||||
# for the skipped prefetch, none for a per-instance fallback
|
||||
content_queries = [
|
||||
query for query in ctx.captured_queries if _content_column() in query["sql"]
|
||||
query
|
||||
for query in ctx.captured_queries
|
||||
if '"documents_document"."content"' in query["sql"]
|
||||
]
|
||||
assert len(content_queries) == 1
|
||||
|
||||
@@ -363,11 +365,6 @@ def _get_document_queries(
|
||||
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(
|
||||
ctx: CaptureQueriesContext,
|
||||
) -> list[dict[str, str]]:
|
||||
@@ -382,7 +379,7 @@ def _get_effective_content_fallback_queries(
|
||||
return [
|
||||
q
|
||||
for q in ctx.captured_queries
|
||||
if q["sql"].startswith(f"SELECT {_content_column()} FROM")
|
||||
if q["sql"].startswith('SELECT "documents_document"."content" FROM')
|
||||
]
|
||||
|
||||
|
||||
|
||||
@@ -1,68 +0,0 @@
|
||||
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