diff --git a/app/tasks/upload_to_email.py b/app/tasks/upload_to_email.py index c97af589..a940aecd 100644 --- a/app/tasks/upload_to_email.py +++ b/app/tasks/upload_to_email.py @@ -15,6 +15,7 @@ from jinja2 import Environment, FileSystemLoader, select_autoescape 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 logger = logging.getLogger(__name__) @@ -159,14 +160,32 @@ def _send_email_with_smtp(msg, filename, recipients): logger.error(error_msg) return {"status": "Failed", "reason": error_msg, "error": str(e)} -@celery.task(base=BaseTaskWithRetry) -def upload_to_email(file_path: str, recipients=None, subject=None, message=None, template_name="default.html", include_metadata=True): +@celery.task(base=BaseTaskWithRetry, bind=True) +def upload_to_email(self, file_path: str, file_id: int = None, recipients=None, subject=None, message=None, template_name="default.html", include_metadata=True): """ Sends a file via email to the specified recipients. If recipients is None, uses the configured default email recipient. + + Args: + file_path: Path to the file to send + file_id: Optional file ID to associate with logs + recipients: Optional list of recipient email addresses + subject: Optional email subject + message: Optional custom message + template_name: Email template to use + include_metadata: Whether to include metadata in the email """ + task_id = self.request.id + logger.info(f"[{task_id}] Starting email send: {file_path}") + log_task_progress( + task_id, "upload_to_email", "in_progress", f"Sending via email: {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_email", "failure", error_msg, file_id=file_id) + raise FileNotFoundError(error_msg) # Extract filename filename = os.path.basename(file_path) @@ -174,16 +193,19 @@ def upload_to_email(file_path: str, recipients=None, subject=None, message=None, # Check if email settings are configured if not settings.email_host: error_msg = "Email host is not configured" - logger.error(error_msg) + logger.error(f"[{task_id}] {error_msg}") + log_task_progress(task_id, "upload_to_email", "skipped", error_msg, file_id=file_id) return {"status": "Skipped", "reason": error_msg} # Log email configuration for debugging - logger.debug(f"Email config - Host: {settings.email_host}, Port: {settings.email_port}, " + logger.debug(f"[{task_id}] Email config - Host: {settings.email_host}, Port: {settings.email_port}, " f"Username: {settings.email_username}, TLS: {settings.email_use_tls}") # Process recipients recipients, error = _prepare_recipients(recipients) if error: + logger.error(f"[{task_id}] {error}") + log_task_progress(task_id, "upload_to_email", "skipped", error, file_id=file_id) return {"status": "Skipped", "reason": error} # Use provided subject or create default @@ -237,8 +259,13 @@ def upload_to_email(file_path: str, recipients=None, subject=None, message=None, # Send the email through SMTP error_result = _send_email_with_smtp(msg, filename, recipients) if error_result: + logger.error(f"[{task_id}] Failed to send email: {error_result.get('reason')}") + log_task_progress(task_id, "upload_to_email", "failure", error_result.get('reason'), file_id=file_id) return error_result + logger.info(f"[{task_id}] Successfully sent {filename} via email to {len(recipients)} recipients") + log_task_progress(task_id, "upload_to_email", "success", f"Sent via email: {filename}", file_id=file_id) + return { "status": "Completed", "file": file_path, @@ -250,5 +277,6 @@ def upload_to_email(file_path: str, recipients=None, subject=None, message=None, except Exception as e: error_msg = f"Failed to send {filename} via email: {str(e)}" - logger.error(error_msg) + logger.error(f"[{task_id}] {error_msg}") + log_task_progress(task_id, "upload_to_email", "failure", error_msg, file_id=file_id) raise Exception(error_msg) diff --git a/app/tasks/upload_to_ftp.py b/app/tasks/upload_to_ftp.py index 4bd81964..8e61e493 100644 --- a/app/tasks/upload_to_ftp.py +++ b/app/tasks/upload_to_ftp.py @@ -5,16 +5,31 @@ import ftplib 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 import logging logger = logging.getLogger(__name__) -@celery.task(base=BaseTaskWithRetry) -def upload_to_ftp(file_path: str): - """Uploads a file to an FTP server in the configured folder.""" +@celery.task(base=BaseTaskWithRetry, bind=True) +def upload_to_ftp(self, file_path: str, file_id: int = None): + """ + Uploads a file to an FTP server in the configured folder. + + 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 FTP upload: {file_path}") + log_task_progress( + task_id, "upload_to_ftp", "in_progress", f"Uploading to FTP: {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_ftp", "failure", error_msg, file_id=file_id) + raise FileNotFoundError(error_msg) # Extract filename filename = os.path.basename(file_path) @@ -22,7 +37,8 @@ def upload_to_ftp(file_path: str): # Check if FTP settings are configured if not settings.ftp_host: error_msg = "FTP host is not configured" - logger.error(error_msg) + logger.error(f"[{task_id}] {error_msg}") + log_task_progress(task_id, "upload_to_ftp", "failure", error_msg, file_id=file_id) raise ValueError(error_msg) try: @@ -123,7 +139,8 @@ def upload_to_ftp(file_path: str): # Close FTP connection ftp.quit() - logger.info(f"Successfully uploaded {filename} to FTP server at {settings.ftp_host}") + logger.info(f"[{task_id}] Successfully uploaded {filename} to FTP server at {settings.ftp_host}") + log_task_progress(task_id, "upload_to_ftp", "success", f"Uploaded to FTP: {filename}", file_id=file_id) return { "status": "Completed", "file": file_path, @@ -134,5 +151,6 @@ def upload_to_ftp(file_path: str): except Exception as e: error_msg = f"Failed to upload {filename} to FTP server: {str(e)}" - logger.error(error_msg) + logger.error(f"[{task_id}] {error_msg}") + log_task_progress(task_id, "upload_to_ftp", "failure", error_msg, file_id=file_id) raise Exception(error_msg) diff --git a/app/tasks/upload_to_google_drive.py b/app/tasks/upload_to_google_drive.py index 6657a6d7..1d539fed 100644 --- a/app/tasks/upload_to_google_drive.py +++ b/app/tasks/upload_to_google_drive.py @@ -16,6 +16,7 @@ from google.auth.exceptions import RefreshError 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 logger = logging.getLogger(__name__) @@ -142,12 +143,27 @@ def truncate_property_value(key, value, max_bytes=100): return str_value -@celery.task(base=BaseTaskWithRetry) -def upload_to_google_drive(file_path: str, include_metadata=True): - """Uploads a file to Google Drive in the configured folder with optional metadata.""" +@celery.task(base=BaseTaskWithRetry, bind=True) +def upload_to_google_drive(self, file_path: str, file_id: int = None, include_metadata=True): + """ + Uploads a file to Google Drive in the configured folder with optional metadata. + + Args: + file_path: Path to the file to upload + file_id: Optional file ID to associate with logs + include_metadata: Whether to include metadata in the upload + """ + task_id = self.request.id + logger.info(f"[{task_id}] Starting Google Drive upload: {file_path}") + log_task_progress( + task_id, "upload_to_google_drive", "in_progress", f"Uploading to Google Drive: {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_google_drive", "failure", error_msg, file_id=file_id) + raise FileNotFoundError(error_msg) # Extract filename from path filename = os.path.basename(file_path) @@ -161,7 +177,10 @@ def upload_to_google_drive(file_path: str, include_metadata=True): # Get Google Drive service service = get_google_drive_service() if not service: - raise Exception("Failed to initialize Google Drive service") + error_msg = "Failed to initialize Google Drive service" + logger.error(f"[{task_id}] {error_msg}") + log_task_progress(task_id, "upload_to_google_drive", "failure", error_msg, file_id=file_id) + raise Exception(error_msg) # Prepare the file metadata file_metadata = { @@ -221,16 +240,17 @@ def upload_to_google_drive(file_path: str, include_metadata=True): ).execute() # Log success details - file_id = file.get('id') + file_id_gd = file.get('id') web_view_link = file.get('webViewLink') - logger.info(f"Successfully uploaded {filename} to Google Drive with ID: {file_id}") - logger.info(f"File accessible at: {web_view_link}") + logger.info(f"[{task_id}] Successfully uploaded {filename} to Google Drive with ID: {file_id_gd}") + logger.info(f"[{task_id}] File accessible at: {web_view_link}") + log_task_progress(task_id, "upload_to_google_drive", "success", f"Uploaded to Google Drive: {filename}", file_id=file_id) result = { "status": "Completed", "file_path": file_path, - "google_drive_file_id": file_id, + "google_drive_file_id": file_id_gd, "google_drive_web_link": web_view_link } @@ -242,5 +262,6 @@ def upload_to_google_drive(file_path: str, include_metadata=True): except Exception as e: error_msg = f"Failed to upload {filename} to Google Drive: {str(e)}" - logger.error(error_msg) + logger.error(f"[{task_id}] {error_msg}") + log_task_progress(task_id, "upload_to_google_drive", "failure", error_msg, file_id=file_id) raise Exception(error_msg) diff --git a/app/tasks/upload_to_sftp.py b/app/tasks/upload_to_sftp.py index 866ce6ed..dbf890a6 100644 --- a/app/tasks/upload_to_sftp.py +++ b/app/tasks/upload_to_sftp.py @@ -8,21 +8,35 @@ 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 logger = logging.getLogger(__name__) -@celery.task(base=BaseTaskWithRetry) -def upload_to_sftp(file_path: str): +@celery.task(base=BaseTaskWithRetry, bind=True) +def upload_to_sftp(self, file_path: str, file_id: int = None): """ Upload a file to an SFTP server. + + 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 SFTP upload: {file_path}") + log_task_progress( + task_id, "upload_to_sftp", "in_progress", f"Uploading to SFTP: {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_sftp", "failure", error_msg, file_id=file_id) raise FileNotFoundError(error_msg) if not (settings.sftp_host and settings.sftp_port and settings.sftp_username): - logger.info("SFTP upload skipped: Missing configuration") + error_msg = "SFTP upload skipped: Missing configuration" + logger.info(f"[{task_id}] {error_msg}") + log_task_progress(task_id, "upload_to_sftp", "skipped", error_msg, file_id=file_id) return {"status": "Skipped", "reason": "SFTP settings not configured"} filename = os.path.basename(file_path) @@ -102,9 +116,10 @@ def upload_to_sftp(file_path: str): logger.warning(f"Failed to create directory structure {remote_dir}: {str(e)}") # Upload the file - logger.info(f"Uploading {filename} to SFTP at {remote_path}") + logger.info(f"[{task_id}] Uploading {filename} to SFTP at {remote_path}") sftp.put(file_path, remote_path) - logger.info(f"Successfully uploaded {filename} to SFTP at {remote_path}") + logger.info(f"[{task_id}] Successfully uploaded {filename} to SFTP at {remote_path}") + log_task_progress(task_id, "upload_to_sftp", "success", f"Uploaded to SFTP: {filename}", file_id=file_id) # Close connections sftp.close() @@ -126,5 +141,6 @@ def upload_to_sftp(file_path: str): pass error_msg = f"Failed to upload {filename} to SFTP server: {str(e)}" - logger.error(error_msg) + logger.error(f"[{task_id}] {error_msg}") + log_task_progress(task_id, "upload_to_sftp", "failure", error_msg, file_id=file_id) raise Exception(error_msg) diff --git a/app/tasks/upload_to_webdav.py b/app/tasks/upload_to_webdav.py index adb0cec4..5a654ba0 100644 --- a/app/tasks/upload_to_webdav.py +++ b/app/tasks/upload_to_webdav.py @@ -6,16 +6,31 @@ from urllib.parse import urljoin 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 import logging logger = logging.getLogger(__name__) -@celery.task(base=BaseTaskWithRetry) -def upload_to_webdav(file_path: str): - """Uploads a file to a WebDAV server in the configured folder.""" +@celery.task(base=BaseTaskWithRetry, bind=True) +def upload_to_webdav(self, file_path: str, file_id: int = None): + """ + Uploads a file to a WebDAV server in the configured folder. + + 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 WebDAV upload: {file_path}") + log_task_progress( + task_id, "upload_to_webdav", "in_progress", f"Uploading to WebDAV: {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_webdav", "failure", error_msg, file_id=file_id) + raise FileNotFoundError(error_msg) # Extract filename filename = os.path.basename(file_path) @@ -23,7 +38,8 @@ def upload_to_webdav(file_path: str): # Check if WebDAV settings are configured if not settings.webdav_url: error_msg = "WebDAV URL is not configured" - logger.error(error_msg) + logger.error(f"[{task_id}] {error_msg}") + log_task_progress(task_id, "upload_to_webdav", "failure", error_msg, file_id=file_id) raise ValueError(error_msg) # Construct the full upload URL @@ -53,14 +69,17 @@ def upload_to_webdav(file_path: str): # Check if upload was successful if response.status_code in (200, 201, 204): - logger.info(f"Successfully uploaded {filename} to WebDAV at {webdav_url}.") + logger.info(f"[{task_id}] Successfully uploaded {filename} to WebDAV at {webdav_url}.") + log_task_progress(task_id, "upload_to_webdav", "success", f"Uploaded to WebDAV: {filename}", file_id=file_id) return {"status": "Completed", "file": file_path, "url": webdav_url} else: error_msg = f"Failed to upload {filename} to WebDAV: {response.status_code} - {response.text}" - logger.error(error_msg) + logger.error(f"[{task_id}] {error_msg}") + log_task_progress(task_id, "upload_to_webdav", "failure", error_msg, file_id=file_id) raise Exception(error_msg) except Exception as e: error_msg = f"Error uploading {filename} to WebDAV: {str(e)}" - logger.error(error_msg) + logger.error(f"[{task_id}] {error_msg}") + log_task_progress(task_id, "upload_to_webdav", "failure", error_msg, file_id=file_id) raise Exception(error_msg)