We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
[BUG] OS reports file as busy – wrong error message?!
### Description

Hello everyone,

I have setup paperless-ngx in Docker in my Synology NAS. I have created a bind mount for the `data/consume` folder and threw in some pdf files. However I always got an error `OS reports file as busy still`. Now, I've made sure that no other client is accessing the file. And the file can be opened from any other network device and the NAS web interface. Turns out, this was actually a permissions issue with wrong User IDs. So, the error message is factually not correct.

So I am under the impression that the error message is not very precise. From my little Python knowledge, I've looked[ at the code](https://github.com/paperless-ngx/paperless-ngx/blob/4df065d8d524870ec18e8fbf2fc488449939a044/src/documents/management/commands/document_consumer.py#L74) and it seems to be a very [generic Exception](https://docs.python.org/3/library/exceptions.html#OSError). It could also be thrown, when the disk is full, etc. There is a specific error number in the Exception. Would it make sense, to add that to the error output in the log and/or to change the message?

### Steps to reproduce

There's already a test for it, if you want to trigger the error message on purpose.

### Webserver logs

```bash
[2022-06-20 10:22:37,016] [WARNING] [paperless.management.consumer] Not consuming file /data/consume/E79221B2X115146_06202022_122344_000002.pdf: OS reports file as busy still

[2022-06-20 10:26:20,714] [WARNING] [paperless.management.consumer] Not consuming file /data/consume/E79221B2X115146_06202022_122344_000003.pdf: OS reports file as busy still

[2022-06-20 10:26:47,094] [WARNING] [paperless.management.consumer] Not consuming file /data/consume/.DS_Store: Unknown file extension.

[2022-06-20 10:26:48,738] [WARNING] [paperless.management.consumer] Not consuming file /data/consume/Fragebogen-Mehrfachbeschäftigung-Minijob.pdf: OS reports file as busy still

[2022-06-20 10:26:53,120] [WARNING] [paperless.management.consumer] Not consuming file /data/consume/.DS_Store: Unknown file extension.
```


### Paperless-ngx version

1.7.1

### Host OS

Synology DSM 7

### Installation method

Docker

### Browser

_No response_

### Configuration changes

_No response_

### Other

_No response_

