gitea-admin/gh-christianlouis-docuelevate
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge pull request #504 from christianlouis/copilot/add-watch-folder-support

Browse Source
This commit is contained in:
Christian Krakau-Louis
2026-03-08 10:05:10 +01:00
committed by GitHub
parent c69dc1b8c1 9a9efae19e
commit bd9da65511
11 changed files with 2814 additions and 8 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.env.demo
+53
View File
@@ -206,7 +206,60 @@ EMAIL_USE_TLS=True
EMAIL_SENDER=DocuElevate System <docuelevate@example.com>
EMAIL_DEFAULT_RECIPIENT=recipient@example.com
# **Watch Folder Ingestion**
# DocuElevate can automatically monitor directories (local, FTP, SFTP, and cloud providers) for new files.
#
# Local watch folders — works with any mounted path (SMB/CIFS, NFS, local disk, etc.)
# Set WATCH_FOLDERS to a comma-separated list of absolute paths inside the container.
WATCH_FOLDERS=
WATCH_FOLDER_POLL_INTERVAL=1
WATCH_FOLDER_DELETE_AFTER_PROCESS=false
# FTP ingest — poll an FTP directory for new files (uses FTP connection settings above)
FTP_INGEST_ENABLED=false
FTP_INGEST_FOLDER=
FTP_INGEST_DELETE_AFTER_PROCESS=false
# SFTP ingest — poll an SFTP directory for new files (uses SFTP connection settings above)
SFTP_INGEST_ENABLED=false
SFTP_INGEST_FOLDER=
SFTP_INGEST_DELETE_AFTER_PROCESS=false
# Dropbox ingest — poll a Dropbox folder (uses Dropbox OAuth credentials above)
DROPBOX_INGEST_ENABLED=false
DROPBOX_INGEST_FOLDER=
DROPBOX_INGEST_DELETE_AFTER_PROCESS=false
# Google Drive ingest — poll a Google Drive folder (uses Google Drive credentials above)
GOOGLE_DRIVE_INGEST_ENABLED=false
GOOGLE_DRIVE_INGEST_FOLDER_ID=
GOOGLE_DRIVE_INGEST_DELETE_AFTER_PROCESS=false
# OneDrive ingest — poll a OneDrive folder (uses OneDrive MSAL credentials above)
ONEDRIVE_INGEST_ENABLED=false
ONEDRIVE_INGEST_FOLDER_PATH=
ONEDRIVE_INGEST_DELETE_AFTER_PROCESS=false
# Nextcloud ingest — poll a Nextcloud folder (uses Nextcloud WebDAV credentials above)
NEXTCLOUD_INGEST_ENABLED=false
NEXTCLOUD_INGEST_FOLDER=
NEXTCLOUD_INGEST_DELETE_AFTER_PROCESS=false
# Amazon S3 ingest — poll an S3 prefix (uses S3/AWS credentials above)
S3_INGEST_ENABLED=false
S3_INGEST_PREFIX=
S3_INGEST_DELETE_AFTER_PROCESS=false
# WebDAV ingest — poll a WebDAV folder (uses WebDAV credentials above)
WEBDAV_INGEST_ENABLED=false
WEBDAV_INGEST_FOLDER=
WEBDAV_INGEST_DELETE_AFTER_PROCESS=false
# **IMAP Settings**
# DocuElevate polls these mailboxes for new email attachments and automatically ingests them.
# No manual forwarding required — DocuElevate acts as an IMAP *client*.
# For HP Scanners / Scan-to-Email: configure the scanner to send to a dedicated mailbox,
# then point DocuElevate at that mailbox using the settings below.
IMAP1_HOST=mail.example.com
IMAP1_PORT=993
IMAP1_USERNAME=<IMAP1_USERNAME>
app/celery_worker.py
+22
View File
@@ -41,6 +41,7 @@ from app.tasks.upload_to_sftp import upload_to_sftp # noqa: F401
from app.tasks.upload_to_webdav import upload_to_webdav # noqa: F401
from app.tasks.upload_with_rclone import send_to_all_rclone_destinations, upload_with_rclone # noqa: F401
from app.tasks.uptime_kuma_tasks import ping_uptime_kuma # noqa: F401
from app.tasks.watch_folder_tasks import scan_all_watch_folders # noqa: F401
from app.tasks.webhook_tasks import deliver_webhook_task # noqa: F401
# Register the settings reload signal handler so workers pick up config changes
@@ -99,6 +100,27 @@ celery.conf.beat_schedule = {
"schedule": crontab(minute="*/1"), # Every minute
"options": {"expires": 55}, # Must complete within 55 seconds
},
# Watch folder scanning — polls local paths, FTP, SFTP, and cloud ingest folders.
# Schedule is controlled by WATCH_FOLDER_POLL_INTERVAL (default: 1 minute).
"scan-watch-folders": (
{
"task": "app.tasks.watch_folder_tasks.scan_all_watch_folders",
"schedule": crontab(minute=f"*/{max(1, settings.watch_folder_poll_interval)}"),
"options": {"expires": 55},
}
if (
settings.watch_folders
or settings.ftp_ingest_enabled
or settings.sftp_ingest_enabled
or settings.dropbox_ingest_enabled
or settings.google_drive_ingest_enabled
or settings.onedrive_ingest_enabled
or settings.nextcloud_ingest_enabled
or settings.s3_ingest_enabled
or settings.webdav_ingest_enabled
)
else None
),
# Backfill embeddings for files that were processed before the
# embedding pipeline was enabled, or where the embedding task failed.
"backfill-missing-embeddings": {
app/config.py
+185
View File
@@ -190,6 +190,191 @@ class Settings(BaseSettings):
stripe_success_url: Optional[str] = None # e.g. https://app.example.com/billing/success
stripe_cancel_url: Optional[str] = None # e.g. https://app.example.com/pricing
# ---------------------------------------------------------------------------
# Watch Folder Ingestion
# ---------------------------------------------------------------------------
# Local filesystem watch folders (comma-separated list of absolute paths).
# DocuElevate will poll each path for new files and automatically ingest them.
# Works with any mounted path, including SMB/CIFS (via system mount), NFS, etc.
# Example: /watchfolders/scanner,/mnt/shared/inbox
watch_folders: Optional[str] = Field(
default=None,
description=(
"Comma-separated list of local filesystem paths (absolute) that DocuElevate will "
"poll for new files to ingest. Each file found is enqueued for document processing. "
"Works with any mounted path including SMB/CIFS (mounted via system) and NFS. "
"Example: /watchfolders/scanner,/mnt/shared/inbox"
),
)
watch_folder_poll_interval: int = Field(
default=1,
description=("Poll interval in minutes for local watch folder scanning. Default: 1 minute."),
)
watch_folder_delete_after_process: bool = Field(
default=False,
description=(
"Delete files from local watch folders after they have been successfully enqueued "
"for processing. When False (default), files are left in place and tracked via a "
"cache file to avoid re-ingesting them."
),
)
# FTP Ingest / Watch Folder
# Uses the existing FTP credentials (ftp_host, ftp_username, ftp_password) to poll
# a source folder on the FTP server for new files to ingest.
ftp_ingest_folder: Optional[str] = Field(
default=None,
description=(
"FTP folder path to monitor for new files to ingest. "
"Uses the existing FTP connection settings (FTP_HOST, FTP_USERNAME, FTP_PASSWORD). "
"When set, DocuElevate will periodically poll this folder and download new files for processing."
),
)
ftp_ingest_enabled: bool = Field(
default=False,
description="Enable FTP watch folder ingestion. Requires FTP_INGEST_FOLDER and FTP connection settings.",
)
ftp_ingest_delete_after_process: bool = Field(
default=False,
description=(
"Delete files from the FTP ingest folder after they have been successfully downloaded "
"and enqueued for processing. Default: False (files are left in place)."
),
)
# SFTP Ingest / Watch Folder
# Uses the existing SFTP credentials (sftp_host, sftp_username, sftp_password/sftp_private_key)
# to poll a source folder on the SFTP server for new files to ingest.
sftp_ingest_folder: Optional[str] = Field(
default=None,
description=(
"SFTP folder path to monitor for new files to ingest. "
"Uses the existing SFTP connection settings (SFTP_HOST, SFTP_USERNAME, SFTP_PASSWORD/SFTP_PRIVATE_KEY). "
"When set, DocuElevate will periodically poll this folder and download new files for processing."
),
)
sftp_ingest_enabled: bool = Field(
default=False,
description="Enable SFTP watch folder ingestion. Requires SFTP_INGEST_FOLDER and SFTP connection settings.",
)
sftp_ingest_delete_after_process: bool = Field(
default=False,
description=(
"Delete files from the SFTP ingest folder after they have been successfully downloaded "
"and enqueued for processing. Default: False (files are left in place)."
),
)
# ---------------------------------------------------------------------------
# Cloud Provider Watch Folders
# ---------------------------------------------------------------------------
# Each cloud provider has three settings:
# <provider>_ingest_enabled — enable the watch-folder for this provider
# <provider>_ingest_folder — the remote path / folder ID to poll
# <provider>_ingest_delete_after_process — delete from cloud after download
# Dropbox ingest — reuses existing Dropbox OAuth credentials
dropbox_ingest_enabled: bool = Field(
default=False,
description="Enable Dropbox watch folder ingestion. Requires Dropbox OAuth credentials.",
)
dropbox_ingest_folder: Optional[str] = Field(
default=None,
description=(
"Dropbox folder path to poll for new files to ingest (e.g. /Inbox/Scanner). "
"Uses the existing Dropbox OAuth credentials."
),
)
dropbox_ingest_delete_after_process: bool = Field(
default=False,
description="Delete files from Dropbox ingest folder after download and enqueue.",
)
# Google Drive ingest — reuses existing Google Drive credentials
google_drive_ingest_enabled: bool = Field(
default=False,
description="Enable Google Drive watch folder ingestion. Requires Google Drive credentials.",
)
google_drive_ingest_folder_id: Optional[str] = Field(
default=None,
description=(
"Google Drive folder ID to poll for new files to ingest. "
"Uses the existing Google Drive service-account or OAuth credentials."
),
)
google_drive_ingest_delete_after_process: bool = Field(
default=False,
description="Delete files from Google Drive ingest folder after download and enqueue.",
)
# OneDrive ingest — reuses existing OneDrive MSAL credentials
onedrive_ingest_enabled: bool = Field(
default=False,
description="Enable OneDrive watch folder ingestion. Requires OneDrive MSAL credentials.",
)
onedrive_ingest_folder_path: Optional[str] = Field(
default=None,
description=(
"OneDrive folder path to poll for new files to ingest (e.g. /Inbox/Scanner). "
"Uses the existing OneDrive client credentials."
),
)
onedrive_ingest_delete_after_process: bool = Field(
default=False,
description="Delete files from OneDrive ingest folder after download and enqueue.",
)
# Nextcloud ingest — reuses existing Nextcloud WebDAV credentials
nextcloud_ingest_enabled: bool = Field(
default=False,
description="Enable Nextcloud watch folder ingestion. Requires Nextcloud WebDAV credentials.",
)
nextcloud_ingest_folder: Optional[str] = Field(
default=None,
description=(
"Nextcloud folder path to poll for new files to ingest (e.g. /Scans/Inbox). "
"Uses the existing Nextcloud upload URL and credentials."
),
)
nextcloud_ingest_delete_after_process: bool = Field(
default=False,
description="Delete files from Nextcloud ingest folder after download and enqueue.",
)
# S3 ingest — reuses existing AWS/S3 credentials
s3_ingest_enabled: bool = Field(
default=False,
description="Enable Amazon S3 watch folder (prefix) ingestion. Requires S3 credentials.",
)
s3_ingest_prefix: Optional[str] = Field(
default=None,
description=(
"S3 key prefix to poll for new objects to ingest (e.g. inbox/scanner/). "
"Uses the existing S3 bucket and AWS credentials."
),
)
s3_ingest_delete_after_process: bool = Field(
default=False,
description="Delete objects from S3 ingest prefix after download and enqueue.",
)
# WebDAV ingest — reuses existing WebDAV credentials
webdav_ingest_enabled: bool = Field(
default=False,
description="Enable WebDAV watch folder ingestion. Requires WebDAV URL and credentials.",
)
webdav_ingest_folder: Optional[str] = Field(
default=None,
description=(
"WebDAV folder path to poll for new files to ingest (e.g. /remote.php/webdav/Inbox). "
"Uses the existing WebDAV URL and credentials."
),
)
webdav_ingest_delete_after_process: bool = Field(
default=False,
description="Delete files from WebDAV ingest folder after download and enqueue.",
)
# IMAP 1
imap1_host: Optional[str] = None
imap1_port: Optional[int] = 993
app/tasks/watch_folder_tasks.py
+1325
View File
File diff suppressed because it is too large Load Diff
app/utils/config_validator/settings_display.py
+29
View File
@@ -106,6 +106,35 @@ def get_settings_for_display(show_values: bool = False) -> dict[str, list[dict[s
"email_sender",
"email_default_recipient",
],
"Watch Folders": [
"watch_folders",
"watch_folder_poll_interval",
"watch_folder_delete_after_process",
"ftp_ingest_enabled",
"ftp_ingest_folder",
"ftp_ingest_delete_after_process",
"sftp_ingest_enabled",
"sftp_ingest_folder",
"sftp_ingest_delete_after_process",
"dropbox_ingest_enabled",
"dropbox_ingest_folder",
"dropbox_ingest_delete_after_process",
"google_drive_ingest_enabled",
"google_drive_ingest_folder_id",
"google_drive_ingest_delete_after_process",
"onedrive_ingest_enabled",
"onedrive_ingest_folder_path",
"onedrive_ingest_delete_after_process",
"nextcloud_ingest_enabled",
"nextcloud_ingest_folder",
"nextcloud_ingest_delete_after_process",
"s3_ingest_enabled",
"s3_ingest_prefix",
"s3_ingest_delete_after_process",
"webdav_ingest_enabled",
"webdav_ingest_folder",
"webdav_ingest_delete_after_process",
],
"IMAP": [
"imap1_host",
"imap1_port",
app/utils/settings_service.py
+238
View File
@@ -971,6 +971,244 @@ SETTING_METADATA = {
"required": False,
"restart_required": False,
},
# Watch Folder / Ingest Settings
"watch_folders": {
"category": "Watch Folders",
"description": (
"Comma-separated list of absolute local filesystem paths that DocuElevate will "
"poll for new files to ingest. Works with any mounted path (SMB/CIFS, NFS, etc.). "
"Example: /watchfolders/scanner,/mnt/shared/inbox"
),
"type": "string",
"sensitive": False,
"required": False,
"restart_required": False,
},
"watch_folder_poll_interval": {
"category": "Watch Folders",
"description": "Poll interval in minutes for local watch folder scanning (default: 1)",
"type": "integer",
"sensitive": False,
"required": False,
"restart_required": False,
},
"watch_folder_delete_after_process": {
"category": "Watch Folders",
"description": (
"Delete files from local watch folders after they have been enqueued for processing. "
"When False (default), processed files are tracked via cache to avoid re-ingestion."
),
"type": "boolean",
"sensitive": False,
"required": False,
"restart_required": False,
},
# FTP Ingest
"ftp_ingest_enabled": {
"category": "Watch Folders",
"description": "Enable FTP watch folder ingestion (requires FTP_INGEST_FOLDER and FTP connection settings)",
"type": "boolean",
"sensitive": False,
"required": False,
"restart_required": False,
},
"ftp_ingest_folder": {
"category": "Watch Folders",
"description": (
"FTP folder path to poll for new documents to ingest. "
"Uses the existing FTP connection settings (FTP_HOST, FTP_USERNAME, FTP_PASSWORD)."
),
"type": "string",
"sensitive": False,
"required": False,
"restart_required": False,
},
"ftp_ingest_delete_after_process": {
"category": "Watch Folders",
"description": "Delete files from the FTP ingest folder after they have been downloaded and enqueued",
"type": "boolean",
"sensitive": False,
"required": False,
"restart_required": False,
},
# SFTP Ingest
"sftp_ingest_enabled": {
"category": "Watch Folders",
"description": "Enable SFTP watch folder ingestion (requires SFTP_INGEST_FOLDER and SFTP connection settings)",
"type": "boolean",
"sensitive": False,
"required": False,
"restart_required": False,
},
"sftp_ingest_folder": {
"category": "Watch Folders",
"description": (
"SFTP folder path to poll for new documents to ingest. "
"Uses the existing SFTP connection settings (SFTP_HOST, SFTP_USERNAME, SFTP_PASSWORD/SFTP_PRIVATE_KEY)."
),
"type": "string",
"sensitive": False,
"required": False,
"restart_required": False,
},
"sftp_ingest_delete_after_process": {
"category": "Watch Folders",
"description": "Delete files from the SFTP ingest folder after they have been downloaded and enqueued",
"type": "boolean",
"sensitive": False,
"required": False,
"restart_required": False,
},
# Cloud Provider Watch Folders — Dropbox
"dropbox_ingest_enabled": {
"category": "Watch Folders",
"description": "Enable Dropbox watch folder ingestion (requires Dropbox OAuth credentials)",
"type": "boolean",
"sensitive": False,
"required": False,
"restart_required": False,
},
"dropbox_ingest_folder": {
"category": "Watch Folders",
"description": "Dropbox folder path to poll for new documents (e.g. /Inbox/Scanner). Uses existing Dropbox credentials.",
"type": "string",
"sensitive": False,
"required": False,
"restart_required": False,
},
"dropbox_ingest_delete_after_process": {
"category": "Watch Folders",
"description": "Delete files from Dropbox ingest folder after download and enqueue",
"type": "boolean",
"sensitive": False,
"required": False,
"restart_required": False,
},
# Cloud Provider Watch Folders — Google Drive
"google_drive_ingest_enabled": {
"category": "Watch Folders",
"description": "Enable Google Drive watch folder ingestion (requires Google Drive credentials)",
"type": "boolean",
"sensitive": False,
"required": False,
"restart_required": False,
},
"google_drive_ingest_folder_id": {
"category": "Watch Folders",
"description": "Google Drive folder ID to poll for new documents. Uses existing Google Drive credentials.",
"type": "string",
"sensitive": False,
"required": False,
"restart_required": False,
},
"google_drive_ingest_delete_after_process": {
"category": "Watch Folders",
"description": "Delete files from Google Drive ingest folder after download and enqueue",
"type": "boolean",
"sensitive": False,
"required": False,
"restart_required": False,
},
# Cloud Provider Watch Folders — OneDrive
"onedrive_ingest_enabled": {
"category": "Watch Folders",
"description": "Enable OneDrive watch folder ingestion (requires OneDrive MSAL credentials)",
"type": "boolean",
"sensitive": False,
"required": False,
"restart_required": False,
},
"onedrive_ingest_folder_path": {
"category": "Watch Folders",
"description": "OneDrive folder path to poll for new documents (e.g. /Inbox/Scanner). Uses existing OneDrive credentials.",
"type": "string",
"sensitive": False,
"required": False,
"restart_required": False,
},
"onedrive_ingest_delete_after_process": {
"category": "Watch Folders",
"description": "Delete files from OneDrive ingest folder after download and enqueue",
"type": "boolean",
"sensitive": False,
"required": False,
"restart_required": False,
},
# Cloud Provider Watch Folders — Nextcloud
"nextcloud_ingest_enabled": {
"category": "Watch Folders",
"description": "Enable Nextcloud watch folder ingestion (requires Nextcloud WebDAV credentials)",
"type": "boolean",
"sensitive": False,
"required": False,
"restart_required": False,
},
"nextcloud_ingest_folder": {
"category": "Watch Folders",
"description": "Nextcloud folder path to poll for new documents (e.g. /Scans/Inbox). Uses existing Nextcloud credentials.",
"type": "string",
"sensitive": False,
"required": False,
"restart_required": False,
},
"nextcloud_ingest_delete_after_process": {
"category": "Watch Folders",
"description": "Delete files from Nextcloud ingest folder after download and enqueue",
"type": "boolean",
"sensitive": False,
"required": False,
"restart_required": False,
},
# Cloud Provider Watch Folders — S3
"s3_ingest_enabled": {
"category": "Watch Folders",
"description": "Enable Amazon S3 prefix (watch folder) ingestion (requires S3/AWS credentials)",
"type": "boolean",
"sensitive": False,
"required": False,
"restart_required": False,
},
"s3_ingest_prefix": {
"category": "Watch Folders",
"description": "S3 key prefix to poll for new objects to ingest (e.g. inbox/scanner/). Uses existing S3 credentials.",
"type": "string",
"sensitive": False,
"required": False,
"restart_required": False,
},
"s3_ingest_delete_after_process": {
"category": "Watch Folders",
"description": "Delete objects from S3 ingest prefix after download and enqueue",
"type": "boolean",
"sensitive": False,
"required": False,
"restart_required": False,
},
# Cloud Provider Watch Folders — WebDAV
"webdav_ingest_enabled": {
"category": "Watch Folders",
"description": "Enable WebDAV watch folder ingestion (requires WebDAV URL and credentials)",
"type": "boolean",
"sensitive": False,
"required": False,
"restart_required": False,
},
"webdav_ingest_folder": {
"category": "Watch Folders",
"description": "WebDAV folder path to poll for new documents. Uses existing WebDAV URL and credentials.",
"type": "string",
"sensitive": False,
"required": False,
"restart_required": False,
},
"webdav_ingest_delete_after_process": {
"category": "Watch Folders",
"description": "Delete files from WebDAV ingest folder after download and enqueue",
"type": "boolean",
"sensitive": False,
"required": False,
"restart_required": False,
},
# IMAP Settings - Account 1
"imap1_host": {
"category": "IMAP",
docs/ConfigurationGuide.md
+150 -2
View File
@@ -121,9 +121,157 @@ MAX_SINGLE_FILE_SIZE=524288000
- **With splitting**: Recommended for servers with limited memory or when processing very large scanned documents
- **Higher limits**: For environments specifically designed to handle large architectural plans, books, or scanned archives
### IMAP Configuration
### Watch Folder Ingestion
DocuElevate can monitor multiple IMAP mailboxes for document attachments. Each mailbox uses a numbered prefix (e.g., `IMAP1_`, `IMAP2_`).
DocuElevate can automatically monitor directories for new files and ingest them without any manual action.
This works for:
- **Local filesystem paths** — including SMB/CIFS shares, NFS mounts, or any path accessible to the Docker container
- **FTP server directories** — using the configured FTP connection credentials
- **SFTP server directories** — using the configured SFTP connection credentials
#### Local Watch Folders
Mount the share or directory into the Docker container and configure one or more paths to watch.
| **Variable** | **Description** | **Default** |
|-------------------------------------|----------------------------------------------------------------------------------------------|-------------|
| `WATCH_FOLDERS` | Comma-separated list of **absolute** local filesystem paths to poll for new files. | *(empty)* |
| `WATCH_FOLDER_POLL_INTERVAL` | How often to scan the folders, in minutes. | `1` |
| `WATCH_FOLDER_DELETE_AFTER_PROCESS` | Delete source files from the watch folder after they are successfully enqueued. When `false`, processed files are tracked in a cache file to prevent re-ingestion. | `false` |
**Example (docker-compose.yaml):**
```yaml
services:
worker:
volumes:
- /mnt/smb/scanner:/watchfolders/scanner # SMB/CIFS share mounted on the host
- /mnt/nfs/inbox:/watchfolders/inbox # NFS mount
environment:
WATCH_FOLDERS: /watchfolders/scanner,/watchfolders/inbox
WATCH_FOLDER_POLL_INTERVAL: 1
WATCH_FOLDER_DELETE_AFTER_PROCESS: false
```
> **Tip for HP Scanners and MFPs**: Configure your scanner's "Scan to Network Folder" to point at an SMB share that is also mounted into the DocuElevate worker container. DocuElevate will pick up the scan files automatically every minute. No email forwarding is required.
#### FTP Ingest (Watch Folder)
DocuElevate can poll an FTP server directory for new files. It reuses the FTP connection settings already configured for uploads.
| **Variable** | **Description** | **Default** |
|-----------------------------------|------------------------------------------------------------------------------------------------|-------------|
| `FTP_INGEST_ENABLED` | Enable FTP folder watching (`true`/`false`). | `false` |
| `FTP_INGEST_FOLDER` | Path on the FTP server to poll (e.g. `/incoming`). Uses the existing FTP connection settings. | *(empty)* |
| `FTP_INGEST_DELETE_AFTER_PROCESS` | Delete files from the FTP server after they are downloaded and enqueued. | `false` |
**Example:**
```dotenv
# Existing FTP upload settings (also used for ingest)
FTP_HOST=ftp.example.com
FTP_USERNAME=docuelevate
FTP_PASSWORD=secret
# FTP ingest configuration
FTP_INGEST_ENABLED=true
FTP_INGEST_FOLDER=/incoming
FTP_INGEST_DELETE_AFTER_PROCESS=false
```
#### SFTP Ingest (Watch Folder)
DocuElevate can poll an SFTP server directory for new files. It reuses the SFTP connection settings already configured for uploads.
| **Variable** | **Description** | **Default** |
|------------------------------------|-------------------------------------------------------------------------------------------------|-------------|
| `SFTP_INGEST_ENABLED` | Enable SFTP folder watching (`true`/`false`). | `false` |
| `SFTP_INGEST_FOLDER` | Path on the SFTP server to poll (e.g. `/uploads/inbox`). Uses the existing SFTP connection settings. | *(empty)* |
| `SFTP_INGEST_DELETE_AFTER_PROCESS` | Delete files from the SFTP server after they are downloaded and enqueued. | `false` |
**Example:**
```dotenv
# Existing SFTP upload settings (also used for ingest)
SFTP_HOST=sftp.example.com
SFTP_USERNAME=docuelevate
SFTP_PRIVATE_KEY=/run/secrets/sftp_key
# SFTP ingest configuration
SFTP_INGEST_ENABLED=true
SFTP_INGEST_FOLDER=/uploads/inbox
SFTP_INGEST_DELETE_AFTER_PROCESS=false
```
#### Supported File Types for Watch Folders
Watch folder ingestion accepts the same file types as the web upload interface: PDF, Word, Excel, PowerPoint, images (JPEG, PNG, TIFF, BMP, GIF), plain text, CSV, RTF, and more. Unsupported files (executables, archives, etc.) are silently skipped.
#### Dropbox Ingest (Watch Folder)
DocuElevate can poll a Dropbox folder for new files. It reuses the Dropbox OAuth credentials already configured for uploads.
| **Variable** | **Description** | **Default** |
|---------------------------------------|----------------------------------------------------------------------------------------------|-------------|
| `DROPBOX_INGEST_ENABLED` | Enable Dropbox folder watching (`true`/`false`). | `false` |
| `DROPBOX_INGEST_FOLDER` | Dropbox folder path to poll (e.g. `/Inbox/Scanner`). Uses the existing Dropbox OAuth credentials. | *(empty)* |
| `DROPBOX_INGEST_DELETE_AFTER_PROCESS` | Delete files from Dropbox after they are downloaded and enqueued. | `false` |
#### Google Drive Ingest (Watch Folder)
DocuElevate can poll a Google Drive folder for new files. It reuses the existing Google Drive service-account or OAuth credentials.
| **Variable** | **Description** | **Default** |
|----------------------------------------------|----------------------------------------------------------------------------------------------|-------------|
| `GOOGLE_DRIVE_INGEST_ENABLED` | Enable Google Drive folder watching (`true`/`false`). | `false` |
| `GOOGLE_DRIVE_INGEST_FOLDER_ID` | Google Drive **folder ID** to poll (copy from the URL of the target folder in Drive). Uses the existing Google Drive credentials. | *(empty)* |
| `GOOGLE_DRIVE_INGEST_DELETE_AFTER_PROCESS` | Delete files from Google Drive after they are downloaded and enqueued. | `false` |
#### OneDrive Ingest (Watch Folder)
DocuElevate can poll a OneDrive folder for new files. It reuses the existing OneDrive MSAL (client ID/secret/refresh token) credentials.
| **Variable** | **Description** | **Default** |
|--------------------------------------------|----------------------------------------------------------------------------------------------|-------------|
| `ONEDRIVE_INGEST_ENABLED` | Enable OneDrive folder watching (`true`/`false`). | `false` |
| `ONEDRIVE_INGEST_FOLDER_PATH` | OneDrive folder path to poll (e.g. `/Inbox/Scanner`). Uses the existing OneDrive credentials. | *(empty)* |
| `ONEDRIVE_INGEST_DELETE_AFTER_PROCESS` | Delete files from OneDrive after they are downloaded and enqueued. | `false` |
#### Nextcloud Ingest (Watch Folder)
DocuElevate can poll a Nextcloud folder via WebDAV for new files. It reuses the existing Nextcloud upload URL and credentials.
| **Variable** | **Description** | **Default** |
|--------------------------------------------|----------------------------------------------------------------------------------------------|-------------|
| `NEXTCLOUD_INGEST_ENABLED` | Enable Nextcloud folder watching (`true`/`false`). | `false` |
| `NEXTCLOUD_INGEST_FOLDER` | Nextcloud folder path to poll (e.g. `/Scans/Inbox`). Uses the existing Nextcloud upload URL and credentials. | *(empty)* |
| `NEXTCLOUD_INGEST_DELETE_AFTER_PROCESS` | Delete files from Nextcloud after they are downloaded and enqueued. | `false` |
#### Amazon S3 Ingest (Watch Folder)
DocuElevate can poll an S3 bucket prefix for new objects. It reuses the existing S3/AWS credentials and bucket name.
| **Variable** | **Description** | **Default** |
|---------------------------------------|----------------------------------------------------------------------------------------------|-------------|
| `S3_INGEST_ENABLED` | Enable S3 prefix watching (`true`/`false`). | `false` |
| `S3_INGEST_PREFIX` | S3 key prefix to poll (e.g. `inbox/scanner/`). Uses the existing S3 bucket and AWS credentials. | *(empty)* |
| `S3_INGEST_DELETE_AFTER_PROCESS` | Delete objects from S3 after they are downloaded and enqueued. | `false` |
#### WebDAV Ingest (Watch Folder)
DocuElevate can poll a WebDAV folder for new files. It reuses the existing WebDAV URL and credentials.
| **Variable** | **Description** | **Default** |
|---------------------------------------|----------------------------------------------------------------------------------------------|-------------|
| `WEBDAV_INGEST_ENABLED` | Enable WebDAV folder watching (`true`/`false`). | `false` |
| `WEBDAV_INGEST_FOLDER` | WebDAV folder path to poll. Uses the existing WebDAV URL and credentials. | *(empty)* |
| `WEBDAV_INGEST_DELETE_AFTER_PROCESS` | Delete files from WebDAV after they are downloaded and enqueued. | `false` |
### IMAP Email Ingestion
DocuElevate can automatically pull document attachments from IMAP mailboxes — no need to forward emails manually. Configure one or two mailboxes and DocuElevate polls them on the schedule you set.
> **For HP Scanners (Scan to Email)**: If your scanner is set up to email scanned documents to a dedicated mailbox, configure that mailbox in DocuElevate using the settings below. DocuElevate will automatically retrieve the scanned PDFs from the inbox and process them. You do **not** need to configure DocuElevate as an email server — it acts as an email *client* that reads from your existing mailbox.
| **Variable** | **Description** | **Example** |
|-------------------------------|--------------------------------------------------------------|-------------------|
docs/UserGuide.md
+29 -6
View File
@@ -74,14 +74,37 @@ For even more convenience, you can upload files directly from the **Files** page
This feature allows you to quickly add new files without navigating away from your document management view.
### Email Attachments
### Email Attachments (IMAP Ingestion)
If configured, DocuElevate can automatically fetch documents from email attachments:
DocuElevate acts as an email *client* that automatically retrieves document attachments from one or more IMAP mailboxes. You do not need to set up DocuElevate as an email server — it simply polls an existing mailbox that you designate for document delivery.
1. Send an email with attachments to the configured email account
2. DocuElevate will poll the mailbox at the configured interval
3. Attachments will be automatically downloaded and processed
4. No further action is required
**How it works:**
1. A document is sent as an email attachment to the configured mailbox (e.g. from a scanner, a colleague, or any email client)
2. DocuElevate polls the mailbox at the configured interval (typically every 1–5 minutes)
3. Email attachments in supported formats are automatically downloaded and enqueued for processing
4. Processed emails are marked with a label or star (Gmail) or tracked locally, so they are not re-processed
> **HP Scanners and MFPs (Scan to Email)**: Configure your scanner's "Scan to Email" feature to send scanned documents to a dedicated email account. Point DocuElevate at that mailbox using the IMAP settings. DocuElevate will retrieve the scanned PDFs automatically — no manual forwarding required.
### Watch Folders (Automatic Folder Ingestion)
Watch folders allow DocuElevate to automatically monitor directories for new files and ingest them without any manual action.
#### Local Watch Folders (including SMB/CIFS and NFS)
Mount a network share or local directory into the DocuElevate worker container and configure the path in `WATCH_FOLDERS`. DocuElevate scans the folder every minute (configurable via `WATCH_FOLDER_POLL_INTERVAL`) and enqueues any new documents it finds.
This is the recommended approach for:
- **HP Scanners / MFPs** using "Scan to Network Folder" — point the scanner at a shared folder that DocuElevate also has access to
- **SMB/CIFS shares** — mount the Windows/Samba share and add the path to `WATCH_FOLDERS`
- **NFS mounts** — works identically, just configure the mount path
- **Any local directory** on the server running DocuElevate
#### FTP / SFTP Watch Folders
DocuElevate can poll an FTP or SFTP directory for new files. Enable this with `FTP_INGEST_ENABLED` or `SFTP_INGEST_ENABLED` and set the corresponding ingest folder. DocuElevate downloads new files, enqueues them for processing, and optionally deletes them from the remote server.
See [Configuration Guide — Watch Folder Ingestion](ConfigurationGuide.md#watch-folder-ingestion) for full setup instructions.
## Managing Documents
frontend/templates/settings.html
+1
View File
@@ -9,6 +9,7 @@
'OCR Engines': 'fas fa-file-alt',
'Storage Providers': 'fas fa-cloud-upload-alt',
'Email': 'fas fa-envelope',
'Watch Folders': 'fas fa-folder-open',
'IMAP': 'fas fa-inbox',
'Monitoring': 'fas fa-chart-line',
'Processing': 'fas fa-cogs',
requirements.txt
+3
View File
@@ -33,6 +33,9 @@ boto3>=1.28.0
# SFTP
paramiko>=3.4.0 # SSH/SFTP implementation for Python (LGPL license)
# Safe XML parsing (protection against XML bomb / XXE attacks)
defusedxml>=0.7.1
# Notification service
apprise>=1.4.0
tests/test_watch_folder_tasks.py
+779
View File
@@ -0,0 +1,779 @@
"""Tests for app/tasks/watch_folder_tasks.py module."""
import os
from datetime import datetime, timedelta, timezone
from unittest.mock import MagicMock, patch
import pytest
@pytest.mark.unit
class TestCacheHelpers:
"""Tests for the cache helper functions in watch_folder_tasks."""
def test_evict_old_entries_removes_old(self):
"""Entries older than CACHE_RETENTION_DAYS should be removed."""
from app.tasks.watch_folder_tasks import _evict_old_entries
old_dt = (datetime.now(timezone.utc) - timedelta(days=40)).isoformat()
recent_dt = datetime.now(timezone.utc).isoformat()
data = {"old_key": old_dt, "new_key": recent_dt}
result = _evict_old_entries(data)
assert "old_key" not in result
assert "new_key" in result
def test_evict_old_entries_empty(self):
"""Empty cache should return empty dict."""
from app.tasks.watch_folder_tasks import _evict_old_entries
assert _evict_old_entries({}) == {}
def test_evict_old_entries_skips_malformed_dates(self):
"""Entries with malformed timestamps should be silently dropped."""
from app.tasks.watch_folder_tasks import _evict_old_entries
data = {"bad_key": "not-a-date", "good_key": datetime.now(timezone.utc).isoformat()}
result = _evict_old_entries(data)
assert "bad_key" not in result
assert "good_key" in result
def test_mark_processed_adds_entry(self):
"""_mark_processed should add an ISO-formatted timestamp for the key."""
from app.tasks.watch_folder_tasks import _mark_processed
cache: dict = {}
_mark_processed(cache, "/some/file.pdf")
assert "/some/file.pdf" in cache
# Timestamp should be parseable
datetime.fromisoformat(cache["/some/file.pdf"])
def test_load_cache_returns_empty_when_no_file(self):
"""_load_cache should return {} when the cache file does not exist."""
from app.tasks.watch_folder_tasks import _load_cache
result = _load_cache("/tmp/does_not_exist_xyz.json")
assert result == {}
def test_save_and_load_roundtrip(self, tmp_path):
"""Saving and loading cache should preserve entries."""
from app.tasks.watch_folder_tasks import _load_cache, _save_cache
cache_file = str(tmp_path / "cache.json")
data = {"key1": datetime.now(timezone.utc).isoformat()}
_save_cache(cache_file, data)
loaded = _load_cache(cache_file)
assert "key1" in loaded
def test_load_cache_handles_invalid_json(self, tmp_path):
"""_load_cache should return {} for corrupted JSON files."""
from app.tasks.watch_folder_tasks import _load_cache
cache_file = str(tmp_path / "bad.json")
with open(cache_file, "w") as f:
f.write("not valid json{{{{")
result = _load_cache(cache_file)
assert result == {}
@pytest.mark.unit
class TestIsAllowedFile:
"""Tests for the _is_allowed_file helper."""
def test_pdf_is_allowed(self):
from app.tasks.watch_folder_tasks import _is_allowed_file
assert _is_allowed_file("document.pdf") is True
assert _is_allowed_file("DOCUMENT.PDF") is True
def test_docx_is_allowed(self):
from app.tasks.watch_folder_tasks import _is_allowed_file
assert _is_allowed_file("report.docx") is True
def test_exe_is_not_allowed(self):
from app.tasks.watch_folder_tasks import _is_allowed_file
assert _is_allowed_file("malware.exe") is False
def test_zip_is_not_allowed(self):
from app.tasks.watch_folder_tasks import _is_allowed_file
assert _is_allowed_file("archive.zip") is False
@pytest.mark.unit
class TestScanLocalFolder:
"""Tests for _scan_local_folder."""
def test_nonexistent_folder_returns_zero(self):
from app.tasks.watch_folder_tasks import _scan_local_folder
count = _scan_local_folder("/tmp/does_not_exist_xyz_abc", {}, False)
assert count == 0
def test_empty_folder_returns_zero(self, tmp_path):
from app.tasks.watch_folder_tasks import _scan_local_folder
count = _scan_local_folder(str(tmp_path), {}, False)
assert count == 0
def test_new_pdf_is_enqueued(self, tmp_path):
"""A new PDF in the watch folder should be enqueued for processing."""
from app.tasks.watch_folder_tasks import _scan_local_folder
pdf_file = tmp_path / "test.pdf"
pdf_file.write_bytes(b"%PDF-1.4 test")
cache: dict = {}
with (
patch("app.tasks.watch_folder_tasks.process_document") as mock_proc,
patch("app.tasks.watch_folder_tasks.settings") as mock_settings,
):
mock_settings.workdir = str(tmp_path / "workdir")
os.makedirs(mock_settings.workdir, exist_ok=True)
count = _scan_local_folder(str(tmp_path), cache, False)
assert count == 1
assert str(pdf_file) in cache
def test_already_cached_file_is_skipped(self, tmp_path):
"""Files already in cache should not be re-processed."""
from app.tasks.watch_folder_tasks import _scan_local_folder
pdf_file = tmp_path / "already.pdf"
pdf_file.write_bytes(b"%PDF-1.4 test")
cache = {str(pdf_file): datetime.now(timezone.utc).isoformat()}
with patch("app.tasks.watch_folder_tasks.settings") as mock_settings:
mock_settings.workdir = str(tmp_path / "workdir")
os.makedirs(mock_settings.workdir, exist_ok=True)
count = _scan_local_folder(str(tmp_path), cache, False)
assert count == 0
def test_unsupported_file_is_skipped(self, tmp_path):
"""Unsupported file types should not be enqueued."""
from app.tasks.watch_folder_tasks import _scan_local_folder
exe_file = tmp_path / "bad.exe"
exe_file.write_bytes(b"MZ malware")
cache: dict = {}
with patch("app.tasks.watch_folder_tasks.settings") as mock_settings:
mock_settings.workdir = str(tmp_path / "workdir")
count = _scan_local_folder(str(tmp_path), cache, False)
assert count == 0
assert str(exe_file) not in cache
def test_delete_after_process_removes_source(self, tmp_path):
"""When delete_after_process=True, source files should be deleted."""
from app.tasks.watch_folder_tasks import _scan_local_folder
pdf_file = tmp_path / "invoice.pdf"
pdf_file.write_bytes(b"%PDF-1.4")
cache: dict = {}
with (
patch("app.tasks.watch_folder_tasks.process_document"),
patch("app.tasks.watch_folder_tasks.settings") as mock_settings,
):
mock_settings.workdir = str(tmp_path / "workdir")
os.makedirs(mock_settings.workdir, exist_ok=True)
_scan_local_folder(str(tmp_path), cache, delete_after=True)
assert not pdf_file.exists()
@pytest.mark.unit
class TestAcquireReleaseLock:
"""Tests for the Redis-based locking helpers."""
def test_acquire_lock_succeeds(self):
from app.tasks.watch_folder_tasks import _acquire_lock
mock_redis = MagicMock()
mock_redis.setnx.return_value = True
with patch("app.tasks.watch_folder_tasks.redis_client", mock_redis):
result = _acquire_lock("test_lock")
assert result is True
def test_acquire_lock_fails_when_held(self):
from app.tasks.watch_folder_tasks import _acquire_lock
mock_redis = MagicMock()
mock_redis.setnx.return_value = False
with patch("app.tasks.watch_folder_tasks.redis_client", mock_redis):
result = _acquire_lock("test_lock")
assert result is False
def test_release_lock_deletes_key(self):
from app.tasks.watch_folder_tasks import _release_lock
mock_redis = MagicMock()
with patch("app.tasks.watch_folder_tasks.redis_client", mock_redis):
_release_lock("test_lock")
mock_redis.delete.assert_called_once_with("test_lock")
@pytest.mark.unit
class TestScanLocalWatchFoldersTask:
"""Tests for the scan_local_watch_folders Celery task."""
def test_returns_skipped_when_no_folders_configured(self):
from app.tasks.watch_folder_tasks import scan_local_watch_folders
with patch("app.tasks.watch_folder_tasks.settings") as mock_settings:
mock_settings.watch_folders = None
result = scan_local_watch_folders()
assert result["status"] == "skipped"
def test_returns_skipped_for_empty_string(self):
from app.tasks.watch_folder_tasks import scan_local_watch_folders
with patch("app.tasks.watch_folder_tasks.settings") as mock_settings:
mock_settings.watch_folders = ""
result = scan_local_watch_folders()
assert result["status"] == "skipped"
def test_scans_configured_folder(self, tmp_path):
"""With a valid folder configured, the task should scan it."""
from app.tasks.watch_folder_tasks import scan_local_watch_folders
with (
patch("app.tasks.watch_folder_tasks.settings") as mock_settings,
patch("app.tasks.watch_folder_tasks._load_cache", return_value={}),
patch("app.tasks.watch_folder_tasks._save_cache"),
patch("app.tasks.watch_folder_tasks._scan_local_folder", return_value=0) as mock_scan,
):
mock_settings.watch_folders = str(tmp_path)
mock_settings.watch_folder_delete_after_process = False
result = scan_local_watch_folders()
mock_scan.assert_called_once()
assert result["status"] == "ok"
assert result["files_enqueued"] == 0
@pytest.mark.unit
class TestScanFtpWatchFolderTask:
"""Tests for the scan_ftp_watch_folder Celery task."""
def test_returns_skipped_when_disabled(self):
from app.tasks.watch_folder_tasks import scan_ftp_watch_folder
with patch("app.tasks.watch_folder_tasks.settings") as mock_settings:
mock_settings.ftp_ingest_enabled = False
result = scan_ftp_watch_folder()
assert result["status"] == "skipped"
def test_returns_skipped_when_no_folder_configured(self):
from app.tasks.watch_folder_tasks import scan_ftp_watch_folder
with patch("app.tasks.watch_folder_tasks.settings") as mock_settings:
mock_settings.ftp_ingest_enabled = True
mock_settings.ftp_ingest_folder = None
result = scan_ftp_watch_folder()
assert result["status"] == "skipped"
def test_returns_error_when_connection_fails(self):
from app.tasks.watch_folder_tasks import scan_ftp_watch_folder
with (
patch("app.tasks.watch_folder_tasks.settings") as mock_settings,
patch("app.tasks.watch_folder_tasks._connect_ftp", return_value=None),
):
mock_settings.ftp_ingest_enabled = True
mock_settings.ftp_ingest_folder = "/inbox"
mock_settings.ftp_ingest_delete_after_process = False
result = scan_ftp_watch_folder()
assert result["status"] == "error"
def test_successful_scan(self):
from app.tasks.watch_folder_tasks import scan_ftp_watch_folder
mock_ftp = MagicMock()
with (
patch("app.tasks.watch_folder_tasks.settings") as mock_settings,
patch("app.tasks.watch_folder_tasks._connect_ftp", return_value=mock_ftp),
patch("app.tasks.watch_folder_tasks._load_cache", return_value={}),
patch("app.tasks.watch_folder_tasks._save_cache"),
patch("app.tasks.watch_folder_tasks._scan_ftp_folder", return_value=2),
):
mock_settings.ftp_ingest_enabled = True
mock_settings.ftp_ingest_folder = "/inbox"
mock_settings.ftp_ingest_delete_after_process = False
result = scan_ftp_watch_folder()
assert result["status"] == "ok"
assert result["files_enqueued"] == 2
@pytest.mark.unit
class TestScanSftpWatchFolderTask:
"""Tests for the scan_sftp_watch_folder Celery task."""
def test_returns_skipped_when_disabled(self):
from app.tasks.watch_folder_tasks import scan_sftp_watch_folder
with patch("app.tasks.watch_folder_tasks.settings") as mock_settings:
mock_settings.sftp_ingest_enabled = False
result = scan_sftp_watch_folder()
assert result["status"] == "skipped"
def test_returns_skipped_when_no_folder_configured(self):
from app.tasks.watch_folder_tasks import scan_sftp_watch_folder
with patch("app.tasks.watch_folder_tasks.settings") as mock_settings:
mock_settings.sftp_ingest_enabled = True
mock_settings.sftp_ingest_folder = None
result = scan_sftp_watch_folder()
assert result["status"] == "skipped"
def test_returns_error_when_connection_fails(self):
from app.tasks.watch_folder_tasks import scan_sftp_watch_folder
with (
patch("app.tasks.watch_folder_tasks.settings") as mock_settings,
patch("app.tasks.watch_folder_tasks._get_sftp_connection", return_value=(None, None)),
):
mock_settings.sftp_ingest_enabled = True
mock_settings.sftp_ingest_folder = "/upload"
mock_settings.sftp_ingest_delete_after_process = False
result = scan_sftp_watch_folder()
assert result["status"] == "error"
def test_successful_scan(self):
from app.tasks.watch_folder_tasks import scan_sftp_watch_folder
mock_ssh = MagicMock()
mock_sftp = MagicMock()
with (
patch("app.tasks.watch_folder_tasks.settings") as mock_settings,
patch("app.tasks.watch_folder_tasks._get_sftp_connection", return_value=(mock_ssh, mock_sftp)),
patch("app.tasks.watch_folder_tasks._load_cache", return_value={}),
patch("app.tasks.watch_folder_tasks._save_cache"),
patch("app.tasks.watch_folder_tasks._scan_sftp_folder", return_value=3),
):
mock_settings.sftp_ingest_enabled = True
mock_settings.sftp_ingest_folder = "/upload"
mock_settings.sftp_ingest_delete_after_process = False
result = scan_sftp_watch_folder()
assert result["status"] == "ok"
assert result["files_enqueued"] == 3
@pytest.mark.unit
class TestScanAllWatchFolders:
"""Tests for the scan_all_watch_folders orchestrator task."""
def test_skips_when_lock_held(self):
from app.tasks.watch_folder_tasks import scan_all_watch_folders
with patch("app.tasks.watch_folder_tasks._acquire_lock", return_value=False):
result = scan_all_watch_folders()
assert result["status"] == "skipped"
def test_runs_all_scans_and_releases_lock(self):
from app.tasks.watch_folder_tasks import scan_all_watch_folders
with (
patch("app.tasks.watch_folder_tasks._acquire_lock", return_value=True),
patch("app.tasks.watch_folder_tasks._release_lock") as mock_release,
patch("app.tasks.watch_folder_tasks.scan_local_watch_folders", return_value={"status": "ok"}),
patch("app.tasks.watch_folder_tasks.scan_ftp_watch_folder", return_value={"status": "skipped"}),
patch("app.tasks.watch_folder_tasks.scan_sftp_watch_folder", return_value={"status": "skipped"}),
):
result = scan_all_watch_folders()
assert result["status"] == "ok"
assert "results" in result
mock_release.assert_called_once()
def test_lock_released_even_on_exception(self):
"""Lock must be released even if a sub-scan raises an exception."""
from app.tasks.watch_folder_tasks import scan_all_watch_folders
with (
patch("app.tasks.watch_folder_tasks._acquire_lock", return_value=True),
patch("app.tasks.watch_folder_tasks._release_lock") as mock_release,
patch("app.tasks.watch_folder_tasks.scan_local_watch_folders", side_effect=RuntimeError("boom")),
):
with pytest.raises(RuntimeError):
scan_all_watch_folders()
mock_release.assert_called_once()
@pytest.mark.unit
class TestConnectFtp:
"""Tests for the _connect_ftp helper."""
def test_returns_none_when_settings_incomplete(self):
from app.tasks.watch_folder_tasks import _connect_ftp
with patch("app.tasks.watch_folder_tasks.settings") as mock_settings:
mock_settings.ftp_host = None
mock_settings.ftp_username = "user"
mock_settings.ftp_password = "pass" # noqa: S105
result = _connect_ftp()
assert result is None
def test_returns_none_when_connection_fails(self):
from app.tasks.watch_folder_tasks import _connect_ftp
with (
patch("app.tasks.watch_folder_tasks.settings") as mock_settings,
patch("ftplib.FTP_TLS") as mock_ftps_cls,
patch("ftplib.FTP") as mock_ftp_cls,
):
mock_settings.ftp_host = "ftp.example.com"
mock_settings.ftp_port = 21
mock_settings.ftp_username = "user"
mock_settings.ftp_password = "pass" # noqa: S105
mock_settings.ftp_use_tls = False
mock_settings.ftp_allow_plaintext = True
mock_ftp_cls.return_value.connect.side_effect = ConnectionRefusedError("refused")
result = _connect_ftp()
assert result is None
@pytest.mark.unit
class TestScanFtpFolder:
"""Tests for the _scan_ftp_folder helper."""
def test_cwd_failure_returns_zero(self):
import ftplib # noqa: S402
from app.tasks.watch_folder_tasks import _scan_ftp_folder
mock_ftp = MagicMock()
mock_ftp.cwd.side_effect = ftplib.error_perm("550 no such directory") # noqa: S321
count = _scan_ftp_folder(mock_ftp, "/missing", {}, False)
assert count == 0
def test_skips_disallowed_files(self):
from app.tasks.watch_folder_tasks import _scan_ftp_folder
mock_ftp = MagicMock()
mock_ftp.cwd.return_value = None
mock_ftp.nlst.return_value = ["photo.exe", "virus.bat"]
count = _scan_ftp_folder(mock_ftp, "/inbox", {}, False)
assert count == 0
def test_downloads_new_allowed_file(self, tmp_path):
from app.tasks.watch_folder_tasks import _scan_ftp_folder
mock_ftp = MagicMock()
mock_ftp.cwd.return_value = None
mock_ftp.nlst.return_value = ["invoice.pdf"]
cache: dict = {}
with (
patch("app.tasks.watch_folder_tasks.settings") as mock_settings,
patch("app.tasks.watch_folder_tasks.process_document"),
):
mock_settings.workdir = str(tmp_path)
# Simulate retrbinary writing bytes
def fake_retrbinary(cmd, callback):
callback(b"%PDF-1.4")
mock_ftp.retrbinary.side_effect = fake_retrbinary
count = _scan_ftp_folder(mock_ftp, "/inbox", cache, False)
assert count == 1
assert "ftp:/inbox/invoice.pdf" in cache
def test_already_cached_file_is_skipped(self, tmp_path):
from app.tasks.watch_folder_tasks import _scan_ftp_folder
mock_ftp = MagicMock()
mock_ftp.cwd.return_value = None
mock_ftp.nlst.return_value = ["invoice.pdf"]
cache = {"ftp:/inbox/invoice.pdf": datetime.now(timezone.utc).isoformat()}
count = _scan_ftp_folder(mock_ftp, "/inbox", cache, False)
assert count == 0
@pytest.mark.unit
class TestDropboxWatchFolderTask:
"""Tests for the scan_dropbox_watch_folder Celery task."""
def test_returns_skipped_when_disabled(self):
from app.tasks.watch_folder_tasks import scan_dropbox_watch_folder
with patch("app.tasks.watch_folder_tasks.settings") as mock_settings:
mock_settings.dropbox_ingest_enabled = False
result = scan_dropbox_watch_folder()
assert result["status"] == "skipped"
def test_returns_skipped_when_no_folder_configured(self):
from app.tasks.watch_folder_tasks import scan_dropbox_watch_folder
with patch("app.tasks.watch_folder_tasks.settings") as mock_settings:
mock_settings.dropbox_ingest_enabled = True
mock_settings.dropbox_ingest_folder = None
result = scan_dropbox_watch_folder()
assert result["status"] == "skipped"
def test_successful_scan(self):
from app.tasks.watch_folder_tasks import scan_dropbox_watch_folder
with (
patch("app.tasks.watch_folder_tasks.settings") as mock_settings,
patch("app.tasks.watch_folder_tasks._load_cache", return_value={}),
patch("app.tasks.watch_folder_tasks._save_cache"),
patch("app.tasks.watch_folder_tasks._scan_dropbox_folder", return_value=1),
):
mock_settings.dropbox_ingest_enabled = True
mock_settings.dropbox_ingest_folder = "/Inbox"
mock_settings.dropbox_ingest_delete_after_process = False
result = scan_dropbox_watch_folder()
assert result["status"] == "ok"
assert result["files_enqueued"] == 1
@pytest.mark.unit
class TestGoogleDriveWatchFolderTask:
"""Tests for the scan_google_drive_watch_folder Celery task."""
def test_returns_skipped_when_disabled(self):
from app.tasks.watch_folder_tasks import scan_google_drive_watch_folder
with patch("app.tasks.watch_folder_tasks.settings") as mock_settings:
mock_settings.google_drive_ingest_enabled = False
result = scan_google_drive_watch_folder()
assert result["status"] == "skipped"
def test_returns_skipped_when_no_folder_id(self):
from app.tasks.watch_folder_tasks import scan_google_drive_watch_folder
with patch("app.tasks.watch_folder_tasks.settings") as mock_settings:
mock_settings.google_drive_ingest_enabled = True
mock_settings.google_drive_ingest_folder_id = None
result = scan_google_drive_watch_folder()
assert result["status"] == "skipped"
def test_successful_scan(self):
from app.tasks.watch_folder_tasks import scan_google_drive_watch_folder
with (
patch("app.tasks.watch_folder_tasks.settings") as mock_settings,
patch("app.tasks.watch_folder_tasks._load_cache", return_value={}),
patch("app.tasks.watch_folder_tasks._save_cache"),
patch("app.tasks.watch_folder_tasks._scan_google_drive_folder", return_value=2),
):
mock_settings.google_drive_ingest_enabled = True
mock_settings.google_drive_ingest_folder_id = "abc123"
mock_settings.google_drive_ingest_delete_after_process = False
result = scan_google_drive_watch_folder()
assert result["status"] == "ok"
assert result["files_enqueued"] == 2
@pytest.mark.unit
class TestOnedriveWatchFolderTask:
"""Tests for the scan_onedrive_watch_folder Celery task."""
def test_returns_skipped_when_disabled(self):
from app.tasks.watch_folder_tasks import scan_onedrive_watch_folder
with patch("app.tasks.watch_folder_tasks.settings") as mock_settings:
mock_settings.onedrive_ingest_enabled = False
result = scan_onedrive_watch_folder()
assert result["status"] == "skipped"
def test_returns_skipped_when_no_folder_path(self):
from app.tasks.watch_folder_tasks import scan_onedrive_watch_folder
with patch("app.tasks.watch_folder_tasks.settings") as mock_settings:
mock_settings.onedrive_ingest_enabled = True
mock_settings.onedrive_ingest_folder_path = None
result = scan_onedrive_watch_folder()
assert result["status"] == "skipped"
def test_successful_scan(self):
from app.tasks.watch_folder_tasks import scan_onedrive_watch_folder
with (
patch("app.tasks.watch_folder_tasks.settings") as mock_settings,
patch("app.tasks.watch_folder_tasks._load_cache", return_value={}),
patch("app.tasks.watch_folder_tasks._save_cache"),
patch("app.tasks.watch_folder_tasks._scan_onedrive_folder", return_value=3),
):
mock_settings.onedrive_ingest_enabled = True
mock_settings.onedrive_ingest_folder_path = "/Inbox/Scanner"
mock_settings.onedrive_ingest_delete_after_process = False
result = scan_onedrive_watch_folder()
assert result["status"] == "ok"
assert result["files_enqueued"] == 3
@pytest.mark.unit
class TestNextcloudWatchFolderTask:
"""Tests for the scan_nextcloud_watch_folder Celery task."""
def test_returns_skipped_when_disabled(self):
from app.tasks.watch_folder_tasks import scan_nextcloud_watch_folder
with patch("app.tasks.watch_folder_tasks.settings") as mock_settings:
mock_settings.nextcloud_ingest_enabled = False
result = scan_nextcloud_watch_folder()
assert result["status"] == "skipped"
def test_returns_skipped_when_no_folder(self):
from app.tasks.watch_folder_tasks import scan_nextcloud_watch_folder
with patch("app.tasks.watch_folder_tasks.settings") as mock_settings:
mock_settings.nextcloud_ingest_enabled = True
mock_settings.nextcloud_ingest_folder = None
result = scan_nextcloud_watch_folder()
assert result["status"] == "skipped"
def test_successful_scan(self):
from app.tasks.watch_folder_tasks import scan_nextcloud_watch_folder
with (
patch("app.tasks.watch_folder_tasks.settings") as mock_settings,
patch("app.tasks.watch_folder_tasks._load_cache", return_value={}),
patch("app.tasks.watch_folder_tasks._save_cache"),
patch("app.tasks.watch_folder_tasks._scan_nextcloud_folder", return_value=1),
):
mock_settings.nextcloud_ingest_enabled = True
mock_settings.nextcloud_ingest_folder = "/Scans/Inbox"
mock_settings.nextcloud_ingest_delete_after_process = False
result = scan_nextcloud_watch_folder()
assert result["status"] == "ok"
assert result["files_enqueued"] == 1
@pytest.mark.unit
class TestS3WatchFolderTask:
"""Tests for the scan_s3_watch_folder Celery task."""
def test_returns_skipped_when_disabled(self):
from app.tasks.watch_folder_tasks import scan_s3_watch_folder
with patch("app.tasks.watch_folder_tasks.settings") as mock_settings:
mock_settings.s3_ingest_enabled = False
result = scan_s3_watch_folder()
assert result["status"] == "skipped"
def test_returns_skipped_when_no_prefix(self):
from app.tasks.watch_folder_tasks import scan_s3_watch_folder
with patch("app.tasks.watch_folder_tasks.settings") as mock_settings:
mock_settings.s3_ingest_enabled = True
mock_settings.s3_ingest_prefix = None
result = scan_s3_watch_folder()
assert result["status"] == "skipped"
def test_successful_scan(self):
from app.tasks.watch_folder_tasks import scan_s3_watch_folder
with (
patch("app.tasks.watch_folder_tasks.settings") as mock_settings,
patch("app.tasks.watch_folder_tasks._load_cache", return_value={}),
patch("app.tasks.watch_folder_tasks._save_cache"),
patch("app.tasks.watch_folder_tasks._scan_s3_prefix", return_value=4),
):
mock_settings.s3_ingest_enabled = True
mock_settings.s3_ingest_prefix = "inbox/scanner/"
mock_settings.s3_ingest_delete_after_process = False
result = scan_s3_watch_folder()
assert result["status"] == "ok"
assert result["files_enqueued"] == 4
@pytest.mark.unit
class TestWebdavWatchFolderTask:
"""Tests for the scan_webdav_watch_folder Celery task."""
def test_returns_skipped_when_disabled(self):
from app.tasks.watch_folder_tasks import scan_webdav_watch_folder
with patch("app.tasks.watch_folder_tasks.settings") as mock_settings:
mock_settings.webdav_ingest_enabled = False
result = scan_webdav_watch_folder()
assert result["status"] == "skipped"
def test_returns_skipped_when_no_folder(self):
from app.tasks.watch_folder_tasks import scan_webdav_watch_folder
with patch("app.tasks.watch_folder_tasks.settings") as mock_settings:
mock_settings.webdav_ingest_enabled = True
mock_settings.webdav_ingest_folder = None
result = scan_webdav_watch_folder()
assert result["status"] == "skipped"
def test_successful_scan(self):
from app.tasks.watch_folder_tasks import scan_webdav_watch_folder
with (
patch("app.tasks.watch_folder_tasks.settings") as mock_settings,
patch("app.tasks.watch_folder_tasks._load_cache", return_value={}),
patch("app.tasks.watch_folder_tasks._save_cache"),
patch("app.tasks.watch_folder_tasks._scan_webdav_folder", return_value=2),
):
mock_settings.webdav_ingest_enabled = True
mock_settings.webdav_ingest_folder = "/remote.php/webdav/Inbox"
mock_settings.webdav_ingest_delete_after_process = False
result = scan_webdav_watch_folder()
assert result["status"] == "ok"
assert result["files_enqueued"] == 2
@pytest.mark.unit
class TestScanAllWatchFoldersCloud:
"""Tests for the extended scan_all_watch_folders with cloud providers."""
def test_all_cloud_providers_called(self):
"""scan_all_watch_folders should call all provider-specific tasks."""
from app.tasks.watch_folder_tasks import scan_all_watch_folders
provider_tasks = [
"scan_local_watch_folders",
"scan_ftp_watch_folder",
"scan_sftp_watch_folder",
"scan_dropbox_watch_folder",
"scan_google_drive_watch_folder",
"scan_onedrive_watch_folder",
"scan_nextcloud_watch_folder",
"scan_s3_watch_folder",
"scan_webdav_watch_folder",
]
with (
patch("app.tasks.watch_folder_tasks._acquire_lock", return_value=True),
patch("app.tasks.watch_folder_tasks._release_lock"),
):
mocks = {}
patches = []
for name in provider_tasks:
m = MagicMock(return_value={"status": "skipped"})
p = patch(f"app.tasks.watch_folder_tasks.{name}", m)
patches.append(p)
mocks[name] = m
# Apply all patches
for p in patches:
p.start()
try:
result = scan_all_watch_folders()
finally:
for p in patches:
p.stop()
assert result["status"] == "ok"
for name in provider_tasks:
mocks[name].assert_called_once()