diff --git a/app/tasks/send_to_all.py b/app/tasks/send_to_all.py index 1f284c82..33a9774f 100644 --- a/app/tasks/send_to_all.py +++ b/app/tasks/send_to_all.py @@ -228,7 +228,7 @@ def send_to_all_destinations(self, file_path: str, use_validator=True, file_id: 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: - task = service["upload_func"].delay(file_path) + task = service["upload_func"].delay(file_path, file_id) 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) diff --git a/app/tasks/upload_to_dropbox.py b/app/tasks/upload_to_dropbox.py index 03415953..d7930115 100644 --- a/app/tasks/upload_to_dropbox.py +++ b/app/tasks/upload_to_dropbox.py @@ -9,6 +9,9 @@ from app.config import settings from app.tasks.retry_config import BaseTaskWithRetry from app.celery_app import celery from app.utils.filename_utils import get_unique_filename, sanitize_filename, extract_remote_path +from app.utils import log_task_progress +from app.database import SessionLocal +from app.models import FileRecord logger = logging.getLogger(__name__) @@ -99,21 +102,31 @@ def get_dropbox_client(): logger.error(f"Error creating Dropbox client: {str(e)}") raise -@celery.task(base=BaseTaskWithRetry) -def upload_to_dropbox(file_path: str): +@celery.task(base=BaseTaskWithRetry, bind=True) +def upload_to_dropbox(self, file_path: str, file_id: int = None): """ Upload a file to Dropbox. + + 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 Dropbox upload: {file_path}") + log_task_progress(task_id, "upload_to_dropbox", "in_progress", f"Uploading to Dropbox: {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(error_msg) + logger.error(f"[{task_id}] {error_msg}") + log_task_progress(task_id, "upload_to_dropbox", "failure", error_msg, file_id=file_id) raise FileNotFoundError(error_msg) # Check if Dropbox is properly configured if not (hasattr(settings, 'dropbox_app_key') and settings.dropbox_app_key and hasattr(settings, 'dropbox_app_secret') and settings.dropbox_app_secret and hasattr(settings, 'dropbox_refresh_token') and settings.dropbox_refresh_token): - logger.info("Dropbox upload skipped: Missing configuration") + logger.info(f"[{task_id}] Dropbox upload skipped: Missing configuration") + log_task_progress(task_id, "upload_to_dropbox", "success", "Skipped: Not configured", file_id=file_id) return {"status": "Skipped", "reason": "Dropbox settings not configured"} filename = os.path.basename(file_path) @@ -144,7 +157,8 @@ def upload_to_dropbox(file_path: str): dropbox_path = get_unique_filename(remote_full_path, check_exists_in_dropbox) # Upload the file - logger.info(f"Uploading {filename} to Dropbox at {dropbox_path}") + logger.info(f"[{task_id}] Uploading {filename} to Dropbox at {dropbox_path}") + log_task_progress(task_id, "upload_file", "in_progress", f"Uploading to {dropbox_path}", file_id=file_id) with open(file_path, 'rb') as file_data: # Use files_upload_session for large files to avoid timeouts file_size = os.path.getsize(file_path) @@ -179,7 +193,8 @@ def upload_to_dropbox(file_path: str): mode=dropbox.files.WriteMode.overwrite ) - logger.info(f"Successfully uploaded {filename} to Dropbox at {dropbox_path}") + logger.info(f"[{task_id}] Successfully uploaded {filename} to Dropbox at {dropbox_path}") + log_task_progress(task_id, "upload_to_dropbox", "success", f"Uploaded to Dropbox: {dropbox_path}", file_id=file_id) return { "status": "Completed", "file_path": file_path, @@ -187,14 +202,17 @@ def upload_to_dropbox(file_path: str): } except AuthError: - error_msg = f"[ERROR] Authentication failed while uploading {filename} to Dropbox. Check token." - logger.error(error_msg) + error_msg = f"Authentication failed while uploading {filename} to Dropbox. Check token." + logger.error(f"[{task_id}] {error_msg}") + log_task_progress(task_id, "upload_to_dropbox", "failure", error_msg, file_id=file_id) raise Exception(error_msg) except ApiError as e: - error_msg = f"[ERROR] Failed to upload {filename} to Dropbox: {e}" - logger.error(error_msg) + error_msg = f"Failed to upload {filename} to Dropbox: {e}" + logger.error(f"[{task_id}] {error_msg}") + log_task_progress(task_id, "upload_to_dropbox", "failure", error_msg, file_id=file_id) raise Exception(error_msg) except Exception as e: - error_msg = f"[ERROR] Unexpected error uploading {filename} to Dropbox: {e}" - logger.error(error_msg) + error_msg = f"Unexpected error uploading {filename} to Dropbox: {e}" + logger.error(f"[{task_id}] {error_msg}") + log_task_progress(task_id, "upload_to_dropbox", "failure", error_msg, file_id=file_id) raise Exception(error_msg) diff --git a/app/tasks/upload_to_nextcloud.py b/app/tasks/upload_to_nextcloud.py index 2a3cebc0..3dd98a22 100644 --- a/app/tasks/upload_to_nextcloud.py +++ b/app/tasks/upload_to_nextcloud.py @@ -8,17 +8,29 @@ from app.config import settings from app.celery_app import celery from app.tasks.retry_config import BaseTaskWithRetry from app.utils.filename_utils import get_unique_filename, sanitize_filename, extract_remote_path +from app.utils import log_task_progress +from app.database import SessionLocal +from app.models import FileRecord logger = logging.getLogger(__name__) -@celery.task(base=BaseTaskWithRetry) -def upload_to_nextcloud(file_path: str): +@celery.task(base=BaseTaskWithRetry, bind=True) +def upload_to_nextcloud(self, file_path: str, file_id: int = 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(error_msg) + 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' @@ -26,7 +38,8 @@ def upload_to_nextcloud(file_path: str): if not (getattr(settings, 'nextcloud_upload_url', None) and getattr(settings, 'nextcloud_username', None) and getattr(settings, 'nextcloud_password', None)): - logger.info("Nextcloud upload skipped: Missing configuration") + 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) @@ -99,7 +112,8 @@ def upload_to_nextcloud(file_path: str): ) # Upload the file - logger.info(f"Uploading {filename} to Nextcloud at {full_url}") + 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, @@ -110,7 +124,8 @@ def upload_to_nextcloud(file_path: str): ) if response.status_code in (201, 204): # Created or No Content - logger.info(f"Successfully uploaded {filename} to Nextcloud at {remote_path}") + 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, @@ -118,11 +133,13 @@ def upload_to_nextcloud(file_path: str): "response_code": response.status_code } else: - error_msg = f"[ERROR] Failed to upload {filename} to Nextcloud: {response.status_code} - {response.text}" - logger.error(error_msg) + 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"[ERROR] Failed to upload {filename} to Nextcloud: {str(e)}" - logger.error(error_msg) + 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) diff --git a/app/tasks/upload_to_paperless.py b/app/tasks/upload_to_paperless.py index bd9b5485..dc8af68b 100644 --- a/app/tasks/upload_to_paperless.py +++ b/app/tasks/upload_to_paperless.py @@ -10,6 +10,9 @@ from typing import Dict, Any from app.config import settings from app.tasks.retry_config import BaseTaskWithRetry from app.celery_app import celery +from app.utils import log_task_progress +from app.database import SessionLocal +from app.models import FileRecord logger = logging.getLogger(__name__) @@ -81,12 +84,24 @@ def poll_task_for_document_id(task_id: str) -> int: f"Task {task_id} didn't reach SUCCESS within {POLL_MAX_ATTEMPTS} attempts." ) -@celery.task(base=BaseTaskWithRetry) -def upload_to_paperless(file_path: str): - """Uploads a file to Paperless-ngx.""" +@celery.task(base=BaseTaskWithRetry, bind=True) +def upload_to_paperless(self, file_path: str, file_id: int = None): + """ + Uploads a file to Paperless-ngx. + + 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 Paperless upload: {file_path}") + log_task_progress(task_id, "upload_to_paperless", "in_progress", f"Uploading to Paperless: {os.path.basename(file_path)}", file_id=file_id) if not os.path.exists(file_path): - raise FileNotFoundError(f"File not found: {file_path}") + error_msg = f"File not found: {file_path}" + logger.error(f"[{task_id}] {error_msg}") + log_task_progress(task_id, "upload_to_paperless", "failure", error_msg, file_id=file_id) + raise FileNotFoundError(error_msg) # Extract filename filename = os.path.basename(file_path) @@ -94,10 +109,13 @@ def upload_to_paperless(file_path: str): # Check if Paperless settings are configured if not settings.paperless_host or not settings.paperless_ngx_api_token: error_msg = "Paperless-ngx credentials are not fully configured" - logger.error(error_msg) + logger.error(f"[{task_id}] {error_msg}") + log_task_progress(task_id, "upload_to_paperless", "failure", error_msg, file_id=file_id) raise ValueError(error_msg) # Upload the PDF + logger.info(f"[{task_id}] Posting document to Paperless") + log_task_progress(task_id, "post_document", "in_progress", "Posting to Paperless API", file_id=file_id) post_url = _paperless_api_url("/api/documents/post_document/") with open(file_path, "rb") as f: files = { @@ -110,18 +128,24 @@ def upload_to_paperless(file_path: str): resp = requests.post(post_url, headers=_get_headers(), files=files, data=data) resp.raise_for_status() except requests.exceptions.RequestException as exc: + error_msg = f"Failed to upload to Paperless: {exc}" logger.error( - "Failed to upload document '%s' to Paperless. Error: %s. Response=%s", + f"[{task_id}] Failed to upload document '%s' to Paperless. Error: %s. Response=%s", file_path, exc, getattr(exc.response, "text", "") ) + log_task_progress(task_id, "upload_to_paperless", "failure", error_msg, file_id=file_id) raise raw_task_id = resp.text.strip().strip('"').strip("'") - logger.info(f"Received Paperless task ID: {raw_task_id}") + logger.info(f"[{task_id}] Received Paperless task ID: {raw_task_id}") + log_task_progress(task_id, "post_document", "success", f"Task ID: {raw_task_id}", file_id=file_id) # Poll tasks until success/fail => get doc_id + logger.info(f"[{task_id}] Polling for document ID") + log_task_progress(task_id, "poll_task", "in_progress", "Waiting for Paperless processing", file_id=file_id) doc_id = poll_task_for_document_id(raw_task_id) - logger.info(f"Document {file_path} successfully ingested => ID={doc_id}") + logger.info(f"[{task_id}] Document {file_path} successfully ingested => ID={doc_id}") + log_task_progress(task_id, "upload_to_paperless", "success", f"Uploaded to Paperless: Doc ID {doc_id}", file_id=file_id) return { "status": "Completed",