--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `src/documents/management/commands/document_consumer.py`
Content:
```
1 import logging
2 import os
3 from pathlib import Path
4 from pathlib import PurePath
5 from threading import Thread
6 from time import monotonic
7 from time import sleep
8 from typing import Final
9 
10 from django.conf import settings
11 from django.core.management.base import BaseCommand
12 from django.core.management.base import CommandError
13 from django_q.tasks import async_task
14 from documents.models import Tag
15 from documents.parsers import is_file_ext_supported
16 from watchdog.events import FileSystemEventHandler
17 from watchdog.observers.polling import PollingObserver
18 
19 try:
20     from inotifyrecursive import INotify, flags
21 except ImportError:
22     INotify = flags = None
23 
24 logger = logging.getLogger("paperless.management.consumer")
25 
26 
27 def _tags_from_path(filepath):
28     """Walk up the directory tree from filepath to CONSUMPTION_DIR
29     and get or create Tag IDs for every directory.
30     """
31     normalized_consumption_dir = os.path.abspath(
32         os.path.normpath(settings.CONSUMPTION_DIR),
33     )
34     tag_ids = set()
35     path_parts = Path(filepath).relative_to(normalized_consumption_dir).parent.parts
36     for part in path_parts:
37         tag_ids.add(
38             Tag.objects.get_or_create(name__iexact=part, defaults={"name": part})[0].pk,
39         )
40 
41     return tag_ids
42 
43 
44 def _is_ignored(filepath: str) -> bool:
45     normalized_consumption_dir = os.path.abspath(
46         os.path.normpath(settings.CONSUMPTION_DIR),
47     )
48     filepath_relative = PurePath(filepath).relative_to(normalized_consumption_dir)
49     return any(filepath_relative.match(p) for p in settings.CONSUMER_IGNORE_PATTERNS)
50 
51 
52 def _consume(filepath):
53     if os.path.isdir(filepath) or _is_ignored(filepath):
54         return
55 
56     if not os.path.isfile(filepath):
57         logger.debug(f"Not consuming file {filepath}: File has moved.")
58         return
59 
60     if not is_file_ext_supported(os.path.splitext(filepath)[1]):
61         logger.warning(f"Not consuming file {filepath}: Unknown file extension.")
62         return
63 
64     # Total wait time: up to 500ms
65     os_error_retry_count: Final[int] = 50
66     os_error_retry_wait: Final[float] = 0.01
67 
68     read_try_count = 0
69     file_open_ok = False
70 
71     while (read_try_count < os_error_retry_count) and not file_open_ok:
72         try:
73             with open(filepath, "rb"):
74                 file_open_ok = True
75         except OSError:
76             read_try_count += 1
77             sleep(os_error_retry_wait)
78 
79     if read_try_count >= os_error_retry_count:
80         logger.warning(f"Not consuming file {filepath}: OS reports file as busy still")
81         return
82 
83     tag_ids = None
84     try:
85         if settings.CONSUMER_SUBDIRS_AS_TAGS:
86             tag_ids = _tags_from_path(filepath)
87     except Exception:
88         logger.exception("Error creating tags from path")
89 
90     try:
91         logger.info(f"Adding {filepath} to the task queue.")
92         async_task(
93             "documents.tasks.consume_file",
94             filepath,
95             override_tag_ids=tag_ids if tag_ids else None,
96             task_name=os.path.basename(filepath)[:100],
97         )
98     except Exception:
99         # Catch all so that the consumer won't crash.
100         # This is also what the test case is listening for to check for
101         # errors.
102         logger.exception("Error while consuming document")
103 
104 
105 def _consume_wait_unmodified(file):
106     if _is_ignored(file):
107         return
108 
109     logger.debug(f"Waiting for file {file} to remain unmodified")
110     mtime = -1
111     size = -1
112     current_try = 0
113     while current_try < settings.CONSUMER_POLLING_RETRY_COUNT:
114         try:
115             stat_data = os.stat(file)
116             new_mtime = stat_data.st_mtime
117             new_size = stat_data.st_size
118         except FileNotFoundError:
119             logger.debug(
120                 f"File {file} moved while waiting for it to remain " f"unmodified.",
121             )
122             return
123         if new_mtime == mtime and new_size == size:
124             _consume(file)
125             return
126         mtime = new_mtime
127         size = new_size
128         sleep(settings.CONSUMER_POLLING_DELAY)
129         current_try += 1
130 
131     logger.error(f"Timeout while waiting on file {file} to remain unmodified.")
132 
133 
134 class Handler(FileSystemEventHandler):
135     def on_created(self, event):
136         Thread(target=_consume_wait_unmodified, args=(event.src_path,)).start()
137 
138     def on_moved(self, event):
139         Thread(target=_consume_wait_unmodified, args=(event.dest_path,)).start()
140 
141 
142 class Command(BaseCommand):
143     """
144     On every iteration of an infinite loop, consume what we can from the
145     consumption directory.
146     """
147 
148     # This is here primarily for the tests and is irrelevant in production.
149     stop_flag = False
150 
151     observer = None
152 
153     def add_arguments(self, parser):
154         parser.add_argument(
155             "directory",
156             default=settings.CONSUMPTION_DIR,
157             nargs="?",
158             help="The consumption directory.",
159         )
160         parser.add_argument("--oneshot", action="store_true", help="Run only once.")
161 
162     def handle(self, *args, **options):
163         directory = options["directory"]
164         recursive = settings.CONSUMER_RECURSIVE
165 
166         if not directory:
167             raise CommandError("CONSUMPTION_DIR does not appear to be set.")
168 
169         directory = os.path.abspath(directory)
170 
171         if not os.path.isdir(directory):
172             raise CommandError(f"Consumption directory {directory} does not exist")
173 
174         if recursive:
175             for dirpath, _, filenames in os.walk(directory):
176                 for filename in filenames:
177                     filepath = os.path.join(dirpath, filename)
178                     _consume(filepath)
179         else:
180             for entry in os.scandir(directory):
181                 _consume(entry.path)
182 
183         if options["oneshot"]:
184             return
185 
186         if settings.CONSUMER_POLLING == 0 and INotify:
187             self.handle_inotify(directory, recursive)
188         else:
189             self.handle_polling(directory, recursive)
190 
191         logger.debug("Consumer exiting.")
192 
193     def handle_polling(self, directory, recursive):
194         logger.info(f"Polling directory for changes: {directory}")
195         self.observer = PollingObserver(timeout=settings.CONSUMER_POLLING)
196         self.observer.schedule(Handler(), directory, recursive=recursive)
197         self.observer.start()
198         try:
199             while self.observer.is_alive():
200                 self.observer.join(1)
201                 if self.stop_flag:
202                     self.observer.stop()
203         except KeyboardInterrupt:
204             self.observer.stop()
205         self.observer.join()
206 
207     def handle_inotify(self, directory, recursive):
208         logger.info(f"Using inotify to watch directory for changes: {directory}")
209 
210         inotify = INotify()
211         inotify_flags = flags.CLOSE_WRITE | flags.MOVED_TO
212         if recursive:
213             descriptor = inotify.add_watch_recursive(directory, inotify_flags)
214         else:
215             descriptor = inotify.add_watch(directory, inotify_flags)
216 
217         try:
218 
219             inotify_debounce: Final[float] = settings.CONSUMER_INOTIFY_DELAY
220             notified_files = {}
221 
222             while not self.stop_flag:
223 
224                 for event in inotify.read(timeout=1000):
225                     if recursive:
226                         path = inotify.get_path(event.wd)
227                     else:
228                         path = directory
229                     filepath = os.path.join(path, event.name)
230                     notified_files[filepath] = monotonic()
231 
232                 # Check the files against the timeout
233                 still_waiting = {}
234                 for filepath in notified_files:
235                     # Time of the last inotify event for this file
236                     last_event_time = notified_files[filepath]
237 
238                     # Current time - last time over the configured timeout
239                     waited_long_enough = (
240                         monotonic() - last_event_time
241                     ) > inotify_debounce
242 
243                     # Also make sure the file exists still, some scanners might write a
244                     # temporary file first
245                     file_still_exists = os.path.exists(filepath) and os.path.isfile(
246                         filepath,
247                     )
248 
249                     if waited_long_enough and file_still_exists:
250                         _consume(filepath)
251                     elif file_still_exists:
252                         still_waiting[filepath] = last_event_time
253 
254                 # These files are still waiting to hit the timeout
255                 notified_files = still_waiting
256 
257         except KeyboardInterrupt:
258             pass
259 
260         inotify.rm_watch(descriptor)
261         inotify.close()
262
```
--- END FILES ---

