From 70fb539b7b358d21c76d5b6f195dde4e86f508ea Mon Sep 17 00:00:00 2001
From: Christian Krakau-Louis <christianlouis@gmail.com>
Date: Wed, 12 Feb 2025 02:26:41 +0100
Subject: [PATCH] putting it all together

---
 app/celery_worker.py                   |  1 +
 app/main.py                            | 54 ++++++++++++++++++++++++++
 app/tasks/finalize_document_storage.py | 18 +++++++--
 app/tasks/send_to_all.py               | 21 ++++++++++
 4 files changed, 91 insertions(+), 3 deletions(-)
 create mode 100644 app/tasks/send_to_all.py

diff --git a/app/celery_worker.py b/app/celery_worker.py
index 6c15a964..9784b81a 100644
--- a/app/celery_worker.py
+++ b/app/celery_worker.py
@@ -20,6 +20,7 @@ from app.tasks.upload_to_dropbox import upload_to_dropbox
 from app.tasks.upload_to_paperless import upload_to_paperless
 from app.tasks.upload_to_nextcloud import upload_to_nextcloud
 from app.tasks.imap_tasks import pull_all_inboxes
+from app.tasks.send_to_all import send_to_all_destinations
 
 celery.conf.task_routes = {
     "app.tasks.*": {"queue": "default"},
diff --git a/app/main.py b/app/main.py
index c6413f05..4a236c5c 100644
--- a/app/main.py
+++ b/app/main.py
@@ -7,6 +7,7 @@ from app.tasks.upload_to_s3 import upload_to_s3
 from app.tasks.upload_to_dropbox import upload_to_dropbox
 from app.tasks.upload_to_paperless import upload_to_paperless
 from app.tasks.upload_to_nextcloud import upload_to_nextcloud
+from app.tasks.send_to_all import send_to_all_destinations
 
 app = FastAPI(title="Document Processing API")
 
@@ -57,3 +58,56 @@ def send_to_nextcloud(file_path: str):
         raise HTTPException(status_code=400, detail=f"File {file_path} not found.")
     task = upload_to_nextcloud.delay(file_path)
     return {"task_id": task.id, "status": "queued"}
+
+
+@app.post("/send_to_all_destinations/")
+def send_to_all_destinations_endpoint(file_path: str):
+    """
+    Call the aggregator task that sends this file to dropbox, nextcloud, and paperless.
+    """
+    if not os.path.isabs(file_path):
+        # If not absolute, assume it's in processed subdir
+        file_path = os.path.join(settings.workdir, 'processed', file_path)
+
+    if not os.path.exists(file_path):
+        raise HTTPException(
+            status_code=400,
+            detail=f"File {file_path} not found."
+        )
+
+    task = send_to_all_destinations.delay(file_path)
+    return {"task_id": task.id, "status": "queued", "file_path": file_path}
+
+
+
+@app.post("/processall")
+def process_all_pdfs_in_workdir():
+    """
+    Finds all .pdf files in <workdir>/processed
+    and enqueues them for upload_to_s3.
+    """
+    target_dir = settings.workdir
+    if not os.path.exists(target_dir):
+        raise HTTPException(status_code=400, detail=f"Directory {target_dir} does not exist.")
+
+    pdf_files = []
+    for filename in os.listdir(target_dir):
+        if filename.lower().endswith(".pdf"):
+            pdf_files.append(filename)
+
+    if not pdf_files:
+        return {"message": "No PDF files found in processed directory."}
+
+    task_ids = []
+    for pdf in pdf_files:
+        file_path = os.path.join(target_dir, pdf)
+        # Enqueue upload_to_s3
+        from app.tasks.upload_to_s3 import upload_to_s3
+        task = upload_to_s3.delay(file_path)
+        task_ids.append(task.id)
+
+    return {
+        "message": f"Enqueued {len(pdf_files)} PDFs to upload_to_s3",
+        "pdf_files": pdf_files,
+        "task_ids": task_ids
+    }
diff --git a/app/tasks/finalize_document_storage.py b/app/tasks/finalize_document_storage.py
index 23d1896a..cf9f6944 100644
--- a/app/tasks/finalize_document_storage.py
+++ b/app/tasks/finalize_document_storage.py
@@ -2,13 +2,25 @@
 
 from app.config import settings
 from app.tasks.retry_config import BaseTaskWithRetry
-
 # Import the shared Celery instance
 from app.celery_app import celery
 
+# 1) Import the aggregator task
+from app.tasks.send_to_all import send_to_all_destinations
+
+
 @celery.task(base=BaseTaskWithRetry)
 def finalize_document_storage(original_file: str, processed_file: str, metadata: dict):
-    """Final storage step after embedding metadata."""
+    """
+    Final storage step after embedding metadata.
+    We will now call 'send_to_all_destinations' to push the final PDF to Dropbox/Nextcloud/Paperless.
+    """
     print(f"[INFO] Finalizing document storage for {processed_file}")
-    return {"status": "Completed", "file": processed_file}
 
+    # 2) Enqueue uploads to all destinations (Dropbox, Nextcloud, Paperless)
+    send_to_all_destinations.delay(processed_file)
+
+    return {
+        "status": "Completed",
+        "file": processed_file
+    }
diff --git a/app/tasks/send_to_all.py b/app/tasks/send_to_all.py
new file mode 100644
index 00000000..fe3cee7f
--- /dev/null
+++ b/app/tasks/send_to_all.py
@@ -0,0 +1,21 @@
+# app/tasks/send_to_all.py
+
+from app.celery_app import celery
+from app.tasks.upload_to_dropbox import upload_to_dropbox
+from app.tasks.upload_to_nextcloud import upload_to_nextcloud
+from app.tasks.upload_to_paperless import upload_to_paperless
+
+@celery.task
+def send_to_all_destinations(file_path: str):
+    """
+    Fires off tasks to upload a single file to Dropbox, Nextcloud, and Paperless.
+    These tasks run in parallel (Celery returns immediately from each .delay()).
+    """
+    upload_to_dropbox.delay(file_path)
+    upload_to_nextcloud.delay(file_path)
+    upload_to_paperless.delay(file_path)
+
+    return {
+        "status": "All upload tasks enqueued",
+        "file_path": file_path
+    }