gh-christianlouis-docuelevate/app/tasks/send_to_all.py

#!/usr/bin/env python3

import logging
import os

from app.celery_app import celery
from app.config import settings
from app.database import SessionLocal
from app.models import FileRecord
from app.tasks.retry_config import BaseTaskWithRetry
from app.tasks.upload_to_dropbox import upload_to_dropbox
from app.tasks.upload_to_email import upload_to_email
from app.tasks.upload_to_ftp import upload_to_ftp
from app.tasks.upload_to_google_drive import upload_to_google_drive
from app.tasks.upload_to_nextcloud import upload_to_nextcloud
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_webdav import upload_to_webdav
from app.utils.config_validator import get_provider_status
from app.utils.logging import log_task_progress

logger = logging.getLogger(__name__)


def _should_upload_to_dropbox():
    return bool(settings.dropbox_app_key and settings.dropbox_app_secret and settings.dropbox_refresh_token)


def _should_upload_to_nextcloud():
    return bool(settings.nextcloud_upload_url and settings.nextcloud_username and settings.nextcloud_password)


def _should_upload_to_paperless():
    return bool(settings.paperless_ngx_api_token and settings.paperless_host)


def _should_upload_to_google_drive():
    # Check for OAuth configuration
    if getattr(settings, "google_drive_use_oauth", False):
        return bool(
            settings.google_drive_client_id
            and settings.google_drive_client_secret
            and settings.google_drive_refresh_token
            and settings.google_drive_folder_id
        )
    # Or check for service account configuration
    else:
        return bool(settings.google_drive_credentials_json and settings.google_drive_folder_id)


def _should_upload_to_webdav():
    return bool(settings.webdav_url and settings.webdav_username and settings.webdav_password)


def _should_upload_to_ftp():
    return bool(settings.ftp_host and settings.ftp_username and settings.ftp_password)


def _should_upload_to_sftp():
    return bool(settings.sftp_host and settings.sftp_username and (settings.sftp_password or settings.sftp_private_key))


def _should_upload_to_email():
    return bool(
        settings.dest_email_host
        and settings.dest_email_username
        and settings.dest_email_password
        and settings.dest_email_default_recipient
    )


def _should_upload_to_onedrive():
    return bool(settings.onedrive_client_id and settings.onedrive_client_secret and settings.onedrive_refresh_token)


def _should_upload_to_s3():
    return bool(settings.s3_bucket_name and settings.aws_access_key_id and settings.aws_secret_access_key)


def get_configured_services_from_validator():
    """
    Use the config validator to determine which services are configured properly.
    Returns a dictionary with service names as keys and boolean values indicating
    whether they're properly configured.
    """
    providers = get_provider_status()

    service_map = {
        "Dropbox": "dropbox",
        "NextCloud": "nextcloud",
        "Paperless-ngx": "paperless",
        "Google Drive": "google_drive",
        "WebDAV": "webdav",
        "FTP Storage": "ftp",
        "SFTP Storage": "sftp",
        "Email": "email",
        "OneDrive": "onedrive",
        "S3 Storage": "s3",
    }

    result = {}
    for provider_name, internal_name in service_map.items():
        if provider_name in providers:
            result[internal_name] = providers[provider_name].get("configured", False)

    return result