Please first localize the bug based on the issue statement, and then generate a patch according to the `git diff` format fenced by three backticks.

Here is an example:

```diff
diff --git a/examples/server_async.py b/examples/server_async.py
--- a/examples/server_async.py
+++ b/examples/server_async.py
@@ -313,4 +313,4 @@
 
 
 if __name__ == "__main__":
-    asyncio.run(run_async_server("."), debug=True)
+    asyncio.run(run_async_server(), debug=True)
diff --git a/examples/server_sync.py b/examples/server_sync.py
--- a/examples/server_sync.py
+++ b/examples/server_sync.py
@@ -313,5 +313,5 @@
 
 
 if __name__ == "__main__":
-    server = run_sync_server(".")
+    server = run_sync_server()
     server.shutdown()

```

diff --git a/src/documents/management/commands/document_consumer.py b/src/documents/management/commands/document_consumer.py
--- a/src/documents/management/commands/document_consumer.py
+++ b/src/documents/management/commands/document_consumer.py
@@ -67,17 +67,19 @@
 
     read_try_count = 0
     file_open_ok = False
+    os_error_str = None
 
     while (read_try_count < os_error_retry_count) and not file_open_ok:
         try:
             with open(filepath, "rb"):
                 file_open_ok = True
-        except OSError:
+        except OSError as e:
             read_try_count += 1
+            os_error_str = str(e)
             sleep(os_error_retry_wait)
 
     if read_try_count >= os_error_retry_count:
-        logger.warning(f"Not consuming file {filepath}: OS reports file as busy still")
+        logger.warning(f"Not consuming file {filepath}: OS reports {os_error_str}")
         return
 
     tag_ids = None

