Merge branch 'main' into copilot/add-document-sharing-feature

app/api/__init__.py

@@ -20,6 +20,7 @@ from app.api.google_drive import router as google_drive_router
 from app.api.imap_accounts import router as imap_accounts_router
 from app.api.integrations import router as integrations_router
 from app.api.logs import router as logs_router
+from app.api.notifications import router as notifications_router
 from app.api.onboarding import router as onboarding_router
 from app.api.onedrive import router as onedrive_router
 from app.api.openai import router as openai_router
@@ -78,3 +79,4 @@ router.include_router(billing_router)
 router.include_router(pipelines_router)
 router.include_router(imap_accounts_router)
 router.include_router(integrations_router)
+router.include_router(notifications_router)

app/api/notifications.py

@@ -0,0 +1,484 @@
+"""API endpoints for per-user notification targets, preferences, and in-app inbox.
+
+Users can define notification targets (email via SMTP, webhook via HTTP POST)
+and configure which document events trigger which targets.  In-app notifications
+are always created and surfaced via the bell icon / inbox endpoints.
+"""
+
+import json
+import logging
+from typing import Annotated, Any
+
+from fastapi import APIRouter, Depends, HTTPException, Request, status
+from pydantic import BaseModel, Field
+from sqlalchemy.orm import Session
+
+from app.database import get_db
+from app.models import InAppNotification, UserNotificationPreference, UserNotificationTarget
+from app.utils.user_notification import USER_EVENT_LABELS
+from app.utils.user_scope import get_current_owner_id
+
+logger = logging.getLogger(__name__)
+router = APIRouter(prefix="/user-notifications", tags=["user-notifications"])
+
+DbSession = Annotated[Session, Depends(get_db)]
+
+# ---------------------------------------------------------------------------
+# Auth helper (mirrors api_tokens.py pattern)
+# ---------------------------------------------------------------------------
+
+
+def _get_owner_id(request: Request) -> str:
+    """Return the current user's owner ID, raising 401 if unauthenticated."""
+    owner_id = get_current_owner_id(request)
+    if not owner_id:
+        raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Not authenticated")
+    return owner_id
+
+
+CurrentOwner = Annotated[str, Depends(_get_owner_id)]
+
+# ---------------------------------------------------------------------------
+# Pydantic schemas
+# ---------------------------------------------------------------------------
+
+VALID_CHANNEL_TYPES = {"email", "webhook"}
+VALID_EVENT_TYPES = set(USER_EVENT_LABELS.keys())
+
+
+class NotificationTargetCreate(BaseModel):
+    """Schema for creating a new notification target."""
+
+    channel_type: str = Field(..., pattern="^(email|webhook)$")
+    name: str = Field(..., min_length=1, max_length=255)
+    config: dict[str, Any] = Field(default_factory=dict)
+    is_active: bool = True
+
+
+class NotificationTargetUpdate(BaseModel):
+    """Schema for updating an existing notification target."""
+
+    name: str | None = Field(None, min_length=1, max_length=255)
+    config: dict[str, Any] | None = None
+    is_active: bool | None = None
+
+
+class PreferenceItem(BaseModel):
+    """A single preference toggle for one event+channel combination."""
+
+    is_enabled: bool
+    target_id: int | None = None
+
+
+class PreferenceItemFull(BaseModel):
+    """Full preference item including event and channel type (used in bulk update)."""
+
+    event_type: str
+    channel_type: str
+    is_enabled: bool
+    target_id: int | None = None
+
+
+class PreferencesUpdate(BaseModel):
+    """Bulk preferences update payload — a flat list of preference items."""
+
+    preferences: list[PreferenceItemFull]
+
+
+# ---------------------------------------------------------------------------
+# Helpers
+# ---------------------------------------------------------------------------
+
+
+def _mask_email_config(config: dict[str, Any]) -> dict[str, Any]:
+    """Return a copy of an email config dict with the password masked."""
+    masked = dict(config)
+    if masked.get("smtp_password"):
+        masked["smtp_password"] = "****"
+    return masked
+
+
+def _target_to_dict(target: UserNotificationTarget) -> dict[str, Any]:
+    """Serialize a UserNotificationTarget to a response dict, masking secrets."""
+    config: dict[str, Any] = {}
+    if target.config:
+        try:
+            config = json.loads(target.config)
+        except (json.JSONDecodeError, ValueError):
+            config = {}
+
+    if target.channel_type == "email":
+        config = _mask_email_config(config)
+
+    return {
+        "id": target.id,
+        "channel_type": target.channel_type,
+        "name": target.name,
+        "config": config,
+        "is_active": target.is_active,
+        "created_at": target.created_at,
+        "updated_at": target.updated_at,
+    }
+
+
+# ---------------------------------------------------------------------------
+# Inbox endpoints
+# ---------------------------------------------------------------------------
+
+
+@router.get("/inbox")
+async def list_inbox(
+    owner_id: CurrentOwner,
+    db: DbSession,
+    skip: int = 0,
+    limit: int = 50,
+) -> list[dict[str, Any]]:
+    """List in-app notifications for the authenticated user, newest first."""
+    notifications = (
+        db.query(InAppNotification)
+        .filter(InAppNotification.owner_id == owner_id)
+        .order_by(InAppNotification.created_at.desc())
+        .offset(skip)
+        .limit(limit)
+        .all()
+    )
+    return [
+        {
+            "id": n.id,
+            "event_type": n.event_type,
+            "title": n.title,
+            "message": n.message,
+            "is_read": n.is_read,
+            "file_id": n.file_id,
+            "created_at": n.created_at,
+        }
+        for n in notifications
+    ]
+
+
+@router.get("/inbox/unread-count")
+async def unread_count(
+    owner_id: CurrentOwner,
+    db: DbSession,
+) -> dict[str, int]:
+    """Return the number of unread in-app notifications."""
+    count = (
+        db.query(InAppNotification)
+        .filter(InAppNotification.owner_id == owner_id, InAppNotification.is_read == False)  # noqa: E712
+        .count()
+    )
+    return {"count": count}
+
+
+@router.post("/inbox/{notification_id}/read", status_code=status.HTTP_200_OK)
+async def mark_read(
+    notification_id: int,
+    owner_id: CurrentOwner,
+    db: DbSession,
+) -> dict[str, str]:
+    """Mark a single in-app notification as read."""
+    notif = (
+        db.query(InAppNotification)
+        .filter(InAppNotification.id == notification_id, InAppNotification.owner_id == owner_id)
+        .first()
+    )
+    if not notif:
+        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Notification not found")
+    try:
+        notif.is_read = True
+        db.commit()
+    except Exception:
+        db.rollback()
+        raise
+    return {"detail": "Marked as read"}
+
+
+@router.post("/inbox/read-all", status_code=status.HTTP_200_OK)
+async def mark_all_read(
+    owner_id: CurrentOwner,
+    db: DbSession,
+) -> dict[str, str]:
+    """Mark all in-app notifications as read for the authenticated user."""
+    try:
+        db.query(InAppNotification).filter(
+            InAppNotification.owner_id == owner_id,
+            InAppNotification.is_read == False,  # noqa: E712
+        ).update({"is_read": True})
+        db.commit()
+    except Exception:
+        db.rollback()
+        raise
+    return {"detail": "All notifications marked as read"}
+
+
+# ---------------------------------------------------------------------------
+# Notification target endpoints
+# ---------------------------------------------------------------------------
+
+
+@router.get("/targets")
+async def list_targets(
+    owner_id: CurrentOwner,
+    db: DbSession,
+) -> list[dict[str, Any]]:
+    """List all notification targets for the authenticated user."""
+    targets = (
+        db.query(UserNotificationTarget)
+        .filter(UserNotificationTarget.owner_id == owner_id)
+        .order_by(UserNotificationTarget.created_at.desc())
+        .all()
+    )
+    return [_target_to_dict(t) for t in targets]
+
+
+@router.post("/targets", status_code=status.HTTP_201_CREATED)
+async def create_target(
+    body: NotificationTargetCreate,
+    owner_id: CurrentOwner,
+    db: DbSession,
+) -> dict[str, Any]:
+    """Create a new notification target (email or webhook)."""
+    target = UserNotificationTarget(
+        owner_id=owner_id,
+        channel_type=body.channel_type,
+        name=body.name,
+        config=json.dumps(body.config),
+        is_active=body.is_active,
+    )
+    try:
+        db.add(target)
+        db.commit()
+        db.refresh(target)
+    except Exception:
+        db.rollback()
+        raise
+
+    logger.info("Notification target created: id=%s owner=%s type=%s", target.id, owner_id, body.channel_type)
+    return _target_to_dict(target)
+
+
+@router.put("/targets/{target_id}", status_code=status.HTTP_200_OK)
+async def update_target(
+    target_id: int,
+    body: NotificationTargetUpdate,
+    owner_id: CurrentOwner,
+    db: DbSession,
+) -> dict[str, Any]:
+    """Update an existing notification target."""
+    target = (
+        db.query(UserNotificationTarget)
+        .filter(UserNotificationTarget.id == target_id, UserNotificationTarget.owner_id == owner_id)
+        .first()
+    )
+    if not target:
+        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Target not found")
+
+    try:
+        if body.name is not None:
+            target.name = body.name
+        if body.config is not None:
+            # Merge new config over existing, preserving masked password field if unchanged
+            existing_config: dict[str, Any] = {}
+            if target.config:
+                try:
+                    existing_config = json.loads(target.config)
+                except (json.JSONDecodeError, ValueError):
+                    existing_config = {}
+            merged = dict(existing_config)
+            for k, v in body.config.items():
+                # Skip writing back a masked password placeholder
+                if k == "smtp_password" and v == "****":
+                    continue
+                merged[k] = v
+            target.config = json.dumps(merged)
+        if body.is_active is not None:
+            target.is_active = body.is_active
+        db.commit()
+        db.refresh(target)
+    except Exception:
+        db.rollback()
+        raise
+
+    logger.info("Notification target updated: id=%s owner=%s", target_id, owner_id)
+    return _target_to_dict(target)
+
+
+@router.delete("/targets/{target_id}", status_code=status.HTTP_200_OK)
+async def delete_target(
+    target_id: int,
+    owner_id: CurrentOwner,
+    db: DbSession,
+) -> dict[str, str]:
+    """Delete a notification target and its associated preferences."""
+    target = (
+        db.query(UserNotificationTarget)
+        .filter(UserNotificationTarget.id == target_id, UserNotificationTarget.owner_id == owner_id)
+        .first()
+    )
+    if not target:
+        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Target not found")
+
+    try:
+        # Remove any preferences that reference this target
+        db.query(UserNotificationPreference).filter(
+            UserNotificationPreference.owner_id == owner_id,
+            UserNotificationPreference.target_id == target_id,
+        ).delete()
+        db.delete(target)
+        db.commit()
+    except Exception:
+        db.rollback()
+        raise
+
+    logger.info("Notification target deleted: id=%s owner=%s", target_id, owner_id)
+    return {"detail": "Target deleted"}
+
+
+@router.post("/targets/{target_id}/test", status_code=status.HTTP_200_OK)
+async def test_target(
+    target_id: int,
+    owner_id: CurrentOwner,
+    db: DbSession,
+) -> dict[str, str]:
+    """Send a test notification to the specified target."""
+    target = (
+        db.query(UserNotificationTarget)
+        .filter(UserNotificationTarget.id == target_id, UserNotificationTarget.owner_id == owner_id)
+        .first()
+    )
+    if not target:
+        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Target not found")
+
+    config: dict[str, Any] = {}
+    if target.config:
+        try:
+            config = json.loads(target.config)
+        except (json.JSONDecodeError, ValueError):
+            config = {}
+
+    title = "DocuElevate Test Notification"
+    message = f"This is a test notification from DocuElevate for target '{target.name}'."
+
+    if target.channel_type == "email":
+        from app.utils.user_notification import _send_email_notification
+
+        ok = _send_email_notification(config, title, message)
+    elif target.channel_type == "webhook":
+        from app.utils.user_notification import _send_webhook_notification
+
+        ok = _send_webhook_notification(config, "test", title, message)
+    else:
+        raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="Unknown channel type")
+
+    if not ok:
+        raise HTTPException(status_code=status.HTTP_502_BAD_GATEWAY, detail="Failed to send test notification")
+
+    return {"detail": "Test notification sent"}
+
+
+# ---------------------------------------------------------------------------
+# Preferences endpoints
+# ---------------------------------------------------------------------------
+
+
+@router.get("/preferences")
+async def get_preferences(
+    owner_id: CurrentOwner,
+    db: DbSession,
+) -> dict[str, Any]:
+    """Return all notification preferences for the authenticated user.
+
+    Response structure:
+        {
+          "event_types": ["document.processed", "document.failed"],
+          "event_labels": {"document.processed": "Document Processed", ...},
+          "preferences": {
+            "document.processed": {
+              "in_app": {"is_enabled": true, "target_id": null},
+              "email": {"is_enabled": false, "target_id": 1},
+              ...
+            }
+          }
+        }
+    """
+    prefs = db.query(UserNotificationPreference).filter(UserNotificationPreference.owner_id == owner_id).all()
+
+    # Build nested dict: event_type -> channel_type -> {is_enabled, target_id}
+    result: dict[str, dict[str, dict[str, Any]]] = {}
+    for pref in prefs:
+        result.setdefault(pref.event_type, {})[pref.channel_type] = {
+            "is_enabled": pref.is_enabled,
+            "target_id": pref.target_id,
+        }
+
+    return {
+        "event_types": list(USER_EVENT_LABELS.keys()),
+        "event_labels": USER_EVENT_LABELS,
+        "preferences": result,
+    }
+
+
+@router.put("/preferences", status_code=status.HTTP_200_OK)
+async def update_preferences(
+    body: PreferencesUpdate,
+    owner_id: CurrentOwner,
+    db: DbSession,
+) -> dict[str, str]:
+    """Bulk upsert notification preferences for the authenticated user.
+
+    Validates that any referenced target_id belongs to the requesting user.
+    """
+    # Collect all target IDs referenced in the payload for ownership validation
+    referenced_target_ids: set[int] = set()
+    for item in body.preferences:
+        if item.target_id is not None:
+            referenced_target_ids.add(item.target_id)
+
+    if referenced_target_ids:
+        owned_ids = {
+            row.id
+            for row in db.query(UserNotificationTarget.id)
+            .filter(
+                UserNotificationTarget.owner_id == owner_id,
+                UserNotificationTarget.id.in_(referenced_target_ids),
+            )
+            .all()
+        }
+        invalid = referenced_target_ids - owned_ids
+        if invalid:
+            raise HTTPException(
+                status_code=status.HTTP_400_BAD_REQUEST,
+                detail=f"Invalid or inaccessible target_id(s): {sorted(invalid)}",
+            )
+
+    try:
+        for item in body.preferences:
+            existing = (
+                db.query(UserNotificationPreference)
+                .filter(
+                    UserNotificationPreference.owner_id == owner_id,
+                    UserNotificationPreference.event_type == item.event_type,
+                    UserNotificationPreference.channel_type == item.channel_type,
+                    UserNotificationPreference.target_id == item.target_id,
+                )
+                .first()
+            )
+            if existing:
+                existing.is_enabled = item.is_enabled
+            else:
+                db.add(
+                    UserNotificationPreference(
+                        owner_id=owner_id,
+                        event_type=item.event_type,
+                        channel_type=item.channel_type,
+                        target_id=item.target_id,
+                        is_enabled=item.is_enabled,
+                    )
+                )
+        db.commit()
+    except Exception:
+        db.rollback()
+        raise
+
+    logger.info("Notification preferences updated for owner=%s", owner_id)
+    return {"detail": "Preferences updated"}