@celery.task(base=BaseTaskWithRetry, bind=True)
def send_to_all_destinations(self, file_path: str, use_validator=True, file_id: int = None, folder_overrides=None):
    """
    Distribute a file to all configured storage destinations.

    Args:
        file_path: Path to the file to distribute
        use_validator: Whether to use the config validator to determine enabled services
                       (if False, falls back to individual checks)
        file_id: Optional file ID to associate with logs
        folder_overrides: Optional dict mapping provider names to folder override strings.
                          When set, the override is passed to the upload task which uses it
                          instead of the provider's default folder.  Example:
                          {"dropbox": "/Documents/pdfa", "s3": "docs/pdfa/"}
    """
    task_id = self.request.id

    if not os.path.exists(file_path):
        logger.error(f"[{task_id}] File not found: {file_path}")
        log_task_progress(task_id, "send_to_all_destinations", "failure", "File not found", file_id=file_id)
        raise FileNotFoundError(f"File not found: {file_path}")

    logger.info(f"[{task_id}] Sending {file_path} to all configured destinations")
    log_task_progress(
        task_id,
        "send_to_all_destinations",
        "in_progress",
        f"Distributing: {os.path.basename(file_path)}",
        file_id=file_id,
    )

    # Get file_id from database if not provided (fallback only, prefer passing file_id explicitly)
    if file_id is None:
        with SessionLocal() as db:
            # Only as a last resort, try to find by basename match
            # This should not be needed if file_id is passed correctly through the chain
            file_record = (
                db.query(FileRecord)
                .filter(FileRecord.local_filename == os.path.join(settings.workdir, "tmp", os.path.basename(file_path)))
                .first()
            )
            if file_record:
                file_id = file_record.id

    results = {}

    # Define service configurations
    services = [
        {
            "name": "dropbox",
            "should_upload": _should_upload_to_dropbox,
            "upload_func": upload_to_dropbox,
        },
        {
            "name": "nextcloud",
            "should_upload": _should_upload_to_nextcloud,
            "upload_func": upload_to_nextcloud,
        },
        {
            "name": "paperless",
            "should_upload": _should_upload_to_paperless,
            "upload_func": upload_to_paperless,
        },
        {
            "name": "google_drive",
            "should_upload": _should_upload_to_google_drive,
            "upload_func": upload_to_google_drive,
        },
        {
            "name": "webdav",
            "should_upload": _should_upload_to_webdav,
            "upload_func": upload_to_webdav,
        },
        {
            "name": "ftp",
            "should_upload": _should_upload_to_ftp,
            "upload_func": upload_to_ftp,
        },
        {
            "name": "sftp",
            "should_upload": _should_upload_to_sftp,
            "upload_func": upload_to_sftp,
        },
        {
            "name": "email",
            "should_upload": _should_upload_to_email,
            "upload_func": upload_to_email,
        },
        {
            "name": "onedrive",
            "should_upload": _should_upload_to_onedrive,
            "upload_func": upload_to_onedrive,
        },
        {
            "name": "s3",
            "should_upload": _should_upload_to_s3,
            "upload_func": upload_to_s3,
        },
    ]

    # Optionally get configuration status from validator
    configured_services = {}
    if use_validator:
        try:
            configured_services = get_configured_services_from_validator()
            logger.info(f"[{task_id}] Configured services according to validator: {configured_services}")
        except Exception as e:
            logger.warning(f"[{task_id}] Failed to get configuration from validator: {str(e)}")
            use_validator = False

    # Process each service
    queued_count = 0
    for service in services:
        service_name = service["name"]

        # Determine if service is configured
        is_configured = False
        if use_validator and service_name in configured_services:
            is_configured = configured_services[service_name]
            logger.debug(f"[{task_id}] {service_name} configuration from validator: {is_configured}")
        else:
            try:
                is_configured = service["should_upload"]()
                logger.debug(f"[{task_id}] {service_name} configuration from function: {is_configured}")
            except Exception as e:
                logger.error(f"[{task_id}] Error checking configuration for {service_name}: {str(e)}")
                is_configured = False

        # Queue the upload task if service is configured
        if is_configured:
            logger.info(f"[{task_id}] Queueing {file_path} for {service_name} upload")
            log_task_progress(
                task_id, f"queue_{service_name}", "in_progress", f"Queueing upload to {service_name}", file_id=file_id
            )
            try:
                kwargs = {"file_id": file_id}
                if folder_overrides and service_name in folder_overrides:
                    kwargs["folder_override"] = folder_overrides[service_name]
                task = service["upload_func"].delay(file_path, **kwargs)
                results[f"{service_name}_task_id"] = task.id
                queued_count += 1
                log_task_progress(
                    task_id, f"queue_{service_name}", "success", f"Queued for {service_name}", file_id=file_id
                )
            except Exception as e:
                logger.error(f"[{task_id}] Failed to queue {service_name} task: {str(e)}")
                results[f"{service_name}_error"] = str(e)
                log_task_progress(task_id, f"queue_{service_name}", "failure", f"Failed: {str(e)}", file_id=file_id)

    logger.info(f"[{task_id}] Queued {queued_count} upload tasks")
    log_task_progress(task_id, "send_to_all_destinations", "success", f"Queued {queued_count} uploads", file_id=file_id)

    return {"status": "Queued", "file_path": file_path, "tasks": results}