mirror of
https://github.com/paperless-ngx/paperless-ngx.git
synced 2026-10-31 12:05:45 +00:00
Fix: scheduled workflow filename filter fails on PostgreSQL
The prefilter passed fnmatch.translate output to an iregex lookup. That output uses (?s:...), atomic groups and named group backreferences, which PostgreSQL rejects with "invalid embedded option". Translate the glob with a helper that only emits syntax shared by Python, PostgreSQL and MariaDB. Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>
This commit is contained in:
1 parent
2cb04b0b8b
commit
f4358cef53
2 files changed
+113
-3
No files matched your search
@@ -3,7 +3,6 @@ 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
|
||||
@@ -548,6 +547,48 @@ 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,
|
||||
@@ -628,8 +669,9 @@ def prefilter_documents_by_workflowtrigger(
|
||||
documents = documents.annotate(**annotations).filter(custom_field_q)
|
||||
|
||||
if trigger.filter_filename:
|
||||
regex = fnmatch_translate(trigger.filter_filename).lstrip("^").rstrip("$")
|
||||
documents = documents.filter(original_filename__iregex=regex)
|
||||
documents = documents.filter(
|
||||
original_filename__iregex=glob_to_database_regex(trigger.filter_filename),
|
||||
)
|
||||
|
||||
return documents
|
||||
|
||||
|
||||
@@ -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