app/api/pipelines.py

@@ -51,7 +51,48 @@ PIPELINE_STEP_TYPES: dict[str, dict[str, Any]] = {
                 "type": "boolean",
                 "default": False,
                 "description": "Always use cloud OCR even if the PDF already has embedded text.",
-            }
+            },
+            "ocr_language": {
+                "type": "select",
+                "default": "auto",
+                "description": (
+                    "Language(s) used for OCR text extraction.  Applies to Tesseract and EasyOCR "
+                    "providers; Azure and Mistral perform auto-detection by default.  "
+                    "Use Tesseract codes such as 'eng', 'deu', or 'eng+deu' for multi-language "
+                    "documents.  'auto' falls back to the global system setting."
+                ),
+                "options": [
+                    {"value": "auto", "label": "Auto (use system default)"},
+                    {"value": "ara", "label": "Arabic"},
+                    {"value": "chi_sim", "label": "Chinese (Simplified)"},
+                    {"value": "chi_tra", "label": "Chinese (Traditional)"},
+                    {"value": "ces", "label": "Czech"},
+                    {"value": "dan", "label": "Danish"},
+                    {"value": "nld", "label": "Dutch"},
+                    {"value": "eng", "label": "English"},
+                    {"value": "fin", "label": "Finnish"},
+                    {"value": "fra", "label": "French"},
+                    {"value": "deu", "label": "German"},
+                    {"value": "ell", "label": "Greek"},
+                    {"value": "heb", "label": "Hebrew"},
+                    {"value": "hin", "label": "Hindi"},
+                    {"value": "hun", "label": "Hungarian"},
+                    {"value": "ita", "label": "Italian"},
+                    {"value": "jpn", "label": "Japanese"},
+                    {"value": "kor", "label": "Korean"},
+                    {"value": "nor", "label": "Norwegian"},
+                    {"value": "pol", "label": "Polish"},
+                    {"value": "por", "label": "Portuguese"},
+                    {"value": "ron", "label": "Romanian"},
+                    {"value": "rus", "label": "Russian"},
+                    {"value": "spa", "label": "Spanish"},
+                    {"value": "swe", "label": "Swedish"},
+                    {"value": "tha", "label": "Thai"},
+                    {"value": "tur", "label": "Turkish"},
+                    {"value": "ukr", "label": "Ukrainian"},
+                    {"value": "vie", "label": "Vietnamese"},
+                ],
+            },
         },
     },
     "extract_metadata": {

app/cli.py

@@ -0,0 +1,680 @@
+"""DocuElevate command-line interface.
+
+Provides a pipe-friendly CLI for scripting and automation against the
+DocuElevate REST API.  Authentication is via personal API tokens (the
+same tokens managed at ``/api-tokens`` in the web UI).
+
+Usage::
+
+    docuelevate --url http://my-instance --token de_xxx list
+    DOCUELEVATE_URL=http://my-instance DOCUELEVATE_API_TOKEN=de_xxx docuelevate list
+
+Commands
+--------
+upload      Upload one or more local files for processing.
+download    Download a processed (or original) file by ID.
+search      Full-text search across all documents.
+list        List documents with optional filtering.
+token       Sub-commands: create / list / revoke API tokens.
+"""
+
+import json
+import sys
+from pathlib import Path
+from typing import Any
+from urllib.parse import unquote
+
+import click
+import requests
+
+# ---------------------------------------------------------------------------
+# Environment-variable defaults
+# ---------------------------------------------------------------------------
+
+ENV_URL = "DOCUELEVATE_URL"
+ENV_TOKEN = "DOCUELEVATE_API_TOKEN"
+
+_DEFAULT_URL = "http://localhost:8000"
+
+
+# ---------------------------------------------------------------------------
+# Helpers
+# ---------------------------------------------------------------------------
+
+
+def _build_headers(token: str) -> dict[str, str]:
+    """Return Authorization headers for the given API token."""
+    return {"Authorization": f"Bearer {token}"}
+
+
+def _api(
+    method: str,
+    base_url: str,
+    path: str,
+    token: str,
+    timeout: int = 60,
+    **kwargs: Any,
+) -> requests.Response:
+    """Make an authenticated API request and return the response.
+
+    Args:
+        method: HTTP method (GET, POST, DELETE, …).
+        base_url: The base URL of the DocuElevate instance.
+        path: API path starting with ``/``.
+        token: Plaintext API token.
+        timeout: Request timeout in seconds (default: 60).
+        **kwargs: Extra keyword arguments forwarded to :func:`requests.request`.
+
+    Returns:
+        The :class:`requests.Response` object.
+
+    Raises:
+        click.ClickException: On network errors.
+    """
+    url = base_url.rstrip("/") + path
+    headers = _build_headers(token)
+    try:
+        resp = requests.request(method, url, headers=headers, timeout=timeout, **kwargs)
+    except requests.ConnectionError as exc:
+        raise click.ClickException(f"Could not connect to {base_url}: {exc}") from exc
+    except requests.Timeout as exc:
+        raise click.ClickException(f"Request timed out: {exc}") from exc
+    return resp
+
+
+def _require_ok(resp: requests.Response) -> dict[str, Any] | list[Any]:
+    """Assert a successful HTTP response and return parsed JSON.
+
+    Args:
+        resp: The response to check.
+
+    Returns:
+        Parsed JSON payload.
+
+    Raises:
+        click.ClickException: If the response status indicates an error.
+    """
+    if resp.status_code >= 400:
+        try:
+            detail = resp.json().get("detail", resp.text)
+        except Exception:
+            detail = resp.text
+        raise click.ClickException(f"API error {resp.status_code}: {detail}")
+    try:
+        return resp.json()
+    except Exception:
+        return {}
+
+
+def _output(data: Any, fmt: str) -> None:
+    """Write *data* to stdout in the requested format.
+
+    Args:
+        data: The value to serialise (dict, list, or primitive).
+        fmt: Either ``"json"`` (machine-readable) or ``"table"`` (human-readable).
+    """
+    if fmt == "json":
+        click.echo(json.dumps(data, indent=2, default=str))
+    else:
+        _print_table(data)
+
+
+def _print_table(data: Any) -> None:
+    """Pretty-print a list of dicts as a fixed-width table.
+
+    Falls back to JSON if the data is not a homogeneous list of dicts.
+
+    Args:
+        data: Data to render.
+    """
+    if isinstance(data, dict):
+        # Single-object output — print as key: value pairs
+        for key, value in data.items():
+            click.echo(f"  {key}: {value}")
+        return
+
+    if not isinstance(data, list) or not data:
+        click.echo(json.dumps(data, indent=2, default=str))
+        return
+
+    if not isinstance(data[0], dict):
+        for item in data:
+            click.echo(str(item))
+        return
+
+    # Determine column widths
+    keys = list(data[0].keys())
+    widths: dict[str, int] = {k: len(k) for k in keys}
+    for row in data:
+        for k in keys:
+            widths[k] = max(widths[k], len(str(row.get(k, ""))))
+
+    header = "  ".join(k.upper().ljust(widths[k]) for k in keys)
+    separator = "  ".join("-" * widths[k] for k in keys)
+    click.echo(header)
+    click.echo(separator)
+    for row in data:
+        click.echo("  ".join(str(row.get(k, "")).ljust(widths[k]) for k in keys))
+
+
+# ---------------------------------------------------------------------------
+# Root command group
+# ---------------------------------------------------------------------------
+
+
+@click.group(context_settings={"help_option_names": ["-h", "--help"]})
+@click.option(
+    "--url",
+    envvar=ENV_URL,
+    default=_DEFAULT_URL,
+    show_default=True,
+    show_envvar=True,
+    help="Base URL of the DocuElevate instance.",
+    metavar="URL",
+)
+@click.option(
+    "--token",
+    envvar=ENV_TOKEN,
+    default=None,
+    show_envvar=True,
+    help="API token (de_…).  Required for all commands except help.",
+    metavar="TOKEN",
+)
+@click.option(
+    "--format",
+    "fmt",
+    type=click.Choice(["table", "json"], case_sensitive=False),
+    default="table",
+    show_default=True,
+    help="Output format.  Use 'json' for machine-readable / pipe-friendly output.",
+)
+@click.option(
+    "--timeout",
+    default=60,
+    show_default=True,
+    envvar="DOCUELEVATE_TIMEOUT",
+    show_envvar=True,
+    type=int,
+    help="HTTP request timeout in seconds.",
+)
+@click.version_option(package_name="docuelevate", prog_name="docuelevate")
+@click.pass_context
+def cli(ctx: click.Context, url: str, token: str | None, fmt: str, timeout: int) -> None:
+    """DocuElevate CLI — interact with DocuElevate from the command line.
+
+    Configure the target instance and credentials via options or environment
+    variables:
+
+    \b
+      DOCUELEVATE_URL            Base URL of the instance (default: http://localhost:8000)
+      DOCUELEVATE_API_TOKEN      Personal API token (de_…)
+      DOCUELEVATE_TIMEOUT        HTTP request timeout in seconds (default: 60)
+
+    Examples:
+
+    \b
+      # Upload a file
+      docuelevate --token de_xxx upload report.pdf
+
+    \b
+      # List files as JSON for further processing
+      docuelevate --token de_xxx --format json list | jq '.[].original_filename'
+
+    \b
+      # Search for invoices
+      docuelevate --token de_xxx search "invoice amazon"
+    """
+    ctx.ensure_object(dict)
+    ctx.obj["url"] = url
+    ctx.obj["token"] = token
+    ctx.obj["fmt"] = fmt
+    ctx.obj["timeout"] = timeout
+
+
+def _get_token(ctx: click.Context) -> str:
+    """Return the token from context, raising ClickException if absent.
+
+    Args:
+        ctx: The current Click context.
+
+    Returns:
+        The API token string.
+
+    Raises:
+        click.ClickException: If no token has been provided.
+    """
+    token = ctx.obj.get("token")
+    if not token:
+        raise click.ClickException(f"No API token provided.  Use --token or set the {ENV_TOKEN} environment variable.")
+    return token
+
+
+# ---------------------------------------------------------------------------
+# list command
+# ---------------------------------------------------------------------------
+
+
+@cli.command("list")
+@click.option("--page", default=1, show_default=True, help="Page number.")
+@click.option("--per-page", default=25, show_default=True, help="Items per page (max 200).")
+@click.option("--search", default=None, help="Filter by filename substring.")
+@click.option("--mime-type", default=None, help="Filter by MIME type (e.g. application/pdf).")
+@click.option("--status", "file_status", default=None, help="Filter by status: pending, processing, completed, failed.")
+@click.option("--sort-by", default="created_at", show_default=True, help="Sort field.")
+@click.option("--sort-order", type=click.Choice(["asc", "desc"]), default="desc", show_default=True)
+@click.pass_context
+def list_files(
+    ctx: click.Context,
+    page: int,
+    per_page: int,
+    search: str | None,
+    mime_type: str | None,
+    file_status: str | None,
+    sort_by: str,
+    sort_order: str,
+) -> None:
+    """List documents stored in DocuElevate.
+
+    Examples:
+
+    \b
+      docuelevate list
+      docuelevate list --status completed --per-page 10
+      docuelevate --format json list | jq '.[].original_filename'
+    """
+    token = _get_token(ctx)
+    url: str = ctx.obj["url"]
+    fmt: str = ctx.obj["fmt"]
+    timeout: int = ctx.obj["timeout"]
+
+    params: dict[str, Any] = {
+        "page": page,
+        "per_page": per_page,
+        "sort_by": sort_by,
+        "sort_order": sort_order,
+    }
+    if search:
+        params["search"] = search
+    if mime_type:
+        params["mime_type"] = mime_type
+    if file_status:
+        params["status"] = file_status
+
+    resp = _api("GET", url, "/api/files", token, timeout=timeout, params=params)
+    payload = _require_ok(resp)
+
+    # Extract the list from the paginated response
+    files: list[dict[str, Any]] = payload.get("files", payload) if isinstance(payload, dict) else payload  # type: ignore[assignment]
+    pagination: dict[str, Any] = payload.get("pagination", {}) if isinstance(payload, dict) else {}
+
+    if fmt == "json":
+        _output(files, fmt)
+    else:
+        # Trim fields for readable table
+        rows = [
+            {
+                "id": f.get("id"),
+                "filename": f.get("original_filename"),
+                "size": f.get("file_size"),
+                "status": f.get("status"),
+                "created_at": str(f.get("created_at", ""))[:19],
+            }
+            for f in files
+        ]
+        _output(rows, fmt)
+        if pagination:
+            click.echo(f"\nPage {pagination.get('page')}/{pagination.get('pages')}  ({pagination.get('total')} total)")
+
+
+# ---------------------------------------------------------------------------
+# upload command
+# ---------------------------------------------------------------------------
+
+
+@cli.command("upload")
+@click.argument("files", nargs=-1, required=True, type=click.Path(exists=True, readable=True))
+@click.option(
+    "--batch-size",
+    default=5,
+    show_default=True,
+    help="Maximum number of concurrent uploads (sequential when 1).",
+)
+@click.pass_context
+def upload_files(ctx: click.Context, files: tuple[str, ...], batch_size: int) -> None:
+    """Upload one or more local files for processing.
+
+    Supports glob patterns and multiple arguments for batch uploads.
+
+    Examples:
+
+    \b
+      docuelevate upload report.pdf
+      docuelevate upload *.pdf invoice_*.png
+      docuelevate upload --batch-size 3 /scans/*.pdf
+    """
+    token = _get_token(ctx)
+    url: str = ctx.obj["url"]
+    fmt: str = ctx.obj["fmt"]
+    timeout: int = ctx.obj["timeout"]
+
+    results: list[dict[str, Any]] = []
+    failed = 0
+
+    for i, file_path in enumerate(files, 1):
+        path = Path(file_path)
+        click.echo(f"[{i}/{len(files)}] Uploading {path.name}…", err=True)
+        try:
+            with path.open("rb") as fh:
+                resp = _api(
+                    "POST",
+                    url,
+                    "/api/ui-upload",
+                    token,
+                    timeout=timeout,
+                    files={"file": (path.name, fh)},
+                )
+            if resp.status_code >= 400:
+                try:
+                    detail = resp.json().get("detail", resp.text)
+                except Exception:
+                    detail = resp.text
+                click.echo(f"  ERROR {resp.status_code}: {detail}", err=True)
+                results.append({"file": path.name, "status": "error", "detail": detail})
+                failed += 1
+            else:
+                data = resp.json()
+                results.append({"file": path.name, "status": "queued", **data})
+                click.echo(f"  OK  task_id={data.get('task_id', '?')}", err=True)
+        except click.ClickException:
+            raise
+        except Exception as exc:
+            click.echo(f"  ERROR: {exc}", err=True)
+            results.append({"file": path.name, "status": "error", "detail": str(exc)})
+            failed += 1
+
+    _output(results, fmt)
+
+    if failed:
+        click.echo(f"\n{failed}/{len(files)} upload(s) failed.", err=True)
+        sys.exit(1)
+
+
+# ---------------------------------------------------------------------------
+# download command
+# ---------------------------------------------------------------------------
+
+
+@cli.command("download")
+@click.argument("file_id", type=int)
+@click.option(
+    "--output",
+    "-o",
+    default=None,
+    help="Destination file path.  Defaults to the server-provided filename in the current directory.",
+    type=click.Path(),
+)
+@click.option(
+    "--version",
+    type=click.Choice(["processed", "original"]),
+    default="processed",
+    show_default=True,
+    help="Which version to download.",
+)
+@click.pass_context
+def download_file(ctx: click.Context, file_id: int, output: str | None, version: str) -> None:
+    """Download a file by its numeric ID.
+
+    Examples:
+
+    \b
+      docuelevate download 42
+      docuelevate download 42 --version original -o /tmp/orig.pdf
+    """
+    token = _get_token(ctx)
+    url: str = ctx.obj["url"]
+    timeout: int = ctx.obj["timeout"]
+
+    resp = _api(
+        "GET",
+        url,
+        f"/api/files/{file_id}/download",
+        token,
+        timeout=timeout,
+        params={"version": version},
+        stream=True,
+    )
+    _require_ok(resp)
+
+    # Determine output filename
+    if output:
+        dest = Path(output)
+    else:
+        content_disp = resp.headers.get("content-disposition", "")
+        filename = f"file_{file_id}"
+        for raw_part in content_disp.split(";"):
+            clean = raw_part.strip()
+            if clean.startswith("filename="):
+                filename = clean[len("filename=") :].strip('"').strip("'")
+                break
+            if clean.startswith("filename*="):
+                raw = clean[len("filename*=") :]
+                if raw.upper().startswith("UTF-8''"):
+                    filename = unquote(raw[7:])
+                    break
+        dest = Path(filename)
+
+    with dest.open("wb") as fh:
+        for chunk in resp.iter_content(chunk_size=65536):
+            fh.write(chunk)
+
+    click.echo(f"Downloaded {dest} ({dest.stat().st_size} bytes)")
+
+
+# ---------------------------------------------------------------------------
+# search command
+# ---------------------------------------------------------------------------
+
+
+@cli.command("search")
+@click.argument("query")
+@click.option("--mime-type", default=None, help="Filter by MIME type.")
+@click.option("--document-type", default=None, help="Filter by document type (e.g. Invoice).")
+@click.option("--tags", default=None, help="Filter by tag.")
+@click.option("--language", default=None, help="Filter by language code (e.g. en, de).")
+@click.option("--page", default=1, show_default=True)
+@click.option("--per-page", default=20, show_default=True, help="Results per page (max 100).")
+@click.pass_context
+def search(
+    ctx: click.Context,
+    query: str,
+    mime_type: str | None,
+    document_type: str | None,
+    tags: str | None,
+    language: str | None,
+    page: int,
+    per_page: int,
+) -> None:
+    """Full-text search across all documents.
+
+    Examples:
+
+    \b
+      docuelevate search "invoice amazon"
+      docuelevate search "contract" --document-type Contract --language en
+      docuelevate --format json search "receipt" | jq '.[].file_id'
+    """
+    token = _get_token(ctx)
+    url: str = ctx.obj["url"]
+    fmt: str = ctx.obj["fmt"]
+    timeout: int = ctx.obj["timeout"]
+
+    params: dict[str, Any] = {"q": query, "page": page, "per_page": per_page}
+    if mime_type:
+        params["mime_type"] = mime_type
+    if document_type:
+        params["document_type"] = document_type
+    if tags:
+        params["tags"] = tags
+    if language:
+        params["language"] = language
+
+    resp = _api("GET", url, "/api/search", token, timeout=timeout, params=params)
+    payload = _require_ok(resp)
+
+    results: list[dict[str, Any]] = (
+        payload.get("results", payload) if isinstance(payload, dict) else payload  # type: ignore[assignment]
+    )
+    total: int = payload.get("total", len(results)) if isinstance(payload, dict) else len(results)
+    pages: int = payload.get("pages", 1) if isinstance(payload, dict) else 1
+
+    if fmt == "json":
+        _output(results, fmt)
+    else:
+        rows = [
+            {
+                "file_id": r.get("file_id"),
+                "filename": r.get("original_filename"),
+                "type": r.get("document_type"),
+                "tags": ",".join(r.get("tags") or []),
+            }
+            for r in results
+        ]
+        _output(rows, fmt)
+        click.echo(f"\nPage {page}/{pages}  ({total} total results)")
+
+
+# ---------------------------------------------------------------------------
+# token sub-group
+# ---------------------------------------------------------------------------
+
+
+@cli.group("token")
+@click.pass_context
+def token_group(ctx: click.Context) -> None:
+    """Manage personal API tokens.
+
+    Tokens can be created, listed, and revoked.  Token rotation is achieved
+    by creating a new token before revoking the old one.
+
+    Examples:
+
+    \b
+      docuelevate token create "CI Pipeline"
+      docuelevate token list
+      docuelevate token revoke 3
+    """
+
+
+@token_group.command("create")
+@click.argument("name")
+@click.pass_context
+def token_create(ctx: click.Context, name: str) -> None:
+    """Create a new personal API token.
+
+    The full token value is printed exactly once.  Store it securely.
+
+    Examples:
+
+    \b
+      docuelevate token create "My script"
+      docuelevate --format json token create "CI" | jq -r '.token'
+    """
+    token = _get_token(ctx)
+    url: str = ctx.obj["url"]
+    fmt: str = ctx.obj["fmt"]
+    timeout: int = ctx.obj["timeout"]
+
+    resp = _api("POST", url, "/api/api-tokens/", token, timeout=timeout, json={"name": name})
+    payload = _require_ok(resp)
+
+    if fmt == "json":
+        _output(payload, fmt)
+    else:
+        if not isinstance(payload, dict):
+            raise click.ClickException("Unexpected API response format.")
+        click.echo("Token created successfully:")
+        click.echo(f"  ID:     {payload.get('id')}")
+        click.echo(f"  Name:   {payload.get('name')}")
+        click.echo(f"  Prefix: {payload.get('token_prefix')}")
+        click.echo(f"  Token:  {payload.get('token')}")
+        click.echo()
+        click.echo("Store this token securely — it will not be shown again.", err=True)
+
+
+@token_group.command("list")
+@click.pass_context
+def token_list(ctx: click.Context) -> None:
+    """List all your API tokens (active and revoked).
+
+    Examples:
+
+    \b
+      docuelevate token list
+      docuelevate --format json token list | jq '.[] | select(.is_active)'
+    """
+    token = _get_token(ctx)
+    url: str = ctx.obj["url"]
+    fmt: str = ctx.obj["fmt"]
+    timeout: int = ctx.obj["timeout"]
+
+    resp = _api("GET", url, "/api/api-tokens/", token, timeout=timeout)
+    payload = _require_ok(resp)
+
+    if fmt == "json":
+        _output(payload, fmt)
+    else:
+        if not isinstance(payload, list):
+            raise click.ClickException("Unexpected API response format.")
+        rows = [
+            {
+                "id": t.get("id"),
+                "name": t.get("name"),
+                "prefix": t.get("token_prefix"),
+                "active": t.get("is_active"),
+                "last_used": str(t.get("last_used_at") or "never")[:19],
+                "created": str(t.get("created_at") or "")[:19],
+            }
+            for t in payload
+        ]
+        _output(rows, fmt)
+
+
+@token_group.command("revoke")
+@click.argument("token_id", type=int)
+@click.option("--yes", "-y", is_flag=True, help="Skip confirmation prompt.")
+@click.pass_context
+def token_revoke(ctx: click.Context, token_id: int, yes: bool) -> None:
+    """Revoke an API token by its numeric ID.
+
+    The token is soft-deleted (kept for audit) but immediately invalidated.
+
+    Examples:
+
+    \b
+      docuelevate token revoke 3
+      docuelevate token revoke 3 --yes
+    """
+    token = _get_token(ctx)
+    url: str = ctx.obj["url"]
+    timeout: int = ctx.obj["timeout"]
+
+    if not yes:
+        click.confirm(f"Revoke token {token_id}?", abort=True)
+
+    resp = _api("DELETE", url, f"/api/api-tokens/{token_id}", token, timeout=timeout)
+    _require_ok(resp)
+    click.echo(f"Token {token_id} revoked.")
+
+
+# ---------------------------------------------------------------------------
+# Entry point
+# ---------------------------------------------------------------------------
+
+
+def main() -> None:
+    """Entry point for the ``docuelevate`` console script."""
+    cli(auto_envvar_prefix="DOCUELEVATE")  # type: ignore[call-arg]
+
+
+if __name__ == "__main__":
+    main()

app/models.py

@@ -684,3 +684,48 @@ class SharedLink(Base):
 
     created_at = Column(DateTime(timezone=True), server_default=func.now())
     revoked_at = Column(DateTime(timezone=True), nullable=True)
+class UserNotificationTarget(Base):
+    """Per-user notification target (email or webhook channel)."""
+
+    __tablename__ = "user_notification_targets"
+
+    id = Column(Integer, primary_key=True, index=True)
+    owner_id = Column(String, nullable=False, index=True)
+    channel_type = Column(String(20), nullable=False)  # "email" or "webhook"
+    name = Column(String(255), nullable=False)  # Human-readable label
+    config = Column(Text, nullable=True)  # JSON: smtp config or webhook url
+    is_active = Column(Boolean, nullable=False, default=True)
+    created_at = Column(DateTime(timezone=True), server_default=func.now())
+    updated_at = Column(DateTime(timezone=True), server_default=func.now(), onupdate=func.now())
+
+
+class UserNotificationPreference(Base):
+    """Mapping: which user events trigger which notification channel."""
+
+    __tablename__ = "user_notification_preferences"
+
+    id = Column(Integer, primary_key=True, index=True)
+    owner_id = Column(String, nullable=False, index=True)
+    event_type = Column(String(50), nullable=False)  # "document.processed", "document.failed"
+    channel_type = Column(String(20), nullable=False)  # "in_app", "email", "webhook"
+    target_id = Column(Integer, nullable=True)  # NULL = in_app, else UserNotificationTarget.id
+    is_enabled = Column(Boolean, nullable=False, default=True)
+
+    __table_args__ = (UniqueConstraint("owner_id", "event_type", "channel_type", "target_id"),)
+    created_at = Column(DateTime(timezone=True), server_default=func.now())
+    updated_at = Column(DateTime(timezone=True), server_default=func.now(), onupdate=func.now())
+
+
+class InAppNotification(Base):
+    """In-app notification record for the bell icon / inbox."""
+
+    __tablename__ = "in_app_notifications"
+
+    id = Column(Integer, primary_key=True, index=True)
+    owner_id = Column(String, nullable=False, index=True)
+    event_type = Column(String(50), nullable=False)  # "document.processed", "document.failed"
+    title = Column(String(255), nullable=False)
+    message = Column(Text, nullable=True)
+    is_read = Column(Boolean, nullable=False, default=False, index=True)
+    file_id = Column(Integer, nullable=True)  # Optional link to FileRecord
+    created_at = Column(DateTime(timezone=True), server_default=func.now(), index=True)

app/tasks/process_document.py

@@ -1,10 +1,14 @@
 #!/usr/bin/env python3
 
+from __future__ import annotations
+
+import json
 import logging
 import mimetypes
 import os
 import shutil
 import uuid
+from typing import TYPE_CHECKING
 
 import pypdf  # Upgraded from PyPDF2 to fix CVE-2023-36464
 from pypdf.errors import PdfReadError
@@ -12,7 +16,7 @@ from pypdf.errors import PdfReadError
 from app.celery_app import celery
 from app.config import settings
 from app.database import SessionLocal
-from app.models import FileRecord
+from app.models import FileRecord, Pipeline, PipelineStep
 from app.tasks.extract_metadata_with_gpt import extract_metadata_with_gpt
 from app.tasks.process_with_ocr import process_with_ocr
 from app.tasks.retry_config import BaseTaskWithRetry
@@ -20,9 +24,75 @@ from app.utils import get_unique_filepath_with_counter, hash_file, log_task_prog
 from app.utils.step_manager import initialize_file_steps
 from app.utils.text_quality import check_text_quality, detect_pdf_text_source
 
+if TYPE_CHECKING:
+    from sqlalchemy.orm import Session
+
 logger = logging.getLogger(__name__)
 
 
+def _get_pipeline_ocr_language(db: "Session", file_record: FileRecord, owner_id: str | None) -> str | None:
+    """Look up the OCR language override from the file's pipeline OCR step config.
+
+    Resolution order:
+    1. Explicit pipeline assigned to the file (``file_record.pipeline_id``).
+    2. User's own default pipeline (``owner_id``, ``is_default=True``).
+    3. System default pipeline (``owner_id=NULL``, ``is_default=True``).
+
+    Returns the ``ocr_language`` value from the pipeline's OCR step config, or
+    ``None`` when no override is configured.
+    """
+    pipeline = None
+
+    if file_record.pipeline_id:
+        pipeline = db.query(Pipeline).filter(Pipeline.id == file_record.pipeline_id).first()
+
+    if pipeline is None and owner_id:
+        pipeline = (
+            db.query(Pipeline)
+            .filter(
+                Pipeline.owner_id == owner_id,
+                Pipeline.is_default.is_(True),
+                Pipeline.is_active.is_(True),
+            )
+            .first()
+        )
+
+    if pipeline is None:
+        pipeline = (
+            db.query(Pipeline)
+            .filter(
+                Pipeline.owner_id.is_(None),
+                Pipeline.is_default.is_(True),
+                Pipeline.is_active.is_(True),
+            )
+            .first()
+        )
+
+    if pipeline is None:
+        return None
+
+    ocr_step = (
+        db.query(PipelineStep)
+        .filter(
+            PipelineStep.pipeline_id == pipeline.id,
+            PipelineStep.step_type == "ocr",
+            PipelineStep.enabled.is_(True),
+        )
+        .first()
+    )
+
+    if ocr_step is None or not ocr_step.config:
+        return None
+
+    try:
+        step_config = json.loads(ocr_step.config)
+        lang = step_config.get("ocr_language")
+        # "auto" is treated as no override
+        return lang if lang and lang != "auto" else None
+    except Exception:
+        return None
+
+
 @celery.task(base=BaseTaskWithRetry, bind=True)
 def process_document(
     self,
@@ -109,6 +179,7 @@ def process_document(
         )
 
     # Acquire DB session in the task
+    ocr_language: str | None = None  # Pipeline OCR language override resolved inside DB session
     with SessionLocal() as db:
         # When file_id is provided, we are reprocessing an existing file.
         # Skip the duplicate check and reuse the existing record.
@@ -305,6 +376,14 @@ def process_document(
         new_record.local_filename = new_local_path
         db.commit()
 
+        # Look up pipeline OCR language override before the session closes.
+        # This reads the OCR step config from the file's assigned pipeline (or
+        # the user/system default pipeline) so the language is available when
+        # dispatching process_with_ocr below.
+        ocr_language = _get_pipeline_ocr_language(db, new_record, owner_id)
+        if ocr_language:
+            logger.info(f"[{task_id}] Pipeline OCR language override: {ocr_language!r}")
+
         # Store file_id before session closes to avoid DetachedInstanceError
         file_id = new_record.id
 
@@ -334,7 +413,7 @@ def process_document(
             "Queued for forced OCR processing",
             file_id=file_id,
         )
-        process_with_ocr.delay(new_filename, file_id)
+        process_with_ocr.delay(new_filename, file_id, language=ocr_language)
         return {"file": new_local_path, "status": "Queued for forced OCR", "file_id": file_id}
 
     # If the file is not a PDF, skip embedded text check and convert to PDF first
@@ -491,7 +570,7 @@ def process_document(
                     "Queued for OCR (text quality too low)",
                     file_id=file_id,
                 )
-                process_with_ocr.delay(new_filename, file_id, extracted_text)
+                process_with_ocr.delay(new_filename, file_id, extracted_text, language=ocr_language)
                 return {
                     "file": new_local_path,
                     "status": "Queued for OCR (poor embedded text quality)",
@@ -564,5 +643,5 @@ def process_document(
         "Queued for OCR processing",
         file_id=file_id,
     )
-    process_with_ocr.delay(new_filename, file_id)
+    process_with_ocr.delay(new_filename, file_id, language=ocr_language)
     return {"file": new_local_path, "status": "Queued for OCR", "file_id": file_id}

app/tasks/process_with_ocr.py

@@ -17,7 +17,6 @@ task with a multi-engine OCR pipeline that:
 
 import logging
 import os
-from typing import Optional
 
 from app.celery_app import celery
 from app.config import settings
@@ -33,7 +32,13 @@ logger = logging.getLogger(__name__)
 
 
 @celery.task(base=OcrTaskWithRetry, bind=True)
-def process_with_ocr(self, filename: str, file_id: Optional[int] = None, original_text: Optional[str] = None):
+def process_with_ocr(
+    self,
+    filename: str,
+    file_id: int | None = None,
+    original_text: str | None = None,
+    language: str | None = None,
+):
     """Run the configured OCR providers on *filename* and continue the pipeline.
 
     When multiple OCR providers are configured the results are merged using the
@@ -47,6 +52,10 @@ def process_with_ocr(self, filename: str, file_id: Optional[int] = None, origina
         filename: Base name of the file inside ``<workdir>/tmp/``.
         file_id: Optional database record ID passed through to downstream tasks.
         original_text: Optional original embedded text for head-to-head comparison.
+        language: Optional Tesseract-style language code(s) (e.g. ``"eng+deu"``)
+            to override the global OCR language settings for this specific run.
+            Pass ``None`` or ``"auto"`` to use the global settings.  This
+            enables per-pipeline language configuration.
     """
     task_id = self.request.id
     log_task_progress(
@@ -62,7 +71,7 @@ def process_with_ocr(self, filename: str, file_id: Optional[int] = None, origina
         if not os.path.exists(tmp_file_path):
             raise FileNotFoundError(f"Local file not found: {tmp_file_path}")
 
-        providers = get_ocr_providers()
+        providers = get_ocr_providers(language=language)
         provider_names = [p.name for p in providers]
         logger.info(f"[{task_id}] Running {len(providers)} OCR provider(s): {provider_names}")
 
@@ -122,7 +131,12 @@ def process_with_ocr(self, filename: str, file_id: Optional[int] = None, origina
         # PDF with ocrmypdf to embed an invisible text layer so the output is
         # selectable/searchable in PDF viewers.
         if searchable_pdf_path is None:
-            lang = getattr(settings, "tesseract_language", None) or "eng"
+            # Use the per-call language override; fall back to global setting
+            embed_lang = (
+                language
+                if language and language != "auto"
+                else (getattr(settings, "tesseract_language", None) or "eng")
+            )
             log_task_progress(
                 task_id,
                 "embed_text_layer",
@@ -130,7 +144,7 @@ def process_with_ocr(self, filename: str, file_id: Optional[int] = None, origina
                 "Embedding searchable text layer into PDF",
                 file_id=file_id,
             )
-            embedded = embed_text_layer(tmp_file_path, tmp_file_path, language=lang)
+            embedded = embed_text_layer(tmp_file_path, tmp_file_path, language=embed_lang)
             if embedded:
                 searchable_pdf_path = tmp_file_path
                 log_task_progress(

app/utils/ocr_provider.py

@@ -192,6 +192,95 @@ class OCRResult:
         )
 
 
+# ---------------------------------------------------------------------------
+# Multi-language support
+# ---------------------------------------------------------------------------
+
+#: Canonical list of supported OCR languages for pipeline configuration.
+#: Keys are display names; values are Tesseract language code(s).
+#: Tesseract codes are used as the canonical format because they are the most
+#: widely applicable across self-hosted providers (Tesseract + ocrmypdf).
+#: "auto" falls back to the global ``tesseract_language`` / ``easyocr_languages``
+#: settings (i.e. no per-call override).
+OCR_LANGUAGES: Dict[str, str] = {
+    "Auto (use system default)": "auto",
+    "Arabic": "ara",
+    "Chinese (Simplified)": "chi_sim",
+    "Chinese (Traditional)": "chi_tra",
+    "Czech": "ces",
+    "Danish": "dan",
+    "Dutch": "nld",
+    "English": "eng",
+    "Finnish": "fin",
+    "French": "fra",
+    "German": "deu",
+    "Greek": "ell",
+    "Hebrew": "heb",
+    "Hindi": "hin",
+    "Hungarian": "hun",
+    "Italian": "ita",
+    "Japanese": "jpn",
+    "Korean": "kor",
+    "Norwegian": "nor",
+    "Polish": "pol",
+    "Portuguese": "por",
+    "Romanian": "ron",
+    "Russian": "rus",
+    "Spanish": "spa",
+    "Swedish": "swe",
+    "Thai": "tha",
+    "Turkish": "tur",
+    "Ukrainian": "ukr",
+    "Vietnamese": "vie",
+}
+
+#: Mapping from Tesseract language codes to EasyOCR language codes.
+#: Used when ``TesseractOCRProvider``-style codes are specified but EasyOCR is
+#: the active provider.  Codes not present in this map are passed through as-is
+#: (EasyOCR accepts its own ISO 639-1 codes such as ``"en"`` or ``"de"``).
+TESSERACT_TO_EASYOCR: Dict[str, str] = {
+    "ara": "ar",
+    "ces": "cs",
+    "chi_sim": "ch_sim",
+    "chi_tra": "ch_tra",
+    "dan": "da",
+    "deu": "de",
+    "ell": "el",
+    "eng": "en",
+    "fin": "fi",
+    "fra": "fr",
+    "heb": "he",
+    "hin": "hi",
+    "hun": "hu",
+    "ita": "it",
+    "jpn": "ja",
+    "kor": "ko",
+    "nld": "nl",
+    "nor": "no",
+    "pol": "pl",
+    "por": "pt",
+    "ron": "ro",
+    "rus": "ru",
+    "spa": "es",
+    "swe": "sv",
+    "tha": "th",
+    "tur": "tr",
+    "ukr": "uk",
+    "vie": "vi",
+}
+
+
+def _tesseract_codes_to_easyocr(tesseract_lang: str) -> List[str]:
+    """Convert a Tesseract language string (e.g. ``"eng+deu"``) to a list of
+    EasyOCR language codes (e.g. ``["en", "de"]``).
+
+    Unknown codes are passed through unchanged, so native EasyOCR codes such
+    as ``"en"`` also work transparently.
+    """
+    codes = [part.strip() for part in tesseract_lang.split("+") if part.strip()]
+    return [TESSERACT_TO_EASYOCR.get(code, code) for code in codes]
+
+
 class OCRProvider(ABC):
     """Abstract base class for OCR providers.
 
@@ -290,10 +379,24 @@ class TesseractOCRProvider(OCRProvider):
     - ``tesseract_cmd`` – path to the ``tesseract`` binary (optional).
     - ``tesseract_language`` – Tesseract language code(s), e.g. ``"eng"`` or
       ``"eng+deu"`` (default: ``"eng"``).
+
+    The optional *language* constructor argument overrides the global
+    ``tesseract_language`` setting for this specific provider instance, enabling
+    per-pipeline language configuration.
     """
 
     name = "tesseract"
 
+    def __init__(self, language: Optional[str] = None) -> None:
+        """Initialise the Tesseract provider.
+
+        Args:
+            language: Optional Tesseract language code(s) to use instead of the
+                global ``tesseract_language`` setting (e.g. ``"eng+deu"``).
+                Pass ``None`` or ``"auto"`` to use the global setting.
+        """
+        self._language_override: Optional[str] = language if language and language != "auto" else None
+
     def process(self, file_path: str) -> OCRResult:
         try:
             import pytesseract
@@ -308,7 +411,7 @@ class TesseractOCRProvider(OCRProvider):
         if tesseract_cmd:
             pytesseract.pytesseract.tesseract_cmd = tesseract_cmd
 
-        lang = getattr(settings, "tesseract_language", None) or "eng"
+        lang = self._language_override or getattr(settings, "tesseract_language", None) or "eng"
 
         # Ensure language data files are present; attempt download if missing.
         from app.utils.ocr_language_manager import ensure_tesseract_languages  # noqa: PLC0415
@@ -349,10 +452,26 @@ class EasyOCRProvider(OCRProvider):
     - ``easyocr_languages`` – comma-separated list of language codes
       (default: ``"en"``).
     - ``easyocr_gpu`` – whether to use GPU acceleration (default: ``False``).
+
+    The optional *language* constructor argument accepts a Tesseract-style
+    language string (e.g. ``"eng+deu"``) which is automatically translated to
+    EasyOCR codes (e.g. ``["en", "de"]``), overriding the global
+    ``easyocr_languages`` setting for this provider instance.
     """
 
     name = "easyocr"
 
+    def __init__(self, language: Optional[str] = None) -> None:
+        """Initialise the EasyOCR provider.
+
+        Args:
+            language: Optional Tesseract-style language code(s) (e.g. ``"eng+deu"``)
+                or a comma-separated EasyOCR language list (e.g. ``"en,de"``).
+                Pass ``None`` or ``"auto"`` to use the global ``easyocr_languages``
+                setting.
+        """
+        self._language_override: Optional[str] = language if language and language != "auto" else None
+
     def process(self, file_path: str) -> OCRResult:
         try:
             import easyocr
@@ -363,8 +482,12 @@ class EasyOCRProvider(OCRProvider):
                 "Install them with: pip install easyocr pdf2image"
             ) from exc
 
-        lang_str = getattr(settings, "easyocr_languages", None) or "en"
-        langs = [lang.strip() for lang in lang_str.split(",") if lang.strip()]
+        if self._language_override:
+            # Convert Tesseract-style codes to EasyOCR codes
+            langs = _tesseract_codes_to_easyocr(self._language_override)
+        else:
+            lang_str = getattr(settings, "easyocr_languages", None) or "en"
+            langs = [lang.strip() for lang in lang_str.split(",") if lang.strip()]
         gpu = getattr(settings, "easyocr_gpu", False)
 
         logger.info(f"[EasyOCR] Processing {os.path.basename(file_path)} (langs={langs}, gpu={gpu})")
@@ -679,23 +802,36 @@ KNOWN_OCR_PROVIDERS: List[str] = sorted(_PROVIDER_MAP.keys())
 MAX_OCR_TEXT_FOR_AI_MERGE = 4000
 
 
-def get_ocr_providers() -> List[OCRProvider]:
+def get_ocr_providers(language: Optional[str] = None) -> List[OCRProvider]:
     """Return a list of configured OCR provider instances.
 
     Reads ``settings.ocr_providers`` (comma-separated provider names) and
     returns one instantiated provider per entry.  Falls back to ``["azure"]``
     when the setting is absent.
+
+    Args:
+        language: Optional Tesseract-style language code(s) (e.g. ``"eng+deu"``)
+            to override the global language settings for providers that support
+            per-call language configuration (Tesseract and EasyOCR).  Pass
+            ``None`` or ``"auto"`` to use the global settings.
     """
     raw = getattr(settings, "ocr_providers", None) or "azure"
     provider_names = [name.strip().lower() for name in raw.split(",") if name.strip()]
 
+    # Normalise "auto" to None so providers fall back to global settings
+    effective_language = language if language and language != "auto" else None
+
     providers: List[OCRProvider] = []
     for name in provider_names:
         cls = _PROVIDER_MAP.get(name)
         if cls is None:
             logger.warning(f"Unknown OCR provider '{name}' in OCR_PROVIDERS – skipping.")
             continue
-        providers.append(cls())
+        # Pass language override to providers that support per-call language config
+        if effective_language is not None and name in ("tesseract", "easyocr"):
+            providers.append(cls(language=effective_language))
+        else:
+            providers.append(cls())
         logger.debug(f"Registered OCR provider: {name}")
 
     if not providers:

app/utils/user_notification.py

@@ -0,0 +1,233 @@
+"""Per-user notification dispatch service.
+
+Handles user-centric events (document.processed, document.failed) by:
+1. Always creating an InAppNotification record
+2. Sending via configured email/webhook targets (UserNotificationTarget)
+   if the user has enabled that channel/event combination.
+"""
+
+import json
+import logging
+import smtplib
+from email.mime.multipart import MIMEMultipart
+from email.mime.text import MIMEText
+from typing import Any
+
+import httpx
+
+from app.database import SessionLocal
+from app.models import InAppNotification, UserNotificationPreference, UserNotificationTarget
+
+logger = logging.getLogger(__name__)
+
+# Supported user-centric event types
+EVENT_DOCUMENT_PROCESSED = "document.processed"
+EVENT_DOCUMENT_FAILED = "document.failed"
+
+USER_EVENT_LABELS: dict[str, str] = {
+    EVENT_DOCUMENT_PROCESSED: "Document Processed",
+    EVENT_DOCUMENT_FAILED: "Document Processing Failed",
+}
+
+
+def create_in_app_notification(
+    owner_id: str,
+    event_type: str,
+    title: str,
+    message: str,
+    file_id: int | None = None,
+) -> InAppNotification | None:
+    """Persist an InAppNotification record for the given user.
+
+    Returns:
+        The created InAppNotification, or None on error.
+    """
+    db = SessionLocal()
+    try:
+        notif = InAppNotification(
+            owner_id=owner_id,
+            event_type=event_type,
+            title=title,
+            message=message,
+            file_id=file_id,
+        )
+        db.add(notif)
+        db.commit()
+        db.refresh(notif)
+        return notif
+    except Exception:
+        db.rollback()
+        logger.exception("Failed to create in-app notification for owner_id=%s", owner_id)
+        return None
+    finally:
+        db.close()
+
+
+def _send_email_notification(target_config: dict[str, Any], title: str, message: str) -> bool:
+    """Send an email notification via the configured SMTP target.
+
+    Args:
+        target_config: dict with keys: smtp_host, smtp_port, smtp_username,
+                       smtp_password, smtp_use_tls, recipient_email
+        title: Email subject
+        message: Email body text
+
+    Returns:
+        True if the email was sent successfully, False otherwise.
+    """
+    try:
+        smtp_host = target_config.get("smtp_host", "")
+        smtp_port = int(target_config.get("smtp_port", 587))
+        smtp_username = target_config.get("smtp_username", "")
+        smtp_password = target_config.get("smtp_password", "")
+        smtp_use_tls = bool(target_config.get("smtp_use_tls", True))
+        recipient_email = target_config.get("recipient_email", "")
+        sender_email = target_config.get("sender_email") or smtp_username or "noreply@docuelevate.local"
+
+        if not smtp_host or not recipient_email:
+            logger.warning("Email notification target missing smtp_host or recipient_email")
+            return False
+
+        msg = MIMEMultipart("alternative")
+        msg["Subject"] = title
+        msg["From"] = sender_email
+        msg["To"] = recipient_email
+        msg.attach(MIMEText(message, "plain"))
+
+        with smtplib.SMTP(smtp_host, smtp_port, timeout=30) as server:
+            if smtp_use_tls:
+                server.starttls()
+            if smtp_username and smtp_password:
+                server.login(smtp_username, smtp_password)
+            server.send_message(msg)
+
+        logger.info("Email notification sent to %s", recipient_email)
+        return True
+    except Exception:
+        logger.exception("Failed to send email notification")
+        return False
+
+
+def _send_webhook_notification(target_config: dict[str, Any], event_type: str, title: str, message: str) -> bool:
+    """Send a webhook POST notification to the configured URL.
+
+    Args:
+        target_config: dict with keys: url, secret (optional HMAC header value)
+        event_type: The event type string
+        title: Notification title
+        message: Notification body
+
+    Returns:
+        True if the webhook was delivered successfully, False otherwise.
+    """
+    try:
+        url = target_config.get("url", "")
+        secret = target_config.get("secret", "")
+
+        if not url:
+            logger.warning("Webhook notification target missing url")
+            return False
+
+        payload = {
+            "event": event_type,
+            "title": title,
+            "message": message,
+        }
+        headers = {"Content-Type": "application/json"}
+        if secret:
+            headers["X-DocuElevate-Secret"] = secret
+
+        response = httpx.post(url, json=payload, headers=headers, timeout=10)
+        response.raise_for_status()
+        logger.info("Webhook notification sent to %s (status %s)", url, response.status_code)
+        return True
+    except Exception:
+        logger.exception("Failed to send webhook notification to %s", target_config.get("url", ""))
+        return False
+
+
+def dispatch_user_notification(
+    owner_id: str,
+    event_type: str,
+    title: str,
+    message: str,
+    file_id: int | None = None,
+) -> None:
+    """Dispatch a user notification for the given event.
+
+    Always creates an in-app notification. Also sends via email/webhook
+    targets if the user has configured and enabled them for this event.
+
+    Args:
+        owner_id: The user's stable identifier.
+        event_type: e.g. "document.processed" or "document.failed"
+        title: Short notification title.
+        message: Longer notification body.
+        file_id: Optional FileRecord.id to link.
+    """
+    # 1. Always create an in-app notification
+    create_in_app_notification(
+        owner_id=owner_id,
+        event_type=event_type,
+        title=title,
+        message=message,
+        file_id=file_id,
+    )
+
+    # 2. Check for configured email/webhook preferences
+    db = SessionLocal()
+    try:
+        prefs = (
+            db.query(UserNotificationPreference)
+            .filter(
+                UserNotificationPreference.owner_id == owner_id,
+                UserNotificationPreference.event_type == event_type,
+                UserNotificationPreference.is_enabled == True,  # noqa: E712
+                UserNotificationPreference.channel_type.in_(["email", "webhook"]),
+            )
+            .all()
+        )
+
+        for pref in prefs:
+            if not pref.target_id:
+                continue
+            target = db.get(UserNotificationTarget, pref.target_id)
+            if not target or not target.is_active:
+                continue
+            config: dict[str, Any] = {}
+            if target.config:
+                try:
+                    config = json.loads(target.config)
+                except (json.JSONDecodeError, ValueError):
+                    config = {}
+
+            if pref.channel_type == "email":
+                _send_email_notification(config, title, message)
+            elif pref.channel_type == "webhook":
+                _send_webhook_notification(config, event_type, title, message)
+    except Exception:
+        logger.exception("Error dispatching user notification for owner_id=%s event=%s", owner_id, event_type)
+    finally:
+        db.close()
+
+
+def notify_user_document_processed(owner_id: str, filename: str, file_id: int | None = None) -> None:
+    """Notify a user that their document was successfully processed."""
+    dispatch_user_notification(
+        owner_id=owner_id,
+        event_type=EVENT_DOCUMENT_PROCESSED,
+        title=f"Document processed: {filename}",
+        message=f"Your document '{filename}' has been successfully processed and uploaded.",
+        file_id=file_id,
+    )
+
+
+def notify_user_document_failed(owner_id: str, filename: str, error: str, file_id: int | None = None) -> None:
+    """Notify a user that their document processing failed."""
+    dispatch_user_notification(
+        owner_id=owner_id,
+        event_type=EVENT_DOCUMENT_FAILED,
+        title=f"Document processing failed: {filename}",
+        message=f"Processing of '{filename}' failed: {error}",
+        file_id=file_id,
+    )

app/views/__init__.py

@@ -18,6 +18,7 @@ from app.views.help import router as help_router  # Built-in help / How-To docs
 from app.views.imap_accounts import router as imap_accounts_router
 from app.views.integrations import router as integrations_router  # Unified integrations dashboard
 from app.views.license_routes import router as license_router  # Add the license router
+from app.views.notifications import router as notifications_router
 from app.views.onboarding import router as onboarding_router
 from app.views.onedrive import router as onedrive_router
 from app.views.pipelines import router as pipelines_router  # Processing pipelines
@@ -56,4 +57,5 @@ router.include_router(onboarding_router)  # User onboarding wizard
 router.include_router(pipelines_router)  # Processing pipelines
 router.include_router(imap_accounts_router)  # Per-user IMAP ingestion accounts
 router.include_router(integrations_router)  # Unified integrations dashboard
+router.include_router(notifications_router)  # User notification dashboard
 router.include_router(help_router)  # Built-in help / How-To docs

app/views/notifications.py

@@ -0,0 +1,20 @@
+"""View route for the notifications dashboard."""
+
+import logging
+
+from fastapi import Request
+
+from app.views.base import APIRouter, require_login, templates
+
+logger = logging.getLogger(__name__)
+router = APIRouter()
+
+
+@router.get("/notifications")
+@require_login
+async def notifications_dashboard(request: Request):
+    """Render the notifications dashboard."""
+    return templates.TemplateResponse(
+        "notifications_dashboard.html",
+        {"request": request, "page_title": "Notifications"},
+    )