From 9a9efae19e9a8e6e4e43ced3426aa0b6a873597a Mon Sep 17 00:00:00 2001 From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com> Date: Sat, 7 Mar 2026 21:22:33 +0000 Subject: [PATCH] feat(watch-folders): add cloud provider watch folders (Dropbox, Drive, OneDrive, Nextcloud, S3, WebDAV) Co-authored-by: christianlouis <361235+christianlouis@users.noreply.github.com> --- .env.demo | 32 +- app/celery_worker.py | 10 +- app/config.py | 114 ++- app/tasks/watch_folder_tasks.py | 787 +++++++++++++++++- .../config_validator/settings_display.py | 18 + app/utils/settings_service.py | 150 ++++ docs/ConfigurationGuide.md | 60 ++ requirements.txt | 3 + tests/test_watch_folder_tasks.py | 381 +++++++-- 9 files changed, 1487 insertions(+), 68 deletions(-) diff --git a/.env.demo b/.env.demo index 5a486aa4..c8eee1cf 100644 --- a/.env.demo +++ b/.env.demo @@ -207,7 +207,7 @@ EMAIL_SENDER=DocuElevate System EMAIL_DEFAULT_RECIPIENT=recipient@example.com # **Watch Folder Ingestion** -# DocuElevate can automatically monitor directories (local filesystem, FTP, SFTP) for new files. +# 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. @@ -225,6 +225,36 @@ 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*. diff --git a/app/celery_worker.py b/app/celery_worker.py index 934ef132..6e290d39 100644 --- a/app/celery_worker.py +++ b/app/celery_worker.py @@ -17,7 +17,6 @@ from app.tasks.extract_metadata_with_gpt import extract_metadata_with_gpt # noq from app.tasks.finalize_document_storage import finalize_document_storage # noqa: F401 from app.tasks.imap_tasks import pull_all_inboxes # noqa: F401 from app.tasks.monitor_stalled_steps import monitor_stalled_steps # noqa: F401 -from app.tasks.watch_folder_tasks import scan_all_watch_folders # noqa: F401 # **Ensure all tasks are imported before Celery starts** from app.tasks.process_document import process_document # noqa: F401 @@ -40,6 +39,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 @@ -98,7 +98,7 @@ 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, and SFTP ingest folders. + # 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": ( { @@ -110,6 +110,12 @@ celery.conf.beat_schedule = { 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 ), diff --git a/app/config.py b/app/config.py index 3680f633..3baa392b 100644 --- a/app/config.py +++ b/app/config.py @@ -208,9 +208,7 @@ class Settings(BaseSettings): ) watch_folder_poll_interval: int = Field( default=1, - description=( - "Poll interval in minutes for local watch folder scanning. Default: 1 minute." - ), + description=("Poll interval in minutes for local watch folder scanning. Default: 1 minute."), ) watch_folder_delete_after_process: bool = Field( default=False, @@ -267,6 +265,116 @@ class Settings(BaseSettings): ), ) + # --------------------------------------------------------------------------- + # Cloud Provider Watch Folders + # --------------------------------------------------------------------------- + # Each cloud provider has three settings: + # _ingest_enabled — enable the watch-folder for this provider + # _ingest_folder — the remote path / folder ID to poll + # _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 diff --git a/app/tasks/watch_folder_tasks.py b/app/tasks/watch_folder_tasks.py index a699484c..19f8d373 100644 --- a/app/tasks/watch_folder_tasks.py +++ b/app/tasks/watch_folder_tasks.py @@ -15,7 +15,6 @@ import ftplib # nosec B402 - FTP usage is intentional for legacy server support import json import logging import os -import tempfile from datetime import datetime, timedelta, timezone import redis @@ -24,7 +23,7 @@ from celery import shared_task from app.config import settings from app.tasks.convert_to_pdf import convert_to_pdf from app.tasks.process_document import process_document -from app.utils.allowed_types import ALLOWED_EXTENSIONS, ALLOWED_MIME_TYPES +from app.utils.allowed_types import ALLOWED_EXTENSIONS logger = logging.getLogger(__name__) @@ -361,8 +360,8 @@ def _get_sftp_connection(): logger.error("SFTP ingest: connection failed: %s", exc) try: ssh.close() - except Exception: - pass + except Exception as close_exc: + logger.debug("SFTP ingest: error closing SSH after failed connection: %s", close_exc) return None, None @@ -424,6 +423,610 @@ def _scan_sftp_folder(sftp, remote_folder: str, cache: dict[str, str], delete_af return count +# --------------------------------------------------------------------------- +# Dropbox watch folder scanning +# --------------------------------------------------------------------------- + +# Additional cache files for cloud providers +DROPBOX_INGEST_CACHE_FILE = os.path.join(settings.workdir, "dropbox_ingest_processed.json") +GDRIVE_INGEST_CACHE_FILE = os.path.join(settings.workdir, "gdrive_ingest_processed.json") +ONEDRIVE_INGEST_CACHE_FILE = os.path.join(settings.workdir, "onedrive_ingest_processed.json") +NEXTCLOUD_INGEST_CACHE_FILE = os.path.join(settings.workdir, "nextcloud_ingest_processed.json") +S3_INGEST_CACHE_FILE = os.path.join(settings.workdir, "s3_ingest_processed.json") +WEBDAV_INGEST_CACHE_FILE = os.path.join(settings.workdir, "webdav_ingest_processed.json") + + +def _scan_dropbox_folder(folder_path: str, cache: dict[str, str], delete_after: bool) -> int: + """ + List *folder_path* in Dropbox and download new allowed files to workdir. + Returns the number of files newly enqueued. + """ + try: + from app.tasks.upload_to_dropbox import get_dropbox_client + except ImportError as exc: + logger.error("Dropbox ingest: dropbox SDK not installed: %s", exc) + return 0 + + try: + dbx = get_dropbox_client() + except Exception as exc: + logger.error("Dropbox ingest: authentication failed: %s", exc) + return 0 + + count = 0 + try: + result = dbx.files_list_folder(folder_path) + entries = result.entries + while result.has_more: + result = dbx.files_list_folder_continue(result.cursor) + entries.extend(result.entries) + except Exception as exc: + logger.error("Dropbox ingest: cannot list folder %s: %s", folder_path, exc) + return 0 + + for entry in entries: + # Only process files, not sub-folders + import dropbox as dropbox_module + + if not isinstance(entry, dropbox_module.files.FileMetadata): + continue + + filename = entry.name + if not _is_allowed_file(filename): + logger.debug("Dropbox ingest: skipping %s (unsupported type)", filename) + continue + + cache_key = f"dropbox:{entry.id}" + if cache_key in cache: + logger.debug("Dropbox ingest: already processed %s", filename) + continue + + dest_path = os.path.join(settings.workdir, f"dropbox_{filename}") + if os.path.exists(dest_path): + base, ext2 = os.path.splitext(f"dropbox_{filename}") + dest_path = os.path.join(settings.workdir, f"{base}_{int(datetime.now().timestamp())}{ext2}") + + try: + _meta, response = dbx.files_download(entry.path_lower) + with open(dest_path, "wb") as f: + f.write(response.content) + logger.info("Dropbox ingest: downloaded %s to %s", filename, dest_path) + except Exception as exc: + logger.error("Dropbox ingest: failed to download %s: %s", filename, exc) + if os.path.exists(dest_path): + os.remove(dest_path) + continue + + _enqueue_file(dest_path, filename=filename) + _mark_processed(cache, cache_key) + count += 1 + + if delete_after: + try: + dbx.files_delete_v2(entry.path_lower) + logger.info("Dropbox ingest: deleted %s after ingestion", entry.path_lower) + except Exception as exc: + logger.warning("Dropbox ingest: could not delete %s: %s", entry.path_lower, exc) + + return count + + +# --------------------------------------------------------------------------- +# Google Drive watch folder scanning +# --------------------------------------------------------------------------- + + +def _scan_google_drive_folder(folder_id: str, cache: dict[str, str], delete_after: bool) -> int: + """ + List files in *folder_id* on Google Drive and download new allowed files to workdir. + Returns the number of files newly enqueued. + """ + try: + from app.tasks.upload_to_google_drive import get_google_drive_service + except ImportError as exc: + logger.error("Google Drive ingest: google-api SDK not installed: %s", exc) + return 0 + + service = get_google_drive_service() + if service is None: + logger.error("Google Drive ingest: could not authenticate.") + return 0 + + count = 0 + query = f"'{folder_id}' in parents and trashed = false and mimeType != 'application/vnd.google-apps.folder'" + page_token = None + + while True: + try: + params: dict = { + "q": query, + "fields": "nextPageToken, files(id, name, mimeType)", + "pageSize": 100, + } + if page_token: + params["pageToken"] = page_token + response = service.files().list(**params).execute() + except Exception as exc: + logger.error("Google Drive ingest: listing folder %s failed: %s", folder_id, exc) + break + + for file_meta in response.get("files", []): + file_id_gd = file_meta["id"] + filename = file_meta["name"] + + if not _is_allowed_file(filename): + logger.debug("Google Drive ingest: skipping %s (unsupported type)", filename) + continue + + cache_key = f"gdrive:{file_id_gd}" + if cache_key in cache: + logger.debug("Google Drive ingest: already processed %s", filename) + continue + + dest_path = os.path.join(settings.workdir, f"gdrive_{filename}") + if os.path.exists(dest_path): + base, ext2 = os.path.splitext(f"gdrive_{filename}") + dest_path = os.path.join(settings.workdir, f"{base}_{int(datetime.now().timestamp())}{ext2}") + + try: + import io + + from googleapiclient.http import MediaIoBaseDownload + + request = service.files().get_media(fileId=file_id_gd) + buf = io.BytesIO() + downloader = MediaIoBaseDownload(buf, request) + done = False + while not done: + _, done = downloader.next_chunk() + with open(dest_path, "wb") as f: + f.write(buf.getvalue()) + logger.info("Google Drive ingest: downloaded %s to %s", filename, dest_path) + except Exception as exc: + logger.error("Google Drive ingest: failed to download %s: %s", filename, exc) + if os.path.exists(dest_path): + os.remove(dest_path) + continue + + _enqueue_file(dest_path, filename=filename) + _mark_processed(cache, cache_key) + count += 1 + + if delete_after: + try: + service.files().delete(fileId=file_id_gd).execute() + logger.info("Google Drive ingest: deleted %s after ingestion", filename) + except Exception as exc: + logger.warning("Google Drive ingest: could not delete %s: %s", filename, exc) + + page_token = response.get("nextPageToken") + if not page_token: + break + + return count + + +# --------------------------------------------------------------------------- +# OneDrive watch folder scanning +# --------------------------------------------------------------------------- + + +def _scan_onedrive_folder(folder_path: str, cache: dict[str, str], delete_after: bool) -> int: + """ + List files in *folder_path* on OneDrive (Microsoft Graph) and download new allowed + files to workdir. Returns the number of files newly enqueued. + """ + import requests as req_lib + + try: + from app.tasks.upload_to_onedrive import get_onedrive_token + except ImportError as exc: + logger.error("OneDrive ingest: msal not installed: %s", exc) + return 0 + + try: + token = get_onedrive_token() + except Exception as exc: + logger.error("OneDrive ingest: authentication failed: %s", exc) + return 0 + + headers = {"Authorization": f"Bearer {token}"} + + # URL-encode the path and construct the Graph API endpoint + import urllib.parse + + encoded_path = urllib.parse.quote(folder_path.lstrip("/")) + list_url = f"https://graph.microsoft.com/v1.0/me/drive/root:/{encoded_path}:/children" + + count = 0 + while list_url: + try: + resp = req_lib.get(list_url, headers=headers, timeout=getattr(settings, "http_request_timeout", 120)) + resp.raise_for_status() + data = resp.json() + except Exception as exc: + logger.error("OneDrive ingest: listing folder %s failed: %s", folder_path, exc) + break + + for item in data.get("value", []): + # Skip folders + if "folder" in item: + continue + + filename = item["name"] + item_id = item["id"] + + if not _is_allowed_file(filename): + logger.debug("OneDrive ingest: skipping %s (unsupported type)", filename) + continue + + cache_key = f"onedrive:{item_id}" + if cache_key in cache: + logger.debug("OneDrive ingest: already processed %s", filename) + continue + + # Get download URL + download_url = item.get("@microsoft.graph.downloadUrl") + if not download_url: + logger.warning("OneDrive ingest: no download URL for %s — skipping.", filename) + continue + + dest_path = os.path.join(settings.workdir, f"onedrive_{filename}") + if os.path.exists(dest_path): + base, ext2 = os.path.splitext(f"onedrive_{filename}") + dest_path = os.path.join(settings.workdir, f"{base}_{int(datetime.now().timestamp())}{ext2}") + + try: + dl_resp = req_lib.get( + download_url, headers=headers, timeout=getattr(settings, "http_request_timeout", 120) + ) + dl_resp.raise_for_status() + with open(dest_path, "wb") as f: + f.write(dl_resp.content) + logger.info("OneDrive ingest: downloaded %s to %s", filename, dest_path) + except Exception as exc: + logger.error("OneDrive ingest: failed to download %s: %s", filename, exc) + if os.path.exists(dest_path): + os.remove(dest_path) + continue + + _enqueue_file(dest_path, filename=filename) + _mark_processed(cache, cache_key) + count += 1 + + if delete_after: + try: + del_resp = req_lib.delete( + f"https://graph.microsoft.com/v1.0/me/drive/items/{item_id}", + headers=headers, + timeout=getattr(settings, "http_request_timeout", 120), + ) + del_resp.raise_for_status() + logger.info("OneDrive ingest: deleted %s after ingestion", filename) + except Exception as exc: + logger.warning("OneDrive ingest: could not delete %s: %s", filename, exc) + + list_url = data.get("@odata.nextLink") + + return count + + +# --------------------------------------------------------------------------- +# Nextcloud watch folder scanning (WebDAV) +# --------------------------------------------------------------------------- + + +def _scan_nextcloud_folder(folder_path: str, cache: dict[str, str], delete_after: bool) -> int: + """ + List files in *folder_path* on Nextcloud (via WebDAV PROPFIND) and download new + allowed files to workdir. Returns the number of files newly enqueued. + """ + import defusedxml.ElementTree as ET + import requests as req_lib + from requests.auth import HTTPBasicAuth + + nc_url: str | None = getattr(settings, "nextcloud_upload_url", None) + nc_user: str | None = getattr(settings, "nextcloud_username", None) + nc_pass: str | None = getattr(settings, "nextcloud_password", None) + + if not (nc_url and nc_user and nc_pass): + logger.warning("Nextcloud ingest: connection settings incomplete — skipping.") + return 0 + + auth = HTTPBasicAuth(nc_user, nc_pass) + timeout = getattr(settings, "http_request_timeout", 120) + + # Build the WebDAV PROPFIND URL + base = nc_url.rstrip("/") + folder = folder_path.strip("/") + propfind_url = f"{base}/{folder}/" if folder else f"{base}/" + + try: + resp = req_lib.request( + "PROPFIND", + propfind_url, + auth=auth, + headers={"Depth": "1", "Content-Type": "application/xml"}, + timeout=timeout, + ) + resp.raise_for_status() + except Exception as exc: + logger.error("Nextcloud ingest: PROPFIND on %s failed: %s", propfind_url, exc) + return 0 + + count = 0 + # Parse WebDAV multistatus response using defusedxml (safe against XML bomb attacks) + try: + root = ET.fromstring(resp.text) # noqa: S314 — defusedxml is safe + except Exception as exc: + logger.error("Nextcloud ingest: failed to parse PROPFIND response: %s", exc) + return 0 + + ns = {"d": "DAV:"} + for response_el in root.findall("d:response", ns): + href_el = response_el.find("d:href", ns) + if href_el is None or href_el.text is None: + continue + + href = href_el.text + # Skip the folder itself + if href.rstrip("/").endswith(folder.rstrip("/")): + continue + + filename = href.rstrip("/").split("/")[-1] + import urllib.parse + + filename = urllib.parse.unquote(filename) + + if not _is_allowed_file(filename): + logger.debug("Nextcloud ingest: skipping %s (unsupported type)", filename) + continue + + # Use the href as cache key (stable across runs) + cache_key = f"nextcloud:{href}" + if cache_key in cache: + logger.debug("Nextcloud ingest: already processed %s", filename) + continue + + # Build absolute download URL + if href.startswith("http"): + file_url = href + else: + from urllib.parse import urlparse + + parsed = urlparse(nc_url) + file_url = f"{parsed.scheme}://{parsed.netloc}{href}" + + dest_path = os.path.join(settings.workdir, f"nc_{filename}") + if os.path.exists(dest_path): + base_name, ext2 = os.path.splitext(f"nc_{filename}") + dest_path = os.path.join(settings.workdir, f"{base_name}_{int(datetime.now().timestamp())}{ext2}") + + try: + dl = req_lib.get(file_url, auth=auth, timeout=timeout) + dl.raise_for_status() + with open(dest_path, "wb") as f: + f.write(dl.content) + logger.info("Nextcloud ingest: downloaded %s to %s", filename, dest_path) + except Exception as exc: + logger.error("Nextcloud ingest: failed to download %s: %s", filename, exc) + if os.path.exists(dest_path): + os.remove(dest_path) + continue + + _enqueue_file(dest_path, filename=filename) + _mark_processed(cache, cache_key) + count += 1 + + if delete_after: + try: + del_resp = req_lib.request("DELETE", file_url, auth=auth, timeout=timeout) + del_resp.raise_for_status() + logger.info("Nextcloud ingest: deleted %s after ingestion", filename) + except Exception as exc: + logger.warning("Nextcloud ingest: could not delete %s: %s", filename, exc) + + return count + + +# --------------------------------------------------------------------------- +# S3 watch folder scanning +# --------------------------------------------------------------------------- + + +def _scan_s3_prefix(prefix: str, cache: dict[str, str], delete_after: bool) -> int: + """ + List objects under *prefix* in the configured S3 bucket and download new allowed + files to workdir. Returns the number of files newly enqueued. + """ + try: + import boto3 + from botocore.exceptions import ClientError + except ImportError as exc: + logger.error("S3 ingest: boto3 not installed: %s", exc) + return 0 + + bucket = getattr(settings, "s3_bucket_name", None) + if not bucket: + logger.warning("S3 ingest: S3_BUCKET_NAME not set — skipping.") + return 0 + + try: + s3 = boto3.client( + "s3", + region_name=getattr(settings, "aws_region", "us-east-1"), + aws_access_key_id=getattr(settings, "aws_access_key_id", None), + aws_secret_access_key=getattr(settings, "aws_secret_access_key", None), + ) + except Exception as exc: + logger.error("S3 ingest: failed to create S3 client: %s", exc) + return 0 + + count = 0 + paginator = s3.get_paginator("list_objects_v2") + + try: + pages = paginator.paginate(Bucket=bucket, Prefix=prefix) + except Exception as exc: + logger.error("S3 ingest: failed to list objects in %s/%s: %s", bucket, prefix, exc) + return 0 + + for page in pages: + for obj in page.get("Contents", []): + key = obj["Key"] + filename = key.split("/")[-1] + + # Skip zero-byte "folder marker" objects and unsupported types + if not filename or not _is_allowed_file(filename): + logger.debug("S3 ingest: skipping %s (unsupported or empty)", key) + continue + + cache_key = f"s3:{bucket}/{key}" + if cache_key in cache: + logger.debug("S3 ingest: already processed %s", key) + continue + + dest_path = os.path.join(settings.workdir, f"s3_{filename}") + if os.path.exists(dest_path): + base2, ext2 = os.path.splitext(f"s3_{filename}") + dest_path = os.path.join(settings.workdir, f"{base2}_{int(datetime.now().timestamp())}{ext2}") + + try: + s3.download_file(bucket, key, dest_path) + logger.info("S3 ingest: downloaded s3://%s/%s to %s", bucket, key, dest_path) + except ClientError as exc: + logger.error("S3 ingest: failed to download %s: %s", key, exc) + if os.path.exists(dest_path): + os.remove(dest_path) + continue + + _enqueue_file(dest_path, filename=filename) + _mark_processed(cache, cache_key) + count += 1 + + if delete_after: + try: + s3.delete_object(Bucket=bucket, Key=key) + logger.info("S3 ingest: deleted s3://%s/%s after ingestion", bucket, key) + except Exception as exc: + logger.warning("S3 ingest: could not delete s3://%s/%s: %s", bucket, key, exc) + + return count + + +# --------------------------------------------------------------------------- +# WebDAV watch folder scanning +# --------------------------------------------------------------------------- + + +def _scan_webdav_folder(folder_path: str, cache: dict[str, str], delete_after: bool) -> int: + """ + List files in *folder_path* on a WebDAV server (PROPFIND) and download new allowed + files to workdir. Returns the number of files newly enqueued. + """ + import defusedxml.ElementTree as ET + import requests as req_lib + from requests.auth import HTTPBasicAuth + + webdav_url: str | None = getattr(settings, "webdav_url", None) + webdav_user: str | None = getattr(settings, "webdav_username", None) + webdav_pass: str | None = getattr(settings, "webdav_password", None) + verify_ssl: bool = getattr(settings, "webdav_verify_ssl", True) + timeout = getattr(settings, "http_request_timeout", 120) + + if not webdav_url: + logger.warning("WebDAV ingest: WEBDAV_URL not configured — skipping.") + return 0 + + from urllib.parse import unquote, urlparse + + base = webdav_url.rstrip("/") + folder = folder_path.strip("/") + propfind_url = f"{base}/{folder}/" if folder else f"{base}/" + + auth = HTTPBasicAuth(webdav_user, webdav_pass) if webdav_user else None + + try: + resp = req_lib.request( + "PROPFIND", + propfind_url, + auth=auth, + headers={"Depth": "1"}, + verify=verify_ssl, + timeout=timeout, + ) + resp.raise_for_status() + except Exception as exc: + logger.error("WebDAV ingest: PROPFIND on %s failed: %s", propfind_url, exc) + return 0 + + count = 0 + try: + root = ET.fromstring(resp.text) # noqa: S314 — defusedxml is safe + except Exception as exc: + logger.error("WebDAV ingest: failed to parse PROPFIND response: %s", exc) + return 0 + + ns = {"d": "DAV:"} + for response_el in root.findall("d:response", ns): + href_el = response_el.find("d:href", ns) + if href_el is None or href_el.text is None: + continue + + href = href_el.text + # Skip the folder itself and anything that looks like a directory + if href.endswith("/"): + continue + + filename = unquote(href.split("/")[-1]) + if not _is_allowed_file(filename): + logger.debug("WebDAV ingest: skipping %s (unsupported type)", filename) + continue + + cache_key = f"webdav:{href}" + if cache_key in cache: + logger.debug("WebDAV ingest: already processed %s", filename) + continue + + # Build absolute URL + if href.startswith("http"): + file_url = href + else: + parsed = urlparse(webdav_url) + file_url = f"{parsed.scheme}://{parsed.netloc}{href}" + + dest_path = os.path.join(settings.workdir, f"webdav_{filename}") + if os.path.exists(dest_path): + base2, ext2 = os.path.splitext(f"webdav_{filename}") + dest_path = os.path.join(settings.workdir, f"{base2}_{int(datetime.now().timestamp())}{ext2}") + + try: + dl = req_lib.get(file_url, auth=auth, verify=verify_ssl, timeout=timeout) + dl.raise_for_status() + with open(dest_path, "wb") as f: + f.write(dl.content) + logger.info("WebDAV ingest: downloaded %s to %s", filename, dest_path) + except Exception as exc: + logger.error("WebDAV ingest: failed to download %s: %s", filename, exc) + if os.path.exists(dest_path): + os.remove(dest_path) + continue + + _enqueue_file(dest_path, filename=filename) + _mark_processed(cache, cache_key) + count += 1 + + if delete_after: + try: + del_resp = req_lib.request("DELETE", file_url, auth=auth, verify=verify_ssl, timeout=timeout) + del_resp.raise_for_status() + logger.info("WebDAV ingest: deleted %s after ingestion", filename) + except Exception as exc: + logger.warning("WebDAV ingest: could not delete %s: %s", filename, exc) + + return count + + # --------------------------------------------------------------------------- # Main Celery tasks # --------------------------------------------------------------------------- @@ -491,8 +1094,8 @@ def scan_ftp_watch_folder() -> dict: finally: try: ftp.quit() - except Exception: - pass + except Exception as exc: + logger.debug("FTP ingest: error during FTP quit: %s", exc) _save_cache(FTP_INGEST_CACHE_FILE, cache) logger.info("FTP ingest: %d new file(s) enqueued from %s.", n, ingest_folder) @@ -529,27 +1132,177 @@ def scan_sftp_watch_folder() -> dict: finally: try: sftp.close() - except Exception: - pass + except Exception as exc: + logger.debug("SFTP ingest: error closing SFTP channel: %s", exc) try: ssh.close() - except Exception: - pass + except Exception as exc: + logger.debug("SFTP ingest: error closing SSH connection: %s", exc) _save_cache(SFTP_INGEST_CACHE_FILE, cache) logger.info("SFTP ingest: %d new file(s) enqueued from %s.", n, ingest_folder) return {"status": "ok", "files_enqueued": n, "folder": ingest_folder} +@shared_task +def scan_dropbox_watch_folder() -> dict: + """ + Celery task: scan the configured Dropbox ingest folder for new files. + + Uses the existing Dropbox OAuth credentials and DROPBOX_INGEST_FOLDER to poll + for new documents. + """ + if not getattr(settings, "dropbox_ingest_enabled", False): + return {"status": "skipped", "reason": "DROPBOX_INGEST_ENABLED is False"} + + ingest_folder: str | None = getattr(settings, "dropbox_ingest_folder", None) + if not ingest_folder: + logger.warning("Dropbox ingest enabled but DROPBOX_INGEST_FOLDER is not set — skipping.") + return {"status": "skipped", "reason": "DROPBOX_INGEST_FOLDER not configured"} + + delete_after: bool = getattr(settings, "dropbox_ingest_delete_after_process", False) + cache = _load_cache(DROPBOX_INGEST_CACHE_FILE) + n = _scan_dropbox_folder(ingest_folder, cache, delete_after) + _save_cache(DROPBOX_INGEST_CACHE_FILE, cache) + logger.info("Dropbox ingest: %d new file(s) enqueued from %s.", n, ingest_folder) + return {"status": "ok", "files_enqueued": n, "folder": ingest_folder} + + +@shared_task +def scan_google_drive_watch_folder() -> dict: + """ + Celery task: scan the configured Google Drive ingest folder for new files. + + Uses the existing Google Drive credentials and GOOGLE_DRIVE_INGEST_FOLDER_ID to + poll for new documents. + """ + if not getattr(settings, "google_drive_ingest_enabled", False): + return {"status": "skipped", "reason": "GOOGLE_DRIVE_INGEST_ENABLED is False"} + + folder_id: str | None = getattr(settings, "google_drive_ingest_folder_id", None) + if not folder_id: + logger.warning("Google Drive ingest enabled but GOOGLE_DRIVE_INGEST_FOLDER_ID is not set — skipping.") + return {"status": "skipped", "reason": "GOOGLE_DRIVE_INGEST_FOLDER_ID not configured"} + + delete_after: bool = getattr(settings, "google_drive_ingest_delete_after_process", False) + cache = _load_cache(GDRIVE_INGEST_CACHE_FILE) + n = _scan_google_drive_folder(folder_id, cache, delete_after) + _save_cache(GDRIVE_INGEST_CACHE_FILE, cache) + logger.info("Google Drive ingest: %d new file(s) enqueued from folder %s.", n, folder_id) + return {"status": "ok", "files_enqueued": n, "folder_id": folder_id} + + +@shared_task +def scan_onedrive_watch_folder() -> dict: + """ + Celery task: scan the configured OneDrive ingest folder for new files. + + Uses the existing OneDrive MSAL credentials and ONEDRIVE_INGEST_FOLDER_PATH to + poll for new documents. + """ + if not getattr(settings, "onedrive_ingest_enabled", False): + return {"status": "skipped", "reason": "ONEDRIVE_INGEST_ENABLED is False"} + + folder_path: str | None = getattr(settings, "onedrive_ingest_folder_path", None) + if not folder_path: + logger.warning("OneDrive ingest enabled but ONEDRIVE_INGEST_FOLDER_PATH is not set — skipping.") + return {"status": "skipped", "reason": "ONEDRIVE_INGEST_FOLDER_PATH not configured"} + + delete_after: bool = getattr(settings, "onedrive_ingest_delete_after_process", False) + cache = _load_cache(ONEDRIVE_INGEST_CACHE_FILE) + n = _scan_onedrive_folder(folder_path, cache, delete_after) + _save_cache(ONEDRIVE_INGEST_CACHE_FILE, cache) + logger.info("OneDrive ingest: %d new file(s) enqueued from %s.", n, folder_path) + return {"status": "ok", "files_enqueued": n, "folder": folder_path} + + +@shared_task +def scan_nextcloud_watch_folder() -> dict: + """ + Celery task: scan the configured Nextcloud ingest folder for new files. + + Uses the existing Nextcloud WebDAV credentials and NEXTCLOUD_INGEST_FOLDER to + poll for new documents. + """ + if not getattr(settings, "nextcloud_ingest_enabled", False): + return {"status": "skipped", "reason": "NEXTCLOUD_INGEST_ENABLED is False"} + + ingest_folder: str | None = getattr(settings, "nextcloud_ingest_folder", None) + if not ingest_folder: + logger.warning("Nextcloud ingest enabled but NEXTCLOUD_INGEST_FOLDER is not set — skipping.") + return {"status": "skipped", "reason": "NEXTCLOUD_INGEST_FOLDER not configured"} + + delete_after: bool = getattr(settings, "nextcloud_ingest_delete_after_process", False) + cache = _load_cache(NEXTCLOUD_INGEST_CACHE_FILE) + n = _scan_nextcloud_folder(ingest_folder, cache, delete_after) + _save_cache(NEXTCLOUD_INGEST_CACHE_FILE, cache) + logger.info("Nextcloud ingest: %d new file(s) enqueued from %s.", n, ingest_folder) + return {"status": "ok", "files_enqueued": n, "folder": ingest_folder} + + +@shared_task +def scan_s3_watch_folder() -> dict: + """ + Celery task: scan the configured S3 ingest prefix for new objects. + + Uses the existing S3/AWS credentials and S3_INGEST_PREFIX to poll for new + documents in the configured S3 bucket. + """ + if not getattr(settings, "s3_ingest_enabled", False): + return {"status": "skipped", "reason": "S3_INGEST_ENABLED is False"} + + ingest_prefix: str | None = getattr(settings, "s3_ingest_prefix", None) + if not ingest_prefix: + logger.warning("S3 ingest enabled but S3_INGEST_PREFIX is not set — skipping.") + return {"status": "skipped", "reason": "S3_INGEST_PREFIX not configured"} + + delete_after: bool = getattr(settings, "s3_ingest_delete_after_process", False) + cache = _load_cache(S3_INGEST_CACHE_FILE) + n = _scan_s3_prefix(ingest_prefix, cache, delete_after) + _save_cache(S3_INGEST_CACHE_FILE, cache) + logger.info("S3 ingest: %d new file(s) enqueued from prefix %s.", n, ingest_prefix) + return {"status": "ok", "files_enqueued": n, "prefix": ingest_prefix} + + +@shared_task +def scan_webdav_watch_folder() -> dict: + """ + Celery task: scan the configured WebDAV ingest folder for new files. + + Uses the existing WebDAV URL/credentials and WEBDAV_INGEST_FOLDER to poll for + new documents. + """ + if not getattr(settings, "webdav_ingest_enabled", False): + return {"status": "skipped", "reason": "WEBDAV_INGEST_ENABLED is False"} + + ingest_folder: str | None = getattr(settings, "webdav_ingest_folder", None) + if not ingest_folder: + logger.warning("WebDAV ingest enabled but WEBDAV_INGEST_FOLDER is not set — skipping.") + return {"status": "skipped", "reason": "WEBDAV_INGEST_FOLDER not configured"} + + delete_after: bool = getattr(settings, "webdav_ingest_delete_after_process", False) + cache = _load_cache(WEBDAV_INGEST_CACHE_FILE) + n = _scan_webdav_folder(ingest_folder, cache, delete_after) + _save_cache(WEBDAV_INGEST_CACHE_FILE, cache) + logger.info("WebDAV ingest: %d new file(s) enqueued from %s.", n, ingest_folder) + return {"status": "ok", "files_enqueued": n, "folder": ingest_folder} + + @shared_task def scan_all_watch_folders() -> dict: """ Main periodic Celery task that runs all watch-folder scans. - Acquires a Redis lock to prevent concurrent runs, then sequentially: - 1. Scans local filesystem watch folders - 2. Scans FTP ingest folder (if enabled) - 3. Scans SFTP ingest folder (if enabled) + Acquires a Redis lock to prevent concurrent runs, then sequentially scans: + 1. Local filesystem watch folders + 2. FTP ingest folder (if enabled) + 3. SFTP ingest folder (if enabled) + 4. Dropbox ingest folder (if enabled) + 5. Google Drive ingest folder (if enabled) + 6. OneDrive ingest folder (if enabled) + 7. Nextcloud ingest folder (if enabled) + 8. Amazon S3 ingest prefix (if enabled) + 9. WebDAV ingest folder (if enabled) """ if not _acquire_lock(WATCH_FOLDER_LOCK_KEY): logger.info("Watch folder scan already running — skipping this cycle.") @@ -560,6 +1313,12 @@ def scan_all_watch_folders() -> dict: results["local"] = scan_local_watch_folders() results["ftp"] = scan_ftp_watch_folder() results["sftp"] = scan_sftp_watch_folder() + results["dropbox"] = scan_dropbox_watch_folder() + results["google_drive"] = scan_google_drive_watch_folder() + results["onedrive"] = scan_onedrive_watch_folder() + results["nextcloud"] = scan_nextcloud_watch_folder() + results["s3"] = scan_s3_watch_folder() + results["webdav"] = scan_webdav_watch_folder() finally: _release_lock(WATCH_FOLDER_LOCK_KEY) diff --git a/app/utils/config_validator/settings_display.py b/app/utils/config_validator/settings_display.py index c8df7cbe..0064721c 100644 --- a/app/utils/config_validator/settings_display.py +++ b/app/utils/config_validator/settings_display.py @@ -116,6 +116,24 @@ def get_settings_for_display(show_values: bool = False) -> dict[str, list[dict[s "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", diff --git a/app/utils/settings_service.py b/app/utils/settings_service.py index f6646833..66707e72 100644 --- a/app/utils/settings_service.py +++ b/app/utils/settings_service.py @@ -1059,6 +1059,156 @@ SETTING_METADATA = { "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", diff --git a/docs/ConfigurationGuide.md b/docs/ConfigurationGuide.md index b4f63b15..8797c35b 100644 --- a/docs/ConfigurationGuide.md +++ b/docs/ConfigurationGuide.md @@ -207,6 +207,66 @@ SFTP_INGEST_DELETE_AFTER_PROCESS=false 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. diff --git a/requirements.txt b/requirements.txt index 3eb8e52d..b8016a6e 100644 --- a/requirements.txt +++ b/requirements.txt @@ -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 diff --git a/tests/test_watch_folder_tasks.py b/tests/test_watch_folder_tasks.py index d5e3e6e6..b08caec2 100644 --- a/tests/test_watch_folder_tasks.py +++ b/tests/test_watch_folder_tasks.py @@ -1,8 +1,6 @@ """Tests for app/tasks/watch_folder_tasks.py module.""" -import json import os -import tempfile from datetime import datetime, timedelta, timezone from unittest.mock import MagicMock, patch @@ -127,9 +125,10 @@ class TestScanLocalFolder: 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: + 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) @@ -175,9 +174,10 @@ class TestScanLocalFolder: 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: + 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) @@ -240,11 +240,12 @@ class TestScanLocalWatchFoldersTask: """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: + 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() @@ -278,8 +279,9 @@ class TestScanFtpWatchFolderTask: 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 + 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" @@ -291,12 +293,12 @@ class TestScanFtpWatchFolderTask: 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 + 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" @@ -331,8 +333,9 @@ class TestScanSftpWatchFolderTask: 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) + 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" @@ -345,12 +348,12 @@ class TestScanSftpWatchFolderTask: 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 + 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" @@ -375,14 +378,12 @@ class TestScanAllWatchFolders: 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"} + 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() @@ -394,10 +395,10 @@ class TestScanAllWatchFolders: """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 ( + 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() @@ -422,9 +423,11 @@ class TestConnectFtp: 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: + 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" @@ -442,12 +445,12 @@ class TestScanFtpFolder: """Tests for the _scan_ftp_folder helper.""" def test_cwd_failure_returns_zero(self): - import ftplib + 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") + 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 @@ -468,10 +471,12 @@ class TestScanFtpFolder: 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" + 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") @@ -492,3 +497,283 @@ class TestScanFtpFolder: 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()