{"golden_diff": "diff --git a/src/documents/management/commands/document_consumer.py b/src/documents/management/commands/document_consumer.py\n--- a/src/documents/management/commands/document_consumer.py\n+++ b/src/documents/management/commands/document_consumer.py\n@@ -67,17 +67,19 @@\n \n     read_try_count = 0\n     file_open_ok = False\n+    os_error_str = None\n \n     while (read_try_count < os_error_retry_count) and not file_open_ok:\n         try:\n             with open(filepath, \"rb\"):\n                 file_open_ok = True\n-        except OSError:\n+        except OSError as e:\n             read_try_count += 1\n+            os_error_str = str(e)\n             sleep(os_error_retry_wait)\n \n     if read_try_count >= os_error_retry_count:\n-        logger.warning(f\"Not consuming file {filepath}: OS reports file as busy still\")\n+        logger.warning(f\"Not consuming file {filepath}: OS reports {os_error_str}\")\n         return\n \n     tag_ids = None\n", "issue": "[BUG] OS reports file as busy \u2013 wrong error message?!\n### Description\n\nHello everyone,\r\n\r\nI have setup paperless-ngx in Docker in my Synology NAS. I have created a bind mount for the `data/consume` folder and threw in some pdf files. However I always got an error `OS reports file as busy still`. Now, I've made sure that no other client is accessing the file. And the file can be opened from any other network device and the NAS web interface. Turns out, this was actually a permissions issue with wrong User IDs. So, the error message is factually not correct.\r\n\r\nSo I am under the impression that the error message is not very precise. From my little Python knowledge, I've looked[ at the code](https://github.com/paperless-ngx/paperless-ngx/blob/4df065d8d524870ec18e8fbf2fc488449939a044/src/documents/management/commands/document_consumer.py#L74) and it seems to be a very [generic Exception](https://docs.python.org/3/library/exceptions.html#OSError). It could also be thrown, when the disk is full, etc. There is a specific error number in the Exception. Would it make sense, to add that to the error output in the log and/or to change the message?\n\n### Steps to reproduce\n\nThere's already a test for it, if you want to trigger the error message on purpose.\n\n### Webserver logs\n\n```bash\n[2022-06-20 10:22:37,016] [WARNING] [paperless.management.consumer] Not consuming file /data/consume/E79221B2X115146_06202022_122344_000002.pdf: OS reports file as busy still\r\n\r\n[2022-06-20 10:26:20,714] [WARNING] [paperless.management.consumer] Not consuming file /data/consume/E79221B2X115146_06202022_122344_000003.pdf: OS reports file as busy still\r\n\r\n[2022-06-20 10:26:47,094] [WARNING] [paperless.management.consumer] Not consuming file /data/consume/.DS_Store: Unknown file extension.\r\n\r\n[2022-06-20 10:26:48,738] [WARNING] [paperless.management.consumer] Not consuming file /data/consume/Fragebogen-Mehrfachbesch\u00e4ftigung-Minijob.pdf: OS reports file as busy still\r\n\r\n[2022-06-20 10:26:53,120] [WARNING] [paperless.management.consumer] Not consuming file /data/consume/.DS_Store: Unknown file extension.\n```\n\n\n### Paperless-ngx version\n\n1.7.1\n\n### Host OS\n\nSynology DSM 7\n\n### Installation method\n\nDocker\n\n### Browser\n\n_No response_\n\n### Configuration changes\n\n_No response_\n\n### Other\n\n_No response_\n", "before_files": [{"content": "import logging\nimport os\nfrom pathlib import Path\nfrom pathlib import PurePath\nfrom threading import Thread\nfrom time import monotonic\nfrom time import sleep\nfrom typing import Final\n\nfrom django.conf import settings\nfrom django.core.management.base import BaseCommand\nfrom django.core.management.base import CommandError\nfrom django_q.tasks import async_task\nfrom documents.models import Tag\nfrom documents.parsers import is_file_ext_supported\nfrom watchdog.events import FileSystemEventHandler\nfrom watchdog.observers.polling import PollingObserver\n\ntry:\n    from inotifyrecursive import INotify, flags\nexcept ImportError:\n    INotify = flags = None\n\nlogger = logging.getLogger(\"paperless.management.consumer\")\n\n\ndef _tags_from_path(filepath):\n    \"\"\"Walk up the directory tree from filepath to CONSUMPTION_DIR\n    and get or create Tag IDs for every directory.\n    \"\"\"\n    normalized_consumption_dir = os.path.abspath(\n        os.path.normpath(settings.CONSUMPTION_DIR),\n    )\n    tag_ids = set()\n    path_parts = Path(filepath).relative_to(normalized_consumption_dir).parent.parts\n    for part in path_parts:\n        tag_ids.add(\n            Tag.objects.get_or_create(name__iexact=part, defaults={\"name\": part})[0].pk,\n        )\n\n    return tag_ids\n\n\ndef _is_ignored(filepath: str) -> bool:\n    normalized_consumption_dir = os.path.abspath(\n        os.path.normpath(settings.CONSUMPTION_DIR),\n    )\n    filepath_relative = PurePath(filepath).relative_to(normalized_consumption_dir)\n    return any(filepath_relative.match(p) for p in settings.CONSUMER_IGNORE_PATTERNS)\n\n\ndef _consume(filepath):\n    if os.path.isdir(filepath) or _is_ignored(filepath):\n        return\n\n    if not os.path.isfile(filepath):\n        logger.debug(f\"Not consuming file {filepath}: File has moved.\")\n        return\n\n    if not is_file_ext_supported(os.path.splitext(filepath)[1]):\n        logger.warning(f\"Not consuming file {filepath}: Unknown file extension.\")\n        return\n\n    # Total wait time: up to 500ms\n    os_error_retry_count: Final[int] = 50\n    os_error_retry_wait: Final[float] = 0.01\n\n    read_try_count = 0\n    file_open_ok = False\n\n    while (read_try_count < os_error_retry_count) and not file_open_ok:\n        try:\n            with open(filepath, \"rb\"):\n                file_open_ok = True\n        except OSError:\n            read_try_count += 1\n            sleep(os_error_retry_wait)\n\n    if read_try_count >= os_error_retry_count:\n        logger.warning(f\"Not consuming file {filepath}: OS reports file as busy still\")\n        return\n\n    tag_ids = None\n    try:\n        if settings.CONSUMER_SUBDIRS_AS_TAGS:\n            tag_ids = _tags_from_path(filepath)\n    except Exception:\n        logger.exception(\"Error creating tags from path\")\n\n    try:\n        logger.info(f\"Adding {filepath} to the task queue.\")\n        async_task(\n            \"documents.tasks.consume_file\",\n            filepath,\n            override_tag_ids=tag_ids if tag_ids else None,\n            task_name=os.path.basename(filepath)[:100],\n        )\n    except Exception:\n        # Catch all so that the consumer won't crash.\n        # This is also what the test case is listening for to check for\n        # errors.\n        logger.exception(\"Error while consuming document\")\n\n\ndef _consume_wait_unmodified(file):\n    if _is_ignored(file):\n        return\n\n    logger.debug(f\"Waiting for file {file} to remain unmodified\")\n    mtime = -1\n    size = -1\n    current_try = 0\n    while current_try < settings.CONSUMER_POLLING_RETRY_COUNT:\n        try:\n            stat_data = os.stat(file)\n            new_mtime = stat_data.st_mtime\n            new_size = stat_data.st_size\n        except FileNotFoundError:\n            logger.debug(\n                f\"File {file} moved while waiting for it to remain \" f\"unmodified.\",\n            )\n            return\n        if new_mtime == mtime and new_size == size:\n            _consume(file)\n            return\n        mtime = new_mtime\n        size = new_size\n        sleep(settings.CONSUMER_POLLING_DELAY)\n        current_try += 1\n\n    logger.error(f\"Timeout while waiting on file {file} to remain unmodified.\")\n\n\nclass Handler(FileSystemEventHandler):\n    def on_created(self, event):\n        Thread(target=_consume_wait_unmodified, args=(event.src_path,)).start()\n\n    def on_moved(self, event):\n        Thread(target=_consume_wait_unmodified, args=(event.dest_path,)).start()\n\n\nclass Command(BaseCommand):\n    \"\"\"\n    On every iteration of an infinite loop, consume what we can from the\n    consumption directory.\n    \"\"\"\n\n    # This is here primarily for the tests and is irrelevant in production.\n    stop_flag = False\n\n    observer = None\n\n    def add_arguments(self, parser):\n        parser.add_argument(\n            \"directory\",\n            default=settings.CONSUMPTION_DIR,\n            nargs=\"?\",\n            help=\"The consumption directory.\",\n        )\n        parser.add_argument(\"--oneshot\", action=\"store_true\", help=\"Run only once.\")\n\n    def handle(self, *args, **options):\n        directory = options[\"directory\"]\n        recursive = settings.CONSUMER_RECURSIVE\n\n        if not directory:\n            raise CommandError(\"CONSUMPTION_DIR does not appear to be set.\")\n\n        directory = os.path.abspath(directory)\n\n        if not os.path.isdir(directory):\n            raise CommandError(f\"Consumption directory {directory} does not exist\")\n\n        if recursive:\n            for dirpath, _, filenames in os.walk(directory):\n                for filename in filenames:\n                    filepath = os.path.join(dirpath, filename)\n                    _consume(filepath)\n        else:\n            for entry in os.scandir(directory):\n                _consume(entry.path)\n\n        if options[\"oneshot\"]:\n            return\n\n        if settings.CONSUMER_POLLING == 0 and INotify:\n            self.handle_inotify(directory, recursive)\n        else:\n            self.handle_polling(directory, recursive)\n\n        logger.debug(\"Consumer exiting.\")\n\n    def handle_polling(self, directory, recursive):\n        logger.info(f\"Polling directory for changes: {directory}\")\n        self.observer = PollingObserver(timeout=settings.CONSUMER_POLLING)\n        self.observer.schedule(Handler(), directory, recursive=recursive)\n        self.observer.start()\n        try:\n            while self.observer.is_alive():\n                self.observer.join(1)\n                if self.stop_flag:\n                    self.observer.stop()\n        except KeyboardInterrupt:\n            self.observer.stop()\n        self.observer.join()\n\n    def handle_inotify(self, directory, recursive):\n        logger.info(f\"Using inotify to watch directory for changes: {directory}\")\n\n        inotify = INotify()\n        inotify_flags = flags.CLOSE_WRITE | flags.MOVED_TO\n        if recursive:\n            descriptor = inotify.add_watch_recursive(directory, inotify_flags)\n        else:\n            descriptor = inotify.add_watch(directory, inotify_flags)\n\n        try:\n\n            inotify_debounce: Final[float] = settings.CONSUMER_INOTIFY_DELAY\n            notified_files = {}\n\n            while not self.stop_flag:\n\n                for event in inotify.read(timeout=1000):\n                    if recursive:\n                        path = inotify.get_path(event.wd)\n                    else:\n                        path = directory\n                    filepath = os.path.join(path, event.name)\n                    notified_files[filepath] = monotonic()\n\n                # Check the files against the timeout\n                still_waiting = {}\n                for filepath in notified_files:\n                    # Time of the last inotify event for this file\n                    last_event_time = notified_files[filepath]\n\n                    # Current time - last time over the configured timeout\n                    waited_long_enough = (\n                        monotonic() - last_event_time\n                    ) > inotify_debounce\n\n                    # Also make sure the file exists still, some scanners might write a\n                    # temporary file first\n                    file_still_exists = os.path.exists(filepath) and os.path.isfile(\n                        filepath,\n                    )\n\n                    if waited_long_enough and file_still_exists:\n                        _consume(filepath)\n                    elif file_still_exists:\n                        still_waiting[filepath] = last_event_time\n\n                # These files are still waiting to hit the timeout\n                notified_files = still_waiting\n\n        except KeyboardInterrupt:\n            pass\n\n        inotify.rm_watch(descriptor)\n        inotify.close()\n", "path": "src/documents/management/commands/document_consumer.py"}], "after_files": [{"content": "import logging\nimport os\nfrom pathlib import Path\nfrom pathlib import PurePath\nfrom threading import Thread\nfrom time import monotonic\nfrom time import sleep\nfrom typing import Final\n\nfrom django.conf import settings\nfrom django.core.management.base import BaseCommand\nfrom django.core.management.base import CommandError\nfrom django_q.tasks import async_task\nfrom documents.models import Tag\nfrom documents.parsers import is_file_ext_supported\nfrom watchdog.events import FileSystemEventHandler\nfrom watchdog.observers.polling import PollingObserver\n\ntry:\n    from inotifyrecursive import INotify, flags\nexcept ImportError:\n    INotify = flags = None\n\nlogger = logging.getLogger(\"paperless.management.consumer\")\n\n\ndef _tags_from_path(filepath):\n    \"\"\"Walk up the directory tree from filepath to CONSUMPTION_DIR\n    and get or create Tag IDs for every directory.\n    \"\"\"\n    normalized_consumption_dir = os.path.abspath(\n        os.path.normpath(settings.CONSUMPTION_DIR),\n    )\n    tag_ids = set()\n    path_parts = Path(filepath).relative_to(normalized_consumption_dir).parent.parts\n    for part in path_parts:\n        tag_ids.add(\n            Tag.objects.get_or_create(name__iexact=part, defaults={\"name\": part})[0].pk,\n        )\n\n    return tag_ids\n\n\ndef _is_ignored(filepath: str) -> bool:\n    normalized_consumption_dir = os.path.abspath(\n        os.path.normpath(settings.CONSUMPTION_DIR),\n    )\n    filepath_relative = PurePath(filepath).relative_to(normalized_consumption_dir)\n    return any(filepath_relative.match(p) for p in settings.CONSUMER_IGNORE_PATTERNS)\n\n\ndef _consume(filepath):\n    if os.path.isdir(filepath) or _is_ignored(filepath):\n        return\n\n    if not os.path.isfile(filepath):\n        logger.debug(f\"Not consuming file {filepath}: File has moved.\")\n        return\n\n    if not is_file_ext_supported(os.path.splitext(filepath)[1]):\n        logger.warning(f\"Not consuming file {filepath}: Unknown file extension.\")\n        return\n\n    # Total wait time: up to 500ms\n    os_error_retry_count: Final[int] = 50\n    os_error_retry_wait: Final[float] = 0.01\n\n    read_try_count = 0\n    file_open_ok = False\n    os_error_str = None\n\n    while (read_try_count < os_error_retry_count) and not file_open_ok:\n        try:\n            with open(filepath, \"rb\"):\n                file_open_ok = True\n        except OSError as e:\n            read_try_count += 1\n            os_error_str = str(e)\n            sleep(os_error_retry_wait)\n\n    if read_try_count >= os_error_retry_count:\n        logger.warning(f\"Not consuming file {filepath}: OS reports {os_error_str}\")\n        return\n\n    tag_ids = None\n    try:\n        if settings.CONSUMER_SUBDIRS_AS_TAGS:\n            tag_ids = _tags_from_path(filepath)\n    except Exception:\n        logger.exception(\"Error creating tags from path\")\n\n    try:\n        logger.info(f\"Adding {filepath} to the task queue.\")\n        async_task(\n            \"documents.tasks.consume_file\",\n            filepath,\n            override_tag_ids=tag_ids if tag_ids else None,\n            task_name=os.path.basename(filepath)[:100],\n        )\n    except Exception:\n        # Catch all so that the consumer won't crash.\n        # This is also what the test case is listening for to check for\n        # errors.\n        logger.exception(\"Error while consuming document\")\n\n\ndef _consume_wait_unmodified(file):\n    if _is_ignored(file):\n        return\n\n    logger.debug(f\"Waiting for file {file} to remain unmodified\")\n    mtime = -1\n    size = -1\n    current_try = 0\n    while current_try < settings.CONSUMER_POLLING_RETRY_COUNT:\n        try:\n            stat_data = os.stat(file)\n            new_mtime = stat_data.st_mtime\n            new_size = stat_data.st_size\n        except FileNotFoundError:\n            logger.debug(\n                f\"File {file} moved while waiting for it to remain \" f\"unmodified.\",\n            )\n            return\n        if new_mtime == mtime and new_size == size:\n            _consume(file)\n            return\n        mtime = new_mtime\n        size = new_size\n        sleep(settings.CONSUMER_POLLING_DELAY)\n        current_try += 1\n\n    logger.error(f\"Timeout while waiting on file {file} to remain unmodified.\")\n\n\nclass Handler(FileSystemEventHandler):\n    def on_created(self, event):\n        Thread(target=_consume_wait_unmodified, args=(event.src_path,)).start()\n\n    def on_moved(self, event):\n        Thread(target=_consume_wait_unmodified, args=(event.dest_path,)).start()\n\n\nclass Command(BaseCommand):\n    \"\"\"\n    On every iteration of an infinite loop, consume what we can from the\n    consumption directory.\n    \"\"\"\n\n    # This is here primarily for the tests and is irrelevant in production.\n    stop_flag = False\n\n    observer = None\n\n    def add_arguments(self, parser):\n        parser.add_argument(\n            \"directory\",\n            default=settings.CONSUMPTION_DIR,\n            nargs=\"?\",\n            help=\"The consumption directory.\",\n        )\n        parser.add_argument(\"--oneshot\", action=\"store_true\", help=\"Run only once.\")\n\n    def handle(self, *args, **options):\n        directory = options[\"directory\"]\n        recursive = settings.CONSUMER_RECURSIVE\n\n        if not directory:\n            raise CommandError(\"CONSUMPTION_DIR does not appear to be set.\")\n\n        directory = os.path.abspath(directory)\n\n        if not os.path.isdir(directory):\n            raise CommandError(f\"Consumption directory {directory} does not exist\")\n\n        if recursive:\n            for dirpath, _, filenames in os.walk(directory):\n                for filename in filenames:\n                    filepath = os.path.join(dirpath, filename)\n                    _consume(filepath)\n        else:\n            for entry in os.scandir(directory):\n                _consume(entry.path)\n\n        if options[\"oneshot\"]:\n            return\n\n        if settings.CONSUMER_POLLING == 0 and INotify:\n            self.handle_inotify(directory, recursive)\n        else:\n            self.handle_polling(directory, recursive)\n\n        logger.debug(\"Consumer exiting.\")\n\n    def handle_polling(self, directory, recursive):\n        logger.info(f\"Polling directory for changes: {directory}\")\n        self.observer = PollingObserver(timeout=settings.CONSUMER_POLLING)\n        self.observer.schedule(Handler(), directory, recursive=recursive)\n        self.observer.start()\n        try:\n            while self.observer.is_alive():\n                self.observer.join(1)\n                if self.stop_flag:\n                    self.observer.stop()\n        except KeyboardInterrupt:\n            self.observer.stop()\n        self.observer.join()\n\n    def handle_inotify(self, directory, recursive):\n        logger.info(f\"Using inotify to watch directory for changes: {directory}\")\n\n        inotify = INotify()\n        inotify_flags = flags.CLOSE_WRITE | flags.MOVED_TO\n        if recursive:\n            descriptor = inotify.add_watch_recursive(directory, inotify_flags)\n        else:\n            descriptor = inotify.add_watch(directory, inotify_flags)\n\n        try:\n\n            inotify_debounce: Final[float] = settings.CONSUMER_INOTIFY_DELAY\n            notified_files = {}\n\n            while not self.stop_flag:\n\n                for event in inotify.read(timeout=1000):\n                    if recursive:\n                        path = inotify.get_path(event.wd)\n                    else:\n                        path = directory\n                    filepath = os.path.join(path, event.name)\n                    notified_files[filepath] = monotonic()\n\n                # Check the files against the timeout\n                still_waiting = {}\n                for filepath in notified_files:\n                    # Time of the last inotify event for this file\n                    last_event_time = notified_files[filepath]\n\n                    # Current time - last time over the configured timeout\n                    waited_long_enough = (\n                        monotonic() - last_event_time\n                    ) > inotify_debounce\n\n                    # Also make sure the file exists still, some scanners might write a\n                    # temporary file first\n                    file_still_exists = os.path.exists(filepath) and os.path.isfile(\n                        filepath,\n                    )\n\n                    if waited_long_enough and file_still_exists:\n                        _consume(filepath)\n                    elif file_still_exists:\n                        still_waiting[filepath] = last_event_time\n\n                # These files are still waiting to hit the timeout\n                notified_files = still_waiting\n\n        except KeyboardInterrupt:\n            pass\n\n        inotify.rm_watch(descriptor)\n        inotify.close()\n", "path": "src/documents/management/commands/document_consumer.py"}]}