Merge branch 'main' into refactor-filename-regex-constant-13933144971632372772

app/tasks/embed_metadata_into_pdf.py

@@ -216,6 +216,30 @@ def embed_metadata_into_pdf(self, local_file_path: str, extracted_text: str, met
                     except Exception as search_exc:
                         logger.warning(f"[{task_id}] Meilisearch indexing failed (non-fatal): {search_exc}")
 
+                    # Cache the detected language on the FileRecord and trigger
+                    # default-language translation when the document is in a
+                    # different language.
+                    detected_lang = metadata.get("language") if metadata else None
+                    if detected_lang and extracted_text:
+                        try:
+                            file_record.detected_language = detected_lang
+                            db.commit()
+
+                            from app.tasks.translate_to_default_language import translate_to_default_language
+
+                            translate_to_default_language.delay(
+                                file_id,
+                                extracted_text,
+                                detected_lang,
+                                owner_id=file_record.owner_id,
+                            )
+                            logger.info(
+                                f"[{task_id}] Queued default-language translation for file {file_id} "
+                                f"(detected: {detected_lang})"
+                            )
+                        except Exception as trans_exc:
+                            logger.warning(f"[{task_id}] Could not queue translation task (non-fatal): {trans_exc}")
+
         # Persist the metadata into a JSON file with the same base name.
         # Include file path references for traceability
         logger.info(f"[{task_id}] Persisting metadata to JSON")

app/tasks/finalize_document_storage.py

@@ -21,8 +21,9 @@ from app.tasks.send_to_all import (
 # Import database and logging utils from main
 from app.utils import log_task_progress
 
-# Import notification utility
+# Import notification utilities
 from app.utils.notification import notify_file_processed
+from app.utils.user_notification import notify_user_document_processed
 
 logger = logging.getLogger(__name__)
 
@@ -139,4 +140,15 @@ def finalize_document_storage(self, original_file: str, processed_file: str, met
     except Exception as e:
         logger.warning(f"[WARNING] Failed to send file processed notification: {e}")
 
+    # 6. Send per-user notification
+    if owner_id:
+        try:
+            notify_user_document_processed(
+                owner_id=owner_id,
+                filename=os.path.basename(processed_file),
+                file_id=file_id,
+            )
+        except Exception as e:
+            logger.warning(f"[WARNING] Failed to send per-user processed notification: {e}")
+
     return {"status": "Completed", "file": processed_file}

app/tasks/imap_tasks.py

@@ -18,6 +18,7 @@ from app.utils.allowed_types import (
     DEFAULT_CATEGORIES,
     get_allowed_types_for_categories,
 )
+from app.utils.network import is_private_ip
 
 # Database session for per-user IMAP accounts (imported lazily to avoid circular imports)
 _db_session_factory = None
@@ -405,6 +406,11 @@ def pull_inbox(
     )
     processed_emails = load_processed_emails()
 
+    # Security: Prevent SSRF by blocking connections to internal IPs
+    if is_private_ip(host):
+        logger.warning("SSRF blocked: Attempt to pull mailbox from private IP %s", host)
+        return
+
     try:
         mail = imaplib.IMAP4_SSL(host, port) if use_ssl else imaplib.IMAP4(host, port)
         mail.login(username, password)

app/tasks/send_to_all.py

@@ -18,6 +18,7 @@ from app.tasks.upload_to_onedrive import upload_to_onedrive
 from app.tasks.upload_to_paperless import upload_to_paperless
 from app.tasks.upload_to_s3 import upload_to_s3
 from app.tasks.upload_to_sftp import upload_to_sftp
+from app.tasks.upload_to_sharepoint import upload_to_sharepoint
 from app.tasks.upload_to_webdav import upload_to_webdav
 from app.utils.config_validator import get_provider_status
 from app.utils.logging import log_task_progress
@@ -121,6 +122,18 @@ def _should_upload_to_icloud():
     return bool(getattr(settings, "icloud_enabled", True) and settings.icloud_username and settings.icloud_password)
 
 
+def _should_upload_to_sharepoint():
+    return bool(
+        settings.sharepoint_client_id
+        and settings.sharepoint_client_secret
+        and settings.sharepoint_site_url
+        and (
+            settings.sharepoint_refresh_token
+            or (settings.sharepoint_tenant_id and settings.sharepoint_tenant_id != "common")
+        )
+    )
+
+
 def get_configured_services_from_validator():
     """
     Use the config validator to determine which services are configured and enabled.
@@ -140,6 +153,7 @@ def get_configured_services_from_validator():
         "Email": "email",
         "OneDrive": "onedrive",
         "S3 Storage": "s3",
+        "SharePoint": "sharepoint",
         "iCloud Drive": "icloud",
     }
 
@@ -250,6 +264,11 @@ def send_to_all_destinations(self, file_path: str, use_validator=True, file_id:
             "should_upload": _should_upload_to_s3,
             "upload_func": upload_to_s3,
         },
+        {
+            "name": "sharepoint",
+            "should_upload": _should_upload_to_sharepoint,
+            "upload_func": upload_to_sharepoint,
+        },
         {
             "name": "icloud",
             "should_upload": _should_upload_to_icloud,

app/tasks/translate_to_default_language.py

@@ -0,0 +1,141 @@
+#!/usr/bin/env python3
+"""Celery task to translate extracted document text into the default target language.
+
+This task is triggered after metadata extraction when the detected document
+language differs from the user's (or system) default document language.  The
+translated text is persisted in ``FileRecord.default_language_text`` so that
+users can always read a reference copy in their preferred language.
+
+Other ad-hoc translations are generated on the fly via the ``/api/files/{id}/translate``
+endpoint and are NOT persisted.
+"""
+
+import logging
+
+from app.celery_app import celery
+from app.config import settings
+from app.database import SessionLocal
+from app.models import FileRecord, UserProfile
+from app.tasks.retry_config import BaseTaskWithRetry
+from app.utils import log_task_progress
+from app.utils.ai_provider import get_ai_provider
+
+logger = logging.getLogger(__name__)
+
+
+def _resolve_default_language(owner_id: str | None) -> str:
+    """Return the default document language for the given owner.
+
+    Resolution order:
+    1. ``UserProfile.default_document_language`` (per-user override)
+    2. ``settings.default_document_language`` (global setting)
+    """
+    if owner_id:
+        with SessionLocal() as db:
+            profile = db.query(UserProfile).filter_by(user_id=owner_id).first()
+            if profile and profile.default_document_language:
+                return profile.default_document_language
+    return settings.default_document_language
+
+
+@celery.task(base=BaseTaskWithRetry, bind=True)
+def translate_to_default_language(
+    self,
+    file_id: int,
+    extracted_text: str,
+    detected_language: str,
+    owner_id: str | None = None,
+) -> dict:
+    """Translate *extracted_text* into the default document language and persist the result.
+
+    Args:
+        file_id: Primary key of the :class:`FileRecord`.
+        extracted_text: The OCR / refined text in the document's original language.
+        detected_language: ISO 639-1 code of the document's detected language.
+        owner_id: Owner identifier used to resolve per-user language preference.
+
+    Returns:
+        A dict with ``status``, ``target_language``, and the translated text length.
+    """
+    task_id = self.request.id
+    target_language = _resolve_default_language(owner_id)
+
+    # Nothing to do when the document is already in the target language.
+    if detected_language == target_language:
+        logger.info(
+            f"[{task_id}] Document {file_id} already in target language '{target_language}', skipping translation"
+        )
+        log_task_progress(
+            task_id,
+            "translate_to_default_language",
+            "skipped",
+            f"Document already in {target_language}",
+            file_id=file_id,
+        )
+        return {"status": "skipped", "reason": "already_in_target_language"}
+
+    logger.info(f"[{task_id}] Translating document {file_id} from '{detected_language}' to '{target_language}'")
+    log_task_progress(
+        task_id,
+        "translate_to_default_language",
+        "in_progress",
+        f"Translating from {detected_language} to {target_language}",
+        file_id=file_id,
+    )
+
+    try:
+        provider = get_ai_provider()
+        model = settings.ai_model or settings.openai_model
+        translated_text = provider.chat_completion(
+            messages=[
+                {
+                    "role": "system",
+                    "content": (
+                        f"You are a professional translator. Translate the following text "
+                        f"from {detected_language} to {target_language}. "
+                        f"Preserve the original formatting, paragraph structure, and meaning. "
+                        f"Do not add any commentary or explanation — output ONLY the translated text."
+                    ),
+                },
+                {"role": "user", "content": extracted_text},
+            ],
+            model=model,
+            temperature=0.3,
+        )
+
+        # Persist the translation.
+        with SessionLocal() as db:
+            record = db.query(FileRecord).filter_by(id=file_id).first()
+            if record:
+                record.default_language_text = translated_text
+                record.default_language_code = target_language
+                record.detected_language = detected_language
+                db.commit()
+                logger.info(
+                    f"[{task_id}] Stored default-language translation ({len(translated_text)} chars) for file {file_id}"
+                )
+
+        log_task_progress(
+            task_id,
+            "translate_to_default_language",
+            "success",
+            f"Translated {len(extracted_text)} → {len(translated_text)} chars ({detected_language} → {target_language})",
+            file_id=file_id,
+        )
+
+        return {
+            "status": "success",
+            "target_language": target_language,
+            "translated_length": len(translated_text),
+        }
+
+    except Exception as exc:
+        logger.exception(f"[{task_id}] Translation failed for file {file_id}: {exc}")
+        log_task_progress(
+            task_id,
+            "translate_to_default_language",
+            "failure",
+            f"Exception: {exc}",
+            file_id=file_id,
+        )
+        raise

app/tasks/upload_to_email.py

@@ -11,6 +11,7 @@ from email.mime.image import MIMEImage
 from email.mime.multipart import MIMEMultipart
 from email.mime.text import MIMEText
 
+import pypdf
 from jinja2 import Environment, FileSystemLoader, select_autoescape
 
 from app.celery_app import celery
@@ -23,6 +24,15 @@ logger = logging.getLogger(__name__)
 # Constants
 _LOGO_FILENAME = "logo.png"
 
+# Mapping from PDF metadata keys (with leading slash stripped) to application-specific names.
+# This mirrors the inverse of the mapping used in app/tasks/embed_metadata_into_pdf.py.
+_PDF_METADATA_KEY_MAP = {
+    "Title": "filename",
+    "Author": "absender",
+    "Subject": "document_type",
+    "Keywords": "tags",
+}
+
 
 def get_email_template(template_name="default.html"):
     """
@@ -63,9 +73,12 @@ def extract_metadata_from_file(file_path):
     """
     Try to extract metadata from a file using several methods:
     1. Check for a .json metadata file with the same name
-    2. Extract metadata from PDF if it's embedded
+    2. Extract embedded metadata from PDF using pypdf
 
-    Returns a dictionary of metadata or None if not found
+    JSON metadata takes precedence; embedded PDF metadata fills in any missing
+    fields using the application's standard key mapping (e.g., /Title → filename).
+
+    Returns a dictionary of metadata (may be empty if none found).
     """
     metadata = {}
 
@@ -76,12 +89,28 @@ def extract_metadata_from_file(file_path):
             with open(metadata_path, "r", encoding="utf-8") as f:
                 metadata = json.load(f)
                 logger.info(f"Loaded metadata from external JSON file: {metadata_path}")
-                return metadata
         except Exception as e:
             logger.warning(f"Failed to load metadata from JSON file: {str(e)}")
 
-    # TODO: For PDF files, try to extract embedded metadata using PyPDF2
-    # This would require additional dependencies, so for now we'll just check for external JSON
+    # Try to extract embedded metadata from PDF
+    if file_path.lower().endswith(".pdf") and os.path.exists(file_path):
+        try:
+            with open(file_path, "rb") as f:
+                pdf_reader = pypdf.PdfReader(f)
+                pdf_metadata = pdf_reader.metadata
+                if pdf_metadata:
+                    for key, value in pdf_metadata.items():
+                        # Remove the leading slash from PDF metadata keys (e.g., '/Title' -> 'Title')
+                        clean_key = key[1:] if key.startswith("/") else key
+                        # Map to application-specific key names where possible
+                        mapped_key = _PDF_METADATA_KEY_MAP.get(clean_key, clean_key)
+                        # Only set if not already present (JSON metadata takes precedence)
+                        if mapped_key not in metadata:
+                            metadata[mapped_key] = str(value)
+
+                    logger.info(f"Extracted embedded metadata from PDF: {file_path}")
+        except Exception as e:
+            logger.warning(f"Failed to extract metadata from PDF {file_path}: {str(e)}")
 
     return metadata
 

app/tasks/upload_to_nextcloud.py

@@ -1,157 +1,141 @@
-#!/usr/bin/env python3
-
-import logging
-import os
-
-import requests
-from requests.auth import HTTPBasicAuth
-
-from app.celery_app import celery
-from app.config import settings
-from app.tasks.retry_config import UploadTaskWithRetry
-from app.utils import log_task_progress
-from app.utils.filename_utils import extract_remote_path, get_unique_filename
-
-logger = logging.getLogger(__name__)
-
-
-@celery.task(base=UploadTaskWithRetry, bind=True)
-def upload_to_nextcloud(self, file_path: str, file_id: int = None, folder_override: str = None):
-    """
-    Upload a file to Nextcloud WebDAV.
-
-    Args:
-        file_path: Path to the file to upload
-        file_id: Optional file ID to associate with logs
-    """
-    task_id = self.request.id
-    logger.info(f"[{task_id}] Starting Nextcloud upload: {file_path}")
-    log_task_progress(
-        task_id,
-        "upload_to_nextcloud",
-        "in_progress",
-        f"Uploading to Nextcloud: {os.path.basename(file_path)}",
-        file_id=file_id,
-    )
-
-    if not os.path.exists(file_path):
-        error_msg = f"File not found: {file_path}"
-        logger.error(f"[{task_id}] {error_msg}")
-        log_task_progress(task_id, "upload_to_nextcloud", "failure", error_msg, file_id=file_id)
-        raise FileNotFoundError(error_msg)
-
-    # For Nextcloud, we need to check for 'nextcloud_upload_url' instead of 'nextcloud_url'
-    # This is what's shown in your env view
-    if not (
-        getattr(settings, "nextcloud_upload_url", None)
-        and getattr(settings, "nextcloud_username", None)
-        and getattr(settings, "nextcloud_password", None)
-    ):
-        logger.info(f"[{task_id}] Nextcloud upload skipped: Missing configuration")
-        log_task_progress(task_id, "upload_to_nextcloud", "success", "Skipped: Not configured", file_id=file_id)
-        return {"status": "Skipped", "reason": "Nextcloud settings not configured"}
-
-    filename = os.path.basename(file_path)
-
-    try:
-        # Prepare WebDAV URL - use nextcloud_upload_url instead of nextcloud_url
-        webdav_url = settings.nextcloud_upload_url
-        if not webdav_url.endswith("/"):
-            webdav_url += "/"
-
-        # Calculate remote path based on local file structure
-        remote_base = (
-            folder_override if folder_override is not None else (getattr(settings, "nextcloud_folder", "") or "")
-        )
-        remote_path = extract_remote_path(file_path, settings.workdir, remote_base)
-        full_url = f"{webdav_url}/{remote_path}"
-
-        # Remove any double slashes (except in http://)
-        full_url = full_url.replace("://", "$PLACEHOLDER$")
-        while "//" in full_url:
-            full_url = full_url.replace("//", "/")
-        full_url = full_url.replace("$PLACEHOLDER$", "://")
-
-        # Function to check if file exists in Nextcloud
-        def check_exists_in_nextcloud(path):
-            check_url = f"{webdav_url}{os.path.dirname(path)}"
-            try:
-                response = requests.request(
-                    "PROPFIND",
-                    check_url,
-                    auth=HTTPBasicAuth(settings.nextcloud_username, settings.nextcloud_password),
-                    headers={"Depth": "1"},
-                    timeout=10,
-                )
-
-                return path in response.text
-            except Exception:
-                # If we can't check, assume it doesn't exist
-                return False
-
-        # Check for potential file collision and get a unique name if needed
-        remote_path = get_unique_filename(remote_path, check_exists_in_nextcloud)
-        full_url = f"{webdav_url}/{remote_path}"
-
-        # Fix double slashes again
-        full_url = full_url.replace("://", "$PLACEHOLDER$")
-        while "//" in full_url:
-            full_url = full_url.replace("//", "/")
-        full_url = full_url.replace("$PLACEHOLDER$", "://")
-
-        # Create necessary parent folders
-        parent_dirs = os.path.dirname(remote_path)
-        if parent_dirs:
-            current_path = ""
-            for folder in parent_dirs.split("/"):
-                if not folder:
-                    continue
-                current_path += f"{folder}/"
-                mkdir_url = f"{webdav_url}/{current_path}"
-                # Fix double slashes
-                mkdir_url = mkdir_url.replace("://", "$PLACEHOLDER$")
-                while "//" in mkdir_url:
-                    mkdir_url = mkdir_url.replace("//", "/")
-                mkdir_url = mkdir_url.replace("$PLACEHOLDER$", "://")
-
-                requests.request(
-                    "MKCOL",
-                    mkdir_url,
-                    auth=HTTPBasicAuth(settings.nextcloud_username, settings.nextcloud_password),
-                    timeout=10,
-                )
-
-        # Upload the file
-        logger.info(f"[{task_id}] Uploading {filename} to Nextcloud at {full_url}")
-        log_task_progress(task_id, "upload_file", "in_progress", f"Uploading to {remote_path}", file_id=file_id)
-        with open(file_path, "rb") as file_data:
-            response = requests.put(
-                full_url,
-                data=file_data,
-                auth=HTTPBasicAuth(settings.nextcloud_username, settings.nextcloud_password),
-                headers={"Content-Type": "application/octet-stream"},
-                timeout=settings.http_request_timeout,  # Use configured timeout for large files
-            )
-
-        if response.status_code in (201, 204):  # Created or No Content
-            logger.info(f"[{task_id}] Successfully uploaded {filename} to Nextcloud at {remote_path}")
-            log_task_progress(
-                task_id, "upload_to_nextcloud", "success", f"Uploaded to Nextcloud: {remote_path}", file_id=file_id
-            )
-            return {
-                "status": "Completed",
-                "file_path": file_path,
-                "nextcloud_path": remote_path,
-                "response_code": response.status_code,
-            }
-        else:
-            error_msg = f"Failed to upload {filename} to Nextcloud: {response.status_code} - {response.text}"
-            logger.error(f"[{task_id}] {error_msg}")
-            log_task_progress(task_id, "upload_to_nextcloud", "failure", error_msg, file_id=file_id)
-            raise Exception(error_msg)
-
-    except Exception as e:
-        error_msg = f"Failed to upload {filename} to Nextcloud: {str(e)}"
-        logger.error(f"[{task_id}] {error_msg}")
-        log_task_progress(task_id, "upload_to_nextcloud", "failure", error_msg, file_id=file_id)
-        raise Exception(error_msg)
+#!/usr/bin/env python3
+
+import logging
+import os
+
+import requests
+from requests.auth import HTTPBasicAuth
+
+from app.celery_app import celery
+from app.config import settings
+from app.tasks.retry_config import UploadTaskWithRetry
+from app.utils import log_task_progress
+from app.utils.filename_utils import extract_remote_path, get_unique_filename
+from app.utils.network import join_url
+
+logger = logging.getLogger(__name__)
+
+
+@celery.task(base=UploadTaskWithRetry, bind=True)
+def upload_to_nextcloud(self, file_path: str, file_id: int = None, folder_override: str = None):
+    """
+    Upload a file to Nextcloud WebDAV.
+
+    Args:
+        file_path: Path to the file to upload
+        file_id: Optional file ID to associate with logs
+    """
+    task_id = self.request.id
+    logger.info(f"[{task_id}] Starting Nextcloud upload: {file_path}")
+    log_task_progress(
+        task_id,
+        "upload_to_nextcloud",
+        "in_progress",
+        f"Uploading to Nextcloud: {os.path.basename(file_path)}",
+        file_id=file_id,
+    )
+
+    if not os.path.exists(file_path):
+        error_msg = f"File not found: {file_path}"
+        logger.error(f"[{task_id}] {error_msg}")
+        log_task_progress(task_id, "upload_to_nextcloud", "failure", error_msg, file_id=file_id)
+        raise FileNotFoundError(error_msg)
+
+    # For Nextcloud, we need to check for 'nextcloud_upload_url' instead of 'nextcloud_url'
+    # This is what's shown in your env view
+    if not (
+        getattr(settings, "nextcloud_upload_url", None)
+        and getattr(settings, "nextcloud_username", None)
+        and getattr(settings, "nextcloud_password", None)
+    ):
+        logger.info(f"[{task_id}] Nextcloud upload skipped: Missing configuration")
+        log_task_progress(task_id, "upload_to_nextcloud", "success", "Skipped: Not configured", file_id=file_id)
+        return {"status": "Skipped", "reason": "Nextcloud settings not configured"}
+
+    filename = os.path.basename(file_path)
+
+    try:
+        # Prepare WebDAV URL - use nextcloud_upload_url instead of nextcloud_url
+        webdav_url = settings.nextcloud_upload_url
+        if not webdav_url.endswith("/"):
+            webdav_url += "/"
+
+        # Calculate remote path based on local file structure
+        remote_base = (
+            folder_override if folder_override is not None else (getattr(settings, "nextcloud_folder", "") or "")
+        )
+        remote_path = extract_remote_path(file_path, settings.workdir, remote_base)
+        full_url = join_url(webdav_url, remote_path)
+
+        # Function to check if file exists in Nextcloud
+        def check_exists_in_nextcloud(path):
+            check_url = join_url(webdav_url, os.path.dirname(path))
+            try:
+                response = requests.request(
+                    "PROPFIND",
+                    check_url,
+                    auth=HTTPBasicAuth(settings.nextcloud_username, settings.nextcloud_password),
+                    headers={"Depth": "1"},
+                    timeout=10,
+                )
+
+                return path in response.text
+            except Exception:
+                # If we can't check, assume it doesn't exist
+                return False
+
+        # Check for potential file collision and get a unique name if needed
+        remote_path = get_unique_filename(remote_path, check_exists_in_nextcloud)
+        full_url = join_url(webdav_url, remote_path)
+
+        # Create necessary parent folders
+        parent_dirs = os.path.dirname(remote_path)
+        if parent_dirs:
+            current_path = ""
+            for folder in parent_dirs.split("/"):
+                if not folder:
+                    continue
+                current_path += f"{folder}/"
+                mkdir_url = join_url(webdav_url, current_path)
+
+                requests.request(
+                    "MKCOL",
+                    mkdir_url,
+                    auth=HTTPBasicAuth(settings.nextcloud_username, settings.nextcloud_password),
+                    timeout=10,
+                )
+
+        # Upload the file
+        logger.info(f"[{task_id}] Uploading {filename} to Nextcloud at {full_url}")
+        log_task_progress(task_id, "upload_file", "in_progress", f"Uploading to {remote_path}", file_id=file_id)
+        with open(file_path, "rb") as file_data:
+            response = requests.put(
+                full_url,
+                data=file_data,
+                auth=HTTPBasicAuth(settings.nextcloud_username, settings.nextcloud_password),
+                headers={"Content-Type": "application/octet-stream"},
+                timeout=settings.http_request_timeout,  # Use configured timeout for large files
+            )
+
+        if response.status_code in (201, 204):  # Created or No Content
+            logger.info(f"[{task_id}] Successfully uploaded {filename} to Nextcloud at {remote_path}")
+            log_task_progress(
+                task_id, "upload_to_nextcloud", "success", f"Uploaded to Nextcloud: {remote_path}", file_id=file_id
+            )
+            return {
+                "status": "Completed",
+                "file_path": file_path,
+                "nextcloud_path": remote_path,
+                "response_code": response.status_code,
+            }
+        else:
+            error_msg = f"Failed to upload {filename} to Nextcloud: {response.status_code} - {response.text}"
+            logger.error(f"[{task_id}] {error_msg}")
+            log_task_progress(task_id, "upload_to_nextcloud", "failure", error_msg, file_id=file_id)
+            raise Exception(error_msg)
+
+    except Exception as e:
+        error_msg = f"Failed to upload {filename} to Nextcloud: {str(e)}"
+        logger.error(f"[{task_id}] {error_msg}")
+        log_task_progress(task_id, "upload_to_nextcloud", "failure", error_msg, file_id=file_id)
+        raise Exception(error_msg)

app/tasks/upload_to_sharepoint.py

@@ -0,0 +1,338 @@
+#!/usr/bin/env python3
+"""Upload documents to Microsoft SharePoint via the Microsoft Graph API.
+
+This module authenticates using MSAL (same OAuth2 flow as OneDrive) and
+uploads files to a configurable SharePoint Online document library using
+the chunked upload session approach for reliability with large files.
+
+Key differences from the OneDrive provider:
+- Uses ``/sites/{siteId}/drives/{driveId}`` instead of ``/me/drive``
+- Requires a SharePoint site URL to resolve the site and drive IDs
+- Targets a named document library (default: ``Documents``)
+"""
+
+import logging
+import os
+import time
+import urllib.parse
+
+import msal
+import requests
+
+from app.celery_app import celery
+from app.config import settings
+from app.tasks.retry_config import UploadTaskWithRetry
+from app.utils import log_task_progress
+
+logger = logging.getLogger(__name__)
+
+
+def get_sharepoint_token() -> str:
+    """Acquire a Microsoft Graph API access token for SharePoint.
+
+    Uses MSAL ``ConfidentialClientApplication`` with the refresh-token flow
+    (delegated permissions) or the client-credentials flow (application
+    permissions) depending on configuration.
+
+    Returns:
+        A valid access token string.
+
+    Raises:
+        ValueError: When required settings are missing or token acquisition fails.
+    """
+    if not settings.sharepoint_client_id or not settings.sharepoint_client_secret:
+        raise ValueError("SharePoint client ID and client secret must be configured")
+
+    tenant = settings.sharepoint_tenant_id or "common"
+    logger.info("Using SharePoint tenant: %s", tenant)
+
+    scopes = ["https://graph.microsoft.com/.default"]
+
+    if settings.sharepoint_refresh_token:
+        app = msal.ConfidentialClientApplication(
+            client_id=settings.sharepoint_client_id,
+            client_credential=settings.sharepoint_client_secret,
+            authority=f"https://login.microsoftonline.com/{tenant}",
+        )
+
+        logger.info("Attempting to acquire SharePoint token using refresh token")
+        token_response = app.acquire_token_by_refresh_token(
+            refresh_token=settings.sharepoint_refresh_token, scopes=scopes
+        )
+
+        if "access_token" not in token_response:
+            error = token_response.get("error", "")
+            error_desc = token_response.get("error_description", "Unknown error")
+            logger.error("Failed to get SharePoint access token: %s - %s", error, error_desc)
+            raise ValueError(f"Failed to get SharePoint access token: {error} - {error_desc}")
+
+        if "refresh_token" in token_response:
+            settings.sharepoint_refresh_token = token_response["refresh_token"]
+            logger.info("Updated SharePoint refresh token in memory")
+
+        return token_response["access_token"]
+
+    elif settings.sharepoint_tenant_id and settings.sharepoint_tenant_id != "common":
+        authority = f"https://login.microsoftonline.com/{settings.sharepoint_tenant_id}"
+        app = msal.ConfidentialClientApplication(
+            client_id=settings.sharepoint_client_id,
+            client_credential=settings.sharepoint_client_secret,
+            authority=authority,
+        )
+
+        token_response = app.acquire_token_for_client(scopes=scopes)
+
+        if "access_token" not in token_response:
+            error = token_response.get("error", "")
+            error_desc = token_response.get("error_description", "Unknown error")
+            raise ValueError(f"Failed to get SharePoint access token: {error} - {error_desc}")
+
+        return token_response["access_token"]
+
+    else:
+        raise ValueError("For SharePoint, either a refresh token or a non-'common' tenant ID is required")
+
+
+def resolve_sharepoint_drive(access_token: str, site_url: str, library_name: str) -> tuple[str, str]:
+    """Resolve the Graph API site ID and drive ID for a SharePoint site.
+
+    Args:
+        access_token: Valid Microsoft Graph API token.
+        site_url: Full SharePoint site URL, e.g.
+            ``https://tenant.sharepoint.com/sites/sitename``.
+        library_name: Display name of the document library (e.g. ``Documents``).
+
+    Returns:
+        A ``(site_id, drive_id)`` tuple.
+
+    Raises:
+        ValueError: When the site URL cannot be parsed.
+        RuntimeError: When the Graph API call fails.
+    """
+    parsed = urllib.parse.urlparse(site_url)
+    hostname = parsed.hostname
+    site_path = parsed.path.rstrip("/")
+
+    if not hostname or not site_path:
+        raise ValueError(
+            f"Invalid SharePoint site URL '{site_url}'. Expected format: https://tenant.sharepoint.com/sites/sitename"
+        )
+
+    headers = {"Authorization": f"Bearer {access_token}"}
+
+    # Resolve site ID
+    site_api_url = f"https://graph.microsoft.com/v1.0/sites/{hostname}:{site_path}"
+    logger.info("Resolving SharePoint site: %s", site_api_url)
+    resp = requests.get(site_api_url, headers=headers, timeout=settings.http_request_timeout)
+
+    if resp.status_code != 200:
+        raise RuntimeError(f"Failed to resolve SharePoint site: {resp.status_code} - {resp.text}")
+
+    site_id = resp.json()["id"]
+    logger.info("Resolved SharePoint site ID: %s", site_id)
+
+    # Resolve drive ID from the document library name
+    drives_url = f"https://graph.microsoft.com/v1.0/sites/{site_id}/drives"
+    resp = requests.get(drives_url, headers=headers, timeout=settings.http_request_timeout)
+
+    if resp.status_code != 200:
+        raise RuntimeError(f"Failed to list SharePoint drives: {resp.status_code} - {resp.text}")
+
+    drives = resp.json().get("value", [])
+    drive_id = None
+    for drive in drives:
+        if drive.get("name", "").lower() == library_name.lower():
+            drive_id = drive["id"]
+            break
+
+    if not drive_id:
+        available = [d.get("name") for d in drives]
+        raise RuntimeError(f"Document library '{library_name}' not found on site. Available libraries: {available}")
+
+    logger.info("Resolved SharePoint drive ID: %s (library: %s)", drive_id, library_name)
+    return site_id, drive_id
+
+
+def create_sharepoint_upload_session(
+    filename: str, folder_path: str | None, drive_id: str, site_id: str, access_token: str
+) -> str:
+    """Create a resumable upload session on a SharePoint document library.
+
+    Args:
+        filename: Name of the file to upload.
+        folder_path: Optional subfolder path inside the library.
+        drive_id: Graph API drive ID of the document library.
+        site_id: Graph API site ID.
+        access_token: Valid access token.
+
+    Returns:
+        The upload session URL for chunked PUT requests.
+
+    Raises:
+        RuntimeError: When session creation fails.
+    """
+    base_url = f"https://graph.microsoft.com/v1.0/sites/{site_id}/drives/{drive_id}"
+
+    if folder_path:
+        folder_path = folder_path.strip("/")
+        path_components = folder_path.split("/")
+        encoded_path = "/".join(urllib.parse.quote(component) for component in path_components)
+        encoded_filename = urllib.parse.quote(filename)
+        item_path = f"/root:/{encoded_path}/{encoded_filename}:/createUploadSession"
+    else:
+        encoded_filename = urllib.parse.quote(filename)
+        item_path = f"/root:/{encoded_filename}:/createUploadSession"
+
+    url = f"{base_url}{item_path}"
+    request_body = {"item": {"@microsoft.graph.conflictBehavior": "replace"}}
+    headers = {"Authorization": f"Bearer {access_token}", "Content-Type": "application/json"}
+
+    logger.info("Creating SharePoint upload session for %s at path %s", filename, folder_path)
+    response = requests.post(url, headers=headers, json=request_body, timeout=settings.http_request_timeout)
+
+    if response.status_code == 200:
+        upload_url = response.json().get("uploadUrl")
+        logger.info("SharePoint upload session created for %s", filename)
+        return upload_url
+    else:
+        raise RuntimeError(f"Failed to create SharePoint upload session: {response.status_code} - {response.text}")
+
+
+def upload_large_file_sharepoint(file_path: str, upload_url: str) -> dict:
+    """Upload a file to SharePoint using a chunked upload session.
+
+    Args:
+        file_path: Local path to the file.
+        upload_url: The upload session URL from ``create_sharepoint_upload_session``.
+
+    Returns:
+        The Graph API response dict containing file metadata.
+
+    Raises:
+        RuntimeError: When a chunk upload fails after retries.
+    """
+    file_size = os.path.getsize(file_path)
+    chunk_size = 10 * 1024 * 1024  # 10 MB
+
+    response = None
+    with open(file_path, "rb") as f:
+        chunk_number = 0
+        while True:
+            chunk = f.read(chunk_size)
+            if not chunk:
+                break
+
+            chunk_start = chunk_number * chunk_size
+            chunk_end = chunk_start + len(chunk) - 1
+            content_range = f"bytes {chunk_start}-{chunk_end}/{file_size}"
+
+            headers = {"Content-Length": str(len(chunk)), "Content-Range": content_range}
+
+            max_retries = 3
+            retry_delay = 2
+
+            for attempt in range(max_retries):
+                try:
+                    response = requests.put(
+                        upload_url, headers=headers, data=chunk, timeout=settings.http_request_timeout
+                    )
+                    if response.status_code in (201, 202):
+                        break
+                    else:
+                        logger.warning(
+                            "SharePoint chunk upload failed (attempt %d): %d", attempt + 1, response.status_code
+                        )
+                        if attempt < max_retries - 1:
+                            time.sleep(retry_delay * (attempt + 1))
+                except Exception as e:
+                    logger.warning("SharePoint chunk upload error (attempt %d): %s", attempt + 1, str(e))
+                    if attempt < max_retries - 1:
+                        time.sleep(retry_delay * (attempt + 1))
+
+            if response is None or response.status_code not in (201, 202):
+                status = response.status_code if response else "no response"
+                text = response.text if response else ""
+                raise RuntimeError(f"Failed to upload chunk after {max_retries} attempts: {status} - {text}")
+
+            chunk_number += 1
+
+    return response.json() if response else {}
+
+
+@celery.task(base=UploadTaskWithRetry, bind=True)
+def upload_to_sharepoint(self, file_path: str, file_id: int = None, folder_override: str = None):
+    """Upload a file to SharePoint Online.
+
+    Args:
+        file_path: Path to the file to upload.
+        file_id: Optional file ID to associate with logs.
+        folder_override: Optional folder path override.
+
+    Returns:
+        A dict with upload status and file details.
+
+    Raises:
+        FileNotFoundError: When the file does not exist.
+        ValueError: When SharePoint is not configured.
+        RuntimeError: When the upload fails.
+    """
+    task_id = self.request.id
+    logger.info("[%s] Starting SharePoint upload: %s", task_id, file_path)
+    log_task_progress(
+        task_id,
+        "upload_to_sharepoint",
+        "in_progress",
+        f"Uploading to SharePoint: {os.path.basename(file_path)}",
+        file_id=file_id,
+    )
+
+    if not os.path.exists(file_path):
+        error_msg = f"File not found: {file_path}"
+        logger.error("[%s] %s", task_id, error_msg)
+        log_task_progress(task_id, "upload_to_sharepoint", "failure", error_msg, file_id=file_id)
+        raise FileNotFoundError(error_msg)
+
+    filename = os.path.basename(file_path)
+
+    if not settings.sharepoint_client_id:
+        error_msg = "SharePoint client ID is not configured"
+        logger.error("[%s] %s", task_id, error_msg)
+        log_task_progress(task_id, "upload_to_sharepoint", "failure", error_msg, file_id=file_id)
+        raise ValueError(error_msg)
+
+    if not settings.sharepoint_site_url:
+        error_msg = "SharePoint site URL is not configured"
+        logger.error("[%s] %s", task_id, error_msg)
+        log_task_progress(task_id, "upload_to_sharepoint", "failure", error_msg, file_id=file_id)
+        raise ValueError(error_msg)
+
+    try:
+        access_token = get_sharepoint_token()
+
+        library_name = settings.sharepoint_document_library or "Documents"
+        site_id, drive_id = resolve_sharepoint_drive(access_token, settings.sharepoint_site_url, library_name)
+
+        folder_path = folder_override if folder_override is not None else settings.sharepoint_folder_path
+
+        upload_url = create_sharepoint_upload_session(filename, folder_path, drive_id, site_id, access_token)
+        result = upload_large_file_sharepoint(file_path, upload_url)
+
+        web_url = result.get("webUrl", "Not available")
+        logger.info("[%s] Successfully uploaded %s to SharePoint", task_id, filename)
+        logger.info("[%s] File accessible at: %s", task_id, web_url)
+        log_task_progress(
+            task_id, "upload_to_sharepoint", "success", f"Uploaded to SharePoint: {filename}", file_id=file_id
+        )
+
+        return {
+            "status": "Completed",
+            "file_path": file_path,
+            "sharepoint_path": f"{folder_path or ''}/{filename}",
+            "web_url": web_url,
+        }
+
+    except Exception as e:
+        error_msg = f"Failed to upload {filename} to SharePoint: {str(e)}"
+        logger.error("[%s] %s", task_id, error_msg)
+        log_task_progress(task_id, "upload_to_sharepoint", "failure", error_msg, file_id=file_id)
+        raise RuntimeError(error_msg) from e

app/tasks/upload_to_user_integration.py

@@ -571,6 +571,113 @@ def _upload_rclone(file_path: str, cfg: dict[str, Any], creds: dict[str, Any], t
     return {"status": "Completed", "rclone_dest": dest}
 
 
+def _upload_sharepoint(file_path: str, cfg: dict[str, Any], creds: dict[str, Any], task_id: str) -> dict[str, Any]:
+    """Upload *file_path* to SharePoint using per-user MSAL credentials."""
+    import urllib.parse
+
+    import msal
+    import requests as _requests
+
+    client_id = creds.get("client_id") or ""
+    client_secret = creds.get("client_secret") or ""
+    refresh_token = creds.get("refresh_token") or ""
+    tenant = cfg.get("tenant_id") or "common"
+    site_url = cfg.get("site_url") or ""
+    library_name = cfg.get("document_library") or "Documents"
+    folder_path = cfg.get("folder_path") or ""
+
+    if not (client_id and client_secret):
+        raise ValueError("SharePoint integration is missing client_id or client_secret in credentials")
+    if not site_url:
+        raise ValueError("SharePoint integration is missing site_url in config")
+
+    scopes = ["https://graph.microsoft.com/.default"]
+    msal_app = msal.ConfidentialClientApplication(
+        client_id=client_id,
+        client_credential=client_secret,
+        authority=f"https://login.microsoftonline.com/{tenant}",
+    )
+
+    if refresh_token:
+        token_resp = msal_app.acquire_token_by_refresh_token(refresh_token=refresh_token, scopes=scopes)
+    else:
+        token_resp = msal_app.acquire_token_for_client(scopes=scopes)
+
+    if "access_token" not in token_resp:
+        raise ValueError(f"SharePoint token acquisition failed: {token_resp.get('error_description', 'unknown')}")
+
+    access_token = token_resp["access_token"]
+    headers = {"Authorization": f"Bearer {access_token}"}
+
+    # Resolve site ID
+    parsed = urllib.parse.urlparse(site_url)
+    hostname = parsed.hostname
+    site_path = parsed.path.rstrip("/")
+    if not hostname or not site_path:
+        raise ValueError(f"Invalid SharePoint site URL: {site_url}")
+
+    resp = _requests.get(f"https://graph.microsoft.com/v1.0/sites/{hostname}:{site_path}", headers=headers, timeout=30)
+    resp.raise_for_status()
+    site_id = resp.json()["id"]
+
+    # Resolve drive ID
+    resp = _requests.get(f"https://graph.microsoft.com/v1.0/sites/{site_id}/drives", headers=headers, timeout=30)
+    resp.raise_for_status()
+    drive_id = None
+    for drive in resp.json().get("value", []):
+        if drive.get("name", "").lower() == library_name.lower():
+            drive_id = drive["id"]
+            break
+    if not drive_id:
+        raise RuntimeError(f"Document library '{library_name}' not found on SharePoint site")
+
+    filename = os.path.basename(file_path)
+
+    # Build upload-session URL
+    base_url = f"https://graph.microsoft.com/v1.0/sites/{site_id}/drives/{drive_id}"
+    if folder_path:
+        folder_path = folder_path.strip("/")
+        encoded_path = "/".join(urllib.parse.quote(p) for p in folder_path.split("/"))
+        encoded_file = urllib.parse.quote(filename)
+        item_path = f"/root:/{encoded_path}/{encoded_file}:/createUploadSession"
+    else:
+        encoded_file = urllib.parse.quote(filename)
+        item_path = f"/root:/{encoded_file}:/createUploadSession"
+
+    session_url = f"{base_url}{item_path}"
+    session_headers = {"Authorization": f"Bearer {access_token}", "Content-Type": "application/json"}
+    resp = _requests.post(
+        session_url,
+        headers=session_headers,
+        json={"item": {"@microsoft.graph.conflictBehavior": "replace"}},
+        timeout=30,
+    )
+    resp.raise_for_status()
+    upload_url = resp.json()["uploadUrl"]
+
+    file_size = os.path.getsize(file_path)
+    chunk_size = 10 * 1024 * 1024
+    with open(file_path, "rb") as fh:
+        chunk_num = 0
+        while True:
+            chunk = fh.read(chunk_size)
+            if not chunk:
+                break
+            start = chunk_num * chunk_size
+            end = start + len(chunk) - 1
+            upload_headers = {
+                "Content-Length": str(len(chunk)),
+                "Content-Range": f"bytes {start}-{end}/{file_size}",
+            }
+            upload_resp = _requests.put(upload_url, headers=upload_headers, data=chunk, timeout=120)
+            if upload_resp.status_code not in (201, 202):
+                raise RuntimeError(f"SharePoint chunk upload failed: {upload_resp.status_code}")
+            chunk_num += 1
+
+    logger.info("[%s] SharePoint upload complete: %s/%s", task_id, folder_path, filename)
+    return {"status": "Completed", "sharepoint_folder": folder_path, "filename": filename}
+
+
 def _upload_icloud(file_path: str, cfg: dict[str, Any], creds: dict[str, Any], task_id: str) -> dict[str, Any]:
     """Upload *file_path* to iCloud Drive using per-user credentials.
 
@@ -615,6 +722,7 @@ _UPLOAD_HANDLERS = {
     IntegrationType.PAPERLESS: _upload_paperless,
     IntegrationType.EMAIL: _upload_email,
     IntegrationType.RCLONE: _upload_rclone,
+    IntegrationType.SHAREPOINT: _upload_sharepoint,
     IntegrationType.ICLOUD: _upload_icloud,
 }
 

app/tasks/upload_with_rclone.py

@@ -55,12 +55,12 @@ def upload_with_rclone(self, file_path: str, destination: str):
 
     try:
         # Ensure the remote path exists (create folders if needed)
-        mkdir_cmd = ["rclone", "mkdir", "--config", rclone_config_path, destination]
+        mkdir_cmd = ["rclone", "mkdir", "--config", rclone_config_path, "--", destination]
 
         subprocess.run(mkdir_cmd, check=True, capture_output=True)  # noqa: S603
 
         # Construct the upload command
-        upload_cmd = ["rclone", "copy", "--config", rclone_config_path, file_path, destination, "--progress"]
+        upload_cmd = ["rclone", "copy", "--config", rclone_config_path, "--progress", "--", file_path, destination]
 
         log_task_progress(task_id, "rclone_upload", "in_progress", f"Executing rclone copy to {destination}")
 
@@ -71,7 +71,7 @@ def upload_with_rclone(self, file_path: str, destination: str):
         if result.returncode == 0:
             # Try to get a public link if possible
             try:
-                link_cmd = ["rclone", "link", "--config", rclone_config_path, f"{destination}/{filename}"]
+                link_cmd = ["rclone", "link", "--config", rclone_config_path, "--", f"{destination}/{filename}"]
                 link_result = subprocess.run(link_cmd, capture_output=True, text=True, check=False)  # noqa: S603
                 public_url = link_result.stdout.strip() if link_result.returncode == 0 else None
             except (subprocess.SubprocessError, OSError) as e: