gitea-admin/gh-christianlouis-docuelevate
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Add database logging to upload tasks

Browse Source 
- Add logging to upload_to_dropbox, upload_to_paperless, upload_to_nextcloud
- Pass file_id through send_to_all to upload tasks
- Log upload progress, success, and failures with context

Co-authored-by: christianlouis <361235+christianlouis@users.noreply.github.com>
This commit is contained in:
copilot-swe-agent[bot]
2026-02-07 14:02:29 +00:00
parent adb0b2329e
commit 2189802ee8
4 changed files with 90 additions and 31 deletions
Download Patch File Download Diff File Expand all files Collapse all files
app/tasks/send_to_all.py
+1 -1
View File
@@ -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") 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) log_task_progress(task_id, f"queue_{service_name}", "in_progress", f"Queueing upload to {service_name}", file_id=file_id)
try: 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 results[f"{service_name}_task_id"] = task.id
queued_count += 1 queued_count += 1
log_task_progress(task_id, f"queue_{service_name}", "success", f"Queued for {service_name}", file_id=file_id) log_task_progress(task_id, f"queue_{service_name}", "success", f"Queued for {service_name}", file_id=file_id)
app/tasks/upload_to_dropbox.py
+30 -12
View File
@@ -9,6 +9,9 @@ from app.config import settings
from app.tasks.retry_config import BaseTaskWithRetry from app.tasks.retry_config import BaseTaskWithRetry
from app.celery_app import celery from app.celery_app import celery
from app.utils.filename_utils import get_unique_filename, sanitize_filename, extract_remote_path 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__) logger = logging.getLogger(__name__)
@@ -99,21 +102,31 @@ def get_dropbox_client():
logger.error(f"Error creating Dropbox client: {str(e)}") logger.error(f"Error creating Dropbox client: {str(e)}")
raise raise
@celery.task(base=BaseTaskWithRetry) @celery.task(base=BaseTaskWithRetry, bind=True)
def upload_to_dropbox(file_path: str): def upload_to_dropbox(self, file_path: str, file_id: int = None):
""" """
Upload a file to Dropbox. 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): if not os.path.exists(file_path):
error_msg = f"File not found: {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) raise FileNotFoundError(error_msg)
# Check if Dropbox is properly configured # Check if Dropbox is properly configured
if not (hasattr(settings, 'dropbox_app_key') and settings.dropbox_app_key and 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_app_secret') and settings.dropbox_app_secret and
hasattr(settings, 'dropbox_refresh_token') and settings.dropbox_refresh_token): 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"} return {"status": "Skipped", "reason": "Dropbox settings not configured"}
filename = os.path.basename(file_path) 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) dropbox_path = get_unique_filename(remote_full_path, check_exists_in_dropbox)
# Upload the file # 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: with open(file_path, 'rb') as file_data:
# Use files_upload_session for large files to avoid timeouts # Use files_upload_session for large files to avoid timeouts
file_size = os.path.getsize(file_path) file_size = os.path.getsize(file_path)
@@ -179,7 +193,8 @@ def upload_to_dropbox(file_path: str):
mode=dropbox.files.WriteMode.overwrite 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 { return {
"status": "Completed", "status": "Completed",
"file_path": file_path, "file_path": file_path,
@@ -187,14 +202,17 @@ def upload_to_dropbox(file_path: str):
} }
except AuthError: except AuthError:
error_msg = f"[ERROR] Authentication failed while uploading {filename} to Dropbox. Check token." error_msg = f"Authentication failed while uploading {filename} to Dropbox. Check token."
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 Exception(error_msg) raise Exception(error_msg)
except ApiError as e: except ApiError as e:
error_msg = f"[ERROR] Failed to upload {filename} to Dropbox: {e}" error_msg = f"Failed to upload {filename} to Dropbox: {e}"
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 Exception(error_msg) raise Exception(error_msg)
except Exception as e: except Exception as e:
error_msg = f"[ERROR] Unexpected error uploading {filename} to Dropbox: {e}" error_msg = f"Unexpected error uploading {filename} to Dropbox: {e}"
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 Exception(error_msg) raise Exception(error_msg)
app/tasks/upload_to_nextcloud.py
+27 -10
View File
@@ -8,17 +8,29 @@ from app.config import settings
from app.celery_app import celery from app.celery_app import celery
from app.tasks.retry_config import BaseTaskWithRetry from app.tasks.retry_config import BaseTaskWithRetry
from app.utils.filename_utils import get_unique_filename, sanitize_filename, extract_remote_path 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__) logger = logging.getLogger(__name__)
@celery.task(base=BaseTaskWithRetry) @celery.task(base=BaseTaskWithRetry, bind=True)
def upload_to_nextcloud(file_path: str): def upload_to_nextcloud(self, file_path: str, file_id: int = None):
""" """
Upload a file to Nextcloud WebDAV. 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): if not os.path.exists(file_path):
error_msg = f"File not found: {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) raise FileNotFoundError(error_msg)
# For Nextcloud, we need to check for 'nextcloud_upload_url' instead of 'nextcloud_url' # 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 if not (getattr(settings, 'nextcloud_upload_url', None) and
getattr(settings, 'nextcloud_username', None) and getattr(settings, 'nextcloud_username', None) and
getattr(settings, 'nextcloud_password', None)): 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"} return {"status": "Skipped", "reason": "Nextcloud settings not configured"}
filename = os.path.basename(file_path) filename = os.path.basename(file_path)
@@ -99,7 +112,8 @@ def upload_to_nextcloud(file_path: str):
) )
# Upload the file # 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: with open(file_path, 'rb') as file_data:
response = requests.put( response = requests.put(
full_url, full_url,
@@ -110,7 +124,8 @@ def upload_to_nextcloud(file_path: str):
) )
if response.status_code in (201, 204): # Created or No Content 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 { return {
"status": "Completed", "status": "Completed",
"file_path": file_path, "file_path": file_path,
@@ -118,11 +133,13 @@ def upload_to_nextcloud(file_path: str):
"response_code": response.status_code "response_code": response.status_code
} }
else: else:
error_msg = f"[ERROR] Failed to upload {filename} to Nextcloud: {response.status_code} - {response.text}" error_msg = f"Failed to upload {filename} to Nextcloud: {response.status_code} - {response.text}"
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 Exception(error_msg) raise Exception(error_msg)
except Exception as e: except Exception as e:
error_msg = f"[ERROR] Failed to upload {filename} to Nextcloud: {str(e)}" error_msg = f"Failed to upload {filename} to Nextcloud: {str(e)}"
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 Exception(error_msg) raise Exception(error_msg)
app/tasks/upload_to_paperless.py
+32 -8
View File
@@ -10,6 +10,9 @@ from typing import Dict, Any
from app.config import settings from app.config import settings
from app.tasks.retry_config import BaseTaskWithRetry from app.tasks.retry_config import BaseTaskWithRetry
from app.celery_app import celery 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__) 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." f"Task {task_id} didn't reach SUCCESS within {POLL_MAX_ATTEMPTS} attempts."
) )
@celery.task(base=BaseTaskWithRetry) @celery.task(base=BaseTaskWithRetry, bind=True)
def upload_to_paperless(file_path: str): def upload_to_paperless(self, file_path: str, file_id: int = None):
"""Uploads a file to Paperless-ngx.""" """
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): 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 # Extract filename
filename = os.path.basename(file_path) filename = os.path.basename(file_path)
@@ -94,10 +109,13 @@ def upload_to_paperless(file_path: str):
# Check if Paperless settings are configured # Check if Paperless settings are configured
if not settings.paperless_host or not settings.paperless_ngx_api_token: if not settings.paperless_host or not settings.paperless_ngx_api_token:
error_msg = "Paperless-ngx credentials are not fully configured" 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) raise ValueError(error_msg)
# Upload the PDF # 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/") post_url = _paperless_api_url("/api/documents/post_document/")
with open(file_path, "rb") as f: with open(file_path, "rb") as f:
files = { 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 = requests.post(post_url, headers=_get_headers(), files=files, data=data)
resp.raise_for_status() resp.raise_for_status()
except requests.exceptions.RequestException as exc: except requests.exceptions.RequestException as exc:
error_msg = f"Failed to upload to Paperless: {exc}"
logger.error( 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", "<no response>") file_path, exc, getattr(exc.response, "text", "<no response>")
) )
log_task_progress(task_id, "upload_to_paperless", "failure", error_msg, file_id=file_id)
raise raise
raw_task_id = resp.text.strip().strip('"').strip("'") 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 # 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) 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 { return {
"status": "Completed", "status": "Completed",