diff --git a/app/api/__init__.py b/app/api/__init__.py index c2efba12..a621adb6 100644 --- a/app/api/__init__.py +++ b/app/api/__init__.py @@ -14,6 +14,7 @@ from app.api.dropbox import router as dropbox_router from app.api.openai import router as openai_router from app.api.azure import router as azure_router from app.api.google_drive import router as google_drive_router +from app.api.logs import router as logs_router # Set up logging logger = logging.getLogger(__name__) @@ -31,3 +32,4 @@ router.include_router(dropbox_router) router.include_router(openai_router) router.include_router(azure_router) router.include_router(google_drive_router) +router.include_router(logs_router) diff --git a/app/api/logs.py b/app/api/logs.py new file mode 100644 index 00000000..36170c88 --- /dev/null +++ b/app/api/logs.py @@ -0,0 +1,164 @@ +""" +Processing logs API endpoints +""" +from fastapi import APIRouter, Request, HTTPException, Depends, Query +from sqlalchemy.orm import Session +from sqlalchemy import desc +from typing import Optional +import logging + +from app.auth import require_login +from app.models import ProcessingLog, FileRecord +from app.api.common import get_db + +# Set up logging +logger = logging.getLogger(__name__) + +router = APIRouter() + +@router.get("/logs") +@require_login +def list_processing_logs( + request: Request, + db: Session = Depends(get_db), + file_id: Optional[int] = Query(None, description="Filter by file ID"), + task_id: Optional[str] = Query(None, description="Filter by task ID"), + limit: int = Query(100, ge=1, le=1000, description="Number of logs to return") +): + """ + Returns a JSON list of ProcessingLog entries. + Protected by `@require_login`, so only logged-in sessions can access. + + Query Parameters: + - file_id: Optional filter by file ID + - task_id: Optional filter by task ID + - limit: Maximum number of logs to return (default 100, max 1000) + + Example response: + [ + { + "id": 1, + "file_id": 123, + "task_id": "abc-123-def", + "step_name": "process_document", + "status": "success", + "message": "Processing completed", + "timestamp": "2025-05-01T12:34:56.789000" + }, + ... + ] + """ + query = db.query(ProcessingLog) + + # Apply filters + if file_id is not None: + query = query.filter(ProcessingLog.file_id == file_id) + if task_id is not None: + query = query.filter(ProcessingLog.task_id == task_id) + + # Order by timestamp descending and limit + logs = query.order_by(desc(ProcessingLog.timestamp)).limit(limit).all() + + # Return a simple list of dicts + result = [] + for log in logs: + result.append({ + "id": log.id, + "file_id": log.file_id, + "task_id": log.task_id, + "step_name": log.step_name, + "status": log.status, + "message": log.message, + "timestamp": log.timestamp.isoformat() if log.timestamp else None + }) + return result + +@router.get("/logs/file/{file_id}") +@require_login +def get_file_processing_logs( + request: Request, + file_id: int, + db: Session = Depends(get_db) +): + """ + Get all processing logs for a specific file. + Returns logs ordered by timestamp (oldest first to show processing flow). + + Also includes file metadata if the file exists. + """ + # Check if file exists + file_record = db.query(FileRecord).filter(FileRecord.id == file_id).first() + if not file_record: + raise HTTPException( + status_code=404, + detail=f"File with ID {file_id} not found" + ) + + # Get all logs for this file + logs = db.query(ProcessingLog).filter( + ProcessingLog.file_id == file_id + ).order_by(ProcessingLog.timestamp).all() + + # Build response + log_list = [] + for log in logs: + log_list.append({ + "id": log.id, + "task_id": log.task_id, + "step_name": log.step_name, + "status": log.status, + "message": log.message, + "timestamp": log.timestamp.isoformat() if log.timestamp else None + }) + + return { + "file": { + "id": file_record.id, + "original_filename": file_record.original_filename, + "file_size": file_record.file_size, + "mime_type": file_record.mime_type, + "created_at": file_record.created_at.isoformat() if file_record.created_at else None + }, + "logs": log_list, + "total_logs": len(log_list) + } + +@router.get("/logs/task/{task_id}") +@require_login +def get_task_processing_logs( + request: Request, + task_id: str, + db: Session = Depends(get_db) +): + """ + Get all processing logs for a specific task. + Returns logs ordered by timestamp (oldest first to show processing flow). + """ + # Get all logs for this task + logs = db.query(ProcessingLog).filter( + ProcessingLog.task_id == task_id + ).order_by(ProcessingLog.timestamp).all() + + if not logs: + raise HTTPException( + status_code=404, + detail=f"No logs found for task {task_id}" + ) + + # Build response + log_list = [] + for log in logs: + log_list.append({ + "id": log.id, + "file_id": log.file_id, + "step_name": log.step_name, + "status": log.status, + "message": log.message, + "timestamp": log.timestamp.isoformat() if log.timestamp else None + }) + + return { + "task_id": task_id, + "logs": log_list, + "total_logs": len(log_list) + } diff --git a/app/tasks/convert_to_pdf.py b/app/tasks/convert_to_pdf.py index 34fec4a5..1631ea1f 100644 --- a/app/tasks/convert_to_pdf.py +++ b/app/tasks/convert_to_pdf.py @@ -7,25 +7,32 @@ import json from celery import shared_task from app.config import settings from app.tasks.process_document import process_document +from app.utils import log_task_progress logger = logging.getLogger(__name__) -@shared_task -def convert_to_pdf(file_path): +@shared_task(bind=True) +def convert_to_pdf(self, file_path): """ Converts a file to PDF using Gotenberg's API. Determines the appropriate Gotenberg endpoint based on the file's MIME type. On success, saves the PDF locally and enqueues it for processing. """ + task_id = self.request.id + logger.info(f"[{task_id}] Starting PDF conversion: {file_path}") + log_task_progress(task_id, "convert_to_pdf", "in_progress", f"Converting file: {os.path.basename(file_path)}") + gotenberg_url = getattr(settings, "gotenberg_url", None) if not gotenberg_url: - logger.error("Gotenberg URL is not configured in settings.") + logger.error(f"[{task_id}] Gotenberg URL is not configured in settings.") + log_task_progress(task_id, "convert_to_pdf", "failure", "Gotenberg URL not configured") return # Try to guess the MIME type based on file content and extension mime_type, encoding = mimetypes.guess_type(file_path) file_ext = os.path.splitext(file_path)[1].lower() - logger.info(f"Guessed MIME type for '{file_path}' is: {mime_type}, extension: {file_ext}") + logger.info(f"[{task_id}] Guessed MIME type for '{file_path}' is: {mime_type}, extension: {file_ext}") + log_task_progress(task_id, "detect_file_type", "success", f"File type: {mime_type or file_ext}") # Determine which Gotenberg endpoint to use endpoint = None @@ -146,11 +153,13 @@ def convert_to_pdf(file_path): logger.warning(f"Using fallback conversion for unknown type: {mime_type} / {file_ext}") if not endpoint: - logger.error(f"Could not determine Gotenberg endpoint for file type: {mime_type}") + logger.error(f"[{task_id}] Could not determine Gotenberg endpoint for file type: {mime_type}") + log_task_progress(task_id, "convert_to_pdf", "failure", f"Unknown file type: {mime_type}") return None try: - logger.info(f"Converting {file_path} using endpoint: {endpoint}") + logger.info(f"[{task_id}] Converting {file_path} using endpoint: {endpoint}") + log_task_progress(task_id, "call_gotenberg", "in_progress", "Calling Gotenberg API") # Send the conversion request to Gotenberg response = requests.post(endpoint, files=files, data=form_data) @@ -161,19 +170,25 @@ def convert_to_pdf(file_path): with open(converted_file_path, "wb") as out_file: out_file.write(response.content) - logger.info(f"Converted file saved as PDF: {converted_file_path}") + logger.info(f"[{task_id}] Converted file saved as PDF: {converted_file_path}") + log_task_progress(task_id, "call_gotenberg", "success", "PDF conversion successful") + log_task_progress(task_id, "convert_to_pdf", "success", f"Converted to PDF: {os.path.basename(converted_file_path)}") # Enqueue the PDF for further processing process_document.delay(converted_file_path) return converted_file_path else: + error_msg = f"Status code: {response.status_code}" logger.error( - f"Conversion failed for {file_path}. " - f"Status code: {response.status_code}, " + f"[{task_id}] Conversion failed for {file_path}. " + f"{error_msg}, " f"Response: {response.text[:500]}..." ) + log_task_progress(task_id, "call_gotenberg", "failure", error_msg) + log_task_progress(task_id, "convert_to_pdf", "failure", f"Conversion failed: {error_msg}") return None except Exception as e: - logger.exception(f"Error converting {file_path} to PDF: {e}") + logger.exception(f"[{task_id}] Error converting {file_path} to PDF: {e}") + log_task_progress(task_id, "convert_to_pdf", "failure", f"Exception: {str(e)}") return None diff --git a/app/tasks/embed_metadata_into_pdf.py b/app/tasks/embed_metadata_into_pdf.py index d35e63c8..a33ae8c1 100644 --- a/app/tasks/embed_metadata_into_pdf.py +++ b/app/tasks/embed_metadata_into_pdf.py @@ -3,6 +3,7 @@ import os import shutil import tempfile +import logging import PyPDF2 # Replace fitz with PyPDF2 import json from app.config import settings @@ -11,6 +12,11 @@ from app.tasks.finalize_document_storage import finalize_document_storage # Import the shared Celery instance 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__) # Directory constants - defined here to avoid hardcoded strings (BAN-B108) # Note: These are application-specific subdirectories within settings.workdir, @@ -47,8 +53,8 @@ def persist_metadata(metadata, final_pdf_path): json.dump(metadata, f, ensure_ascii=False, indent=2) return json_path -@celery.task(base=BaseTaskWithRetry) -def embed_metadata_into_pdf(local_file_path: str, extracted_text: str, metadata: dict): +@celery.task(base=BaseTaskWithRetry, bind=True) +def embed_metadata_into_pdf(self, local_file_path: str, extracted_text: str, metadata: dict, file_id: int = None): """ Embeds extracted metadata into the PDF's standard metadata fields. The mapping is as follows: @@ -62,13 +68,25 @@ def embed_metadata_into_pdf(local_file_path: str, extracted_text: str, metadata: where is derived from metadata["filename"]. Additionally, the metadata is persisted to a JSON file with the same base name. """ + task_id = self.request.id + logger.info(f"[{task_id}] Starting metadata embedding for: {local_file_path}") + log_task_progress(task_id, "embed_metadata_into_pdf", "in_progress", f"Embedding metadata into {os.path.basename(local_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: + file_record = db.query(FileRecord).filter_by(local_filename=local_file_path).first() + if file_record: + file_id = file_record.id + # Check for file existence; if not found, try the known shared tmp directory. if not os.path.exists(local_file_path): alt_path = os.path.join(settings.workdir, TMP_SUBDIR, os.path.basename(local_file_path)) if os.path.exists(alt_path): local_file_path = alt_path else: - print(f"[ERROR] Local file {local_file_path} not found, cannot embed metadata.") + logger.error(f"[{task_id}] Local file {local_file_path} not found, cannot embed metadata.") + log_task_progress(task_id, "embed_metadata_into_pdf", "failure", "File not found", file_id=file_id) return {"error": "File not found"} # Work on a safe copy in a secure temporary directory @@ -83,7 +101,8 @@ def embed_metadata_into_pdf(local_file_path: str, extracted_text: str, metadata: shutil.copy(original_file, processed_file) try: - print(f"[DEBUG] Embedding metadata into {processed_file}...") + logger.info(f"[{task_id}] Embedding metadata into {processed_file}...") + log_task_progress(task_id, "modify_pdf", "in_progress", "Modifying PDF metadata", file_id=file_id) # Open the PDF and modify metadata with open(processed_file, 'rb') as file: @@ -106,7 +125,8 @@ def embed_metadata_into_pdf(local_file_path: str, extracted_text: str, metadata: with open(processed_file, 'wb') as output_file: pdf_writer.write(output_file) - print(f"[INFO] Metadata embedded successfully in {processed_file}") + logger.info(f"[{task_id}] Metadata embedded successfully in {processed_file}") + log_task_progress(task_id, "modify_pdf", "success", "PDF metadata embedded", file_id=file_id) # Use the suggested filename from metadata; if not provided, use the original basename. suggested_filename = metadata.get("filename", os.path.splitext(os.path.basename(local_file_path))[0]) @@ -118,37 +138,46 @@ def embed_metadata_into_pdf(local_file_path: str, extracted_text: str, metadata: # Get a unique filepath in case of collisions. final_file_path = unique_filepath(final_dir, suggested_filename, extension=".pdf") + logger.info(f"[{task_id}] Moving file to: {final_file_path}") + log_task_progress(task_id, "move_to_processed", "in_progress", f"Moving to processed: {suggested_filename}.pdf", file_id=file_id) # Move the processed file using shutil.move to handle cross-device moves. shutil.move(processed_file, final_file_path) # Ensure the temporary file is deleted if it still exists. if os.path.exists(processed_file): os.remove(processed_file) + log_task_progress(task_id, "move_to_processed", "success", f"Moved to: {os.path.basename(final_file_path)}", file_id=file_id) # Persist the metadata into a JSON file with the same base name. + logger.info(f"[{task_id}] Persisting metadata to JSON") + log_task_progress(task_id, "save_metadata_json", "in_progress", "Saving metadata JSON", file_id=file_id) json_path = persist_metadata(metadata, final_file_path) - print(f"[INFO] Metadata persisted to {json_path}") + logger.info(f"[{task_id}] Metadata persisted to {json_path}") + log_task_progress(task_id, "save_metadata_json", "success", f"Saved: {os.path.basename(json_path)}", file_id=file_id) # Trigger the next step: final storage. - finalize_document_storage.delay(original_file, final_file_path, metadata) + logger.info(f"[{task_id}] Queueing final storage task") + log_task_progress(task_id, "embed_metadata_into_pdf", "success", "Metadata embedded, queuing finalization", file_id=file_id) + finalize_document_storage.delay(original_file, final_file_path, metadata, file_id) # After triggering final storage, delete the original file if it is in workdir/tmp. workdir_tmp = os.path.join(settings.workdir, TMP_SUBDIR) if original_file.startswith(workdir_tmp) and os.path.exists(original_file): try: os.remove(original_file) - print(f"[INFO] Deleted original file from {original_file}") + logger.info(f"[{task_id}] Deleted original file from {original_file}") except Exception as e: - print(f"[ERROR] Could not delete original file {original_file}: {e}") + logger.error(f"[{task_id}] Could not delete original file {original_file}: {e}") return {"file": final_file_path, "metadata_file": json_path, "status": "Metadata embedded"} except Exception as e: - print(f"[ERROR] Failed to embed metadata into {processed_file}: {e}") + logger.exception(f"[{task_id}] Failed to embed metadata into {processed_file}: {e}") + log_task_progress(task_id, "embed_metadata_into_pdf", "failure", f"Exception: {str(e)}", file_id=file_id) # Clean up temporary file in case of error if os.path.exists(processed_file): try: os.remove(processed_file) - print(f"[INFO] Cleaned up temporary file {processed_file}") + logger.info(f"[{task_id}] Cleaned up temporary file {processed_file}") except Exception as cleanup_error: - print(f"[ERROR] Could not clean up temporary file {processed_file}: {cleanup_error}") + logger.error(f"[{task_id}] Could not clean up temporary file {processed_file}: {cleanup_error}") return {"error": str(e)} diff --git a/app/tasks/extract_metadata_with_gpt.py b/app/tasks/extract_metadata_with_gpt.py index 2aab5283..52d465bb 100644 --- a/app/tasks/extract_metadata_with_gpt.py +++ b/app/tasks/extract_metadata_with_gpt.py @@ -2,6 +2,7 @@ import json import re +import os from app.config import settings from app.tasks.retry_config import BaseTaskWithRetry from app.tasks.embed_metadata_into_pdf import embed_metadata_into_pdf @@ -10,6 +11,9 @@ from app.tasks.embed_metadata_into_pdf import embed_metadata_into_pdf from app.celery_app import celery import openai import logging +from app.utils import log_task_progress +from app.database import SessionLocal +from app.models import FileRecord logger = logging.getLogger(__name__) @@ -41,9 +45,23 @@ def extract_json_from_text(text): return text[start:end+1] return None -@celery.task(base=BaseTaskWithRetry) -def extract_metadata_with_gpt(filename: str, cleaned_text: str): +@celery.task(base=BaseTaskWithRetry, bind=True) +def extract_metadata_with_gpt(self, filename: str, cleaned_text: str, file_id: int = None): """Uses OpenAI to classify document metadata.""" + task_id = self.request.id + logger.info(f"[{task_id}] Starting metadata extraction for: {filename}") + log_task_progress(task_id, "extract_metadata_with_gpt", "in_progress", f"Extracting metadata for {filename}", file_id=file_id) + + # Get file_id from database if not provided + if file_id is None: + tmp_dir = os.path.join(settings.workdir, "tmp") + file_path = os.path.join(tmp_dir, filename) + if os.path.exists(file_path): + with SessionLocal() as db: + file_record = db.query(FileRecord).filter_by(local_filename=file_path).first() + if file_record: + file_id = file_record.id + prompt = f""" You are a specialized document analyzer trained to extract structured metadata from documents. Your task is to analyze the given text and return a well-structured JSON object. @@ -77,7 +95,8 @@ Return only valid JSON with no additional commentary. """ try: - print(f"[DEBUG] Sending classification request for {filename}...") + logger.info(f"[{task_id}] Sending classification request for {filename}...") + log_task_progress(task_id, "call_openai", "in_progress", "Calling OpenAI API", file_id=file_id) completion = client.chat.completions.create( model=settings.openai_model, messages=[ @@ -88,21 +107,27 @@ Return only valid JSON with no additional commentary. ) content = completion.choices[0].message.content - print(f"[DEBUG] Raw classification response for {filename}: {content}") + logger.info(f"[{task_id}] Raw classification response for {filename}: {content[:200]}...") + log_task_progress(task_id, "call_openai", "success", "Received OpenAI response", file_id=file_id) json_text = extract_json_from_text(content) if not json_text: - print(f"[ERROR] Could not find valid JSON in GPT response for {filename}.") + logger.error(f"[{task_id}] Could not find valid JSON in GPT response for {filename}.") + log_task_progress(task_id, "extract_metadata_with_gpt", "failure", "Invalid JSON in response", file_id=file_id) return {} metadata = json.loads(json_text) - print(f"[DEBUG] Extracted metadata: {metadata}") + logger.info(f"[{task_id}] Extracted metadata: {metadata}") + log_task_progress(task_id, "parse_metadata", "success", f"Parsed metadata: {list(metadata.keys())}", file_id=file_id) # Trigger the next step: embedding metadata into the PDF - embed_metadata_into_pdf.delay(filename, cleaned_text, metadata) + logger.info(f"[{task_id}] Queueing metadata embedding task") + log_task_progress(task_id, "extract_metadata_with_gpt", "success", "Metadata extracted, queuing embed task", file_id=file_id) + embed_metadata_into_pdf.delay(filename, cleaned_text, metadata, file_id) return {"s3_file": filename, "metadata": metadata} except Exception as e: - print(f"[ERROR] OpenAI classification failed for {filename}: {e}") + logger.exception(f"[{task_id}] OpenAI classification failed for {filename}: {e}") + log_task_progress(task_id, "extract_metadata_with_gpt", "failure", f"Exception: {str(e)}", file_id=file_id) return {} diff --git a/app/tasks/finalize_document_storage.py b/app/tasks/finalize_document_storage.py index 6d5216d0..05ad24e1 100644 --- a/app/tasks/finalize_document_storage.py +++ b/app/tasks/finalize_document_storage.py @@ -1,5 +1,7 @@ #!/usr/bin/env python3 +import logging +import os from app.config import settings from app.tasks.retry_config import BaseTaskWithRetry # Import the shared Celery instance @@ -7,18 +9,39 @@ from app.celery_app import celery # 1) Import the aggregator task from app.tasks.send_to_all import send_to_all_destinations +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 finalize_document_storage(original_file: str, processed_file: str, metadata: dict): +@celery.task(base=BaseTaskWithRetry, bind=True) +def finalize_document_storage(self, original_file: str, processed_file: str, metadata: dict, file_id: int = None): """ 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}") + task_id = self.request.id + logger.info(f"[{task_id}] Finalizing document storage for {processed_file}") + log_task_progress(task_id, "finalize_document_storage", "in_progress", f"Finalizing: {os.path.basename(processed_file)}", 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 exact match on local_filename + # This should not be needed if file_id is passed correctly through the chain + tmp_path = os.path.join(settings.workdir, "tmp", os.path.basename(original_file)) + file_record = db.query(FileRecord).filter( + FileRecord.local_filename == tmp_path + ).first() + if file_record: + file_id = file_record.id # 2) Enqueue uploads to all destinations (Dropbox, Nextcloud, Paperless) - send_to_all_destinations.delay(processed_file) + logger.info(f"[{task_id}] Queueing uploads to all destinations") + log_task_progress(task_id, "finalize_document_storage", "success", "Queuing uploads to destinations", file_id=file_id) + send_to_all_destinations.delay(processed_file, True, file_id) return { "status": "Completed", diff --git a/app/tasks/process_document.py b/app/tasks/process_document.py index fa795600..05cfa74b 100644 --- a/app/tasks/process_document.py +++ b/app/tasks/process_document.py @@ -4,6 +4,7 @@ import os import uuid import shutil import mimetypes +import logging import PyPDF2 # Replace fitz with PyPDF2 from app.config import settings @@ -13,11 +14,13 @@ from app.tasks.extract_metadata_with_gpt import extract_metadata_with_gpt from app.celery_app import celery from app.database import SessionLocal from app.models import FileRecord -from app.utils import hash_file +from app.utils import hash_file, log_task_progress + +logger = logging.getLogger(__name__) -@celery.task(base=BaseTaskWithRetry) -def process_document(original_local_file: str): +@celery.task(base=BaseTaskWithRetry, bind=True) +def process_document(self, original_local_file: str): """ Process a document file and trigger appropriate text extraction. @@ -28,24 +31,34 @@ def process_document(original_local_file: str): - Check for embedded text. If present, run local GPT extraction - Otherwise, queue Azure Document Intelligence processing """ + task_id = self.request.id + logger.info(f"[{task_id}] Starting document processing: {original_local_file}") + log_task_progress(task_id, "process_document", "in_progress", f"Processing file: {original_local_file}") if not os.path.exists(original_local_file): - print(f"[ERROR] File {original_local_file} not found.") + logger.error(f"[{task_id}] File {original_local_file} not found.") + log_task_progress(task_id, "process_document", "failure", "File not found") return {"error": "File not found"} # 0. Compute the file hash and check for duplicates + logger.info(f"[{task_id}] Computing file hash...") + log_task_progress(task_id, "hash_file", "in_progress", "Computing file hash") filehash = hash_file(original_local_file) original_filename = os.path.basename(original_local_file) file_size = os.path.getsize(original_local_file) mime_type, _ = mimetypes.guess_type(original_local_file) if not mime_type: mime_type = "application/octet-stream" + + logger.info(f"[{task_id}] File hash: {filehash[:10]}..., Size: {file_size} bytes, MIME: {mime_type}") + log_task_progress(task_id, "hash_file", "success", f"Hash: {filehash[:10]}..., Size: {file_size} bytes") # Acquire DB session in the task with SessionLocal() as db: existing = db.query(FileRecord).filter_by(filehash=filehash).one_or_none() if existing: - print(f"[INFO] Duplicate file detected (hash={filehash[:10]}...) Skipping processing.") + logger.info(f"[{task_id}] Duplicate file detected (hash={filehash[:10]}...) Skipping processing.") + log_task_progress(task_id, "process_document", "success", "Duplicate file detected, skipping", file_id=existing.id) return { "status": "duplicate_file", "file_id": existing.id, @@ -53,6 +66,8 @@ def process_document(original_local_file: str): } # Not a duplicate -> insert a new record + logger.info(f"[{task_id}] Creating new file record in database") + log_task_progress(task_id, "create_file_record", "in_progress", "Creating file record") new_record = FileRecord( filehash=filehash, original_filename=original_filename, @@ -63,6 +78,8 @@ def process_document(original_local_file: str): db.add(new_record) db.commit() db.refresh(new_record) + logger.info(f"[{task_id}] File record created with ID: {new_record.id}") + log_task_progress(task_id, "create_file_record", "success", f"File record ID: {new_record.id}", file_id=new_record.id) # 1. Generate a UUID-based filename and place it in /workdir/tmp file_ext = os.path.splitext(original_local_file)[1] @@ -73,14 +90,19 @@ def process_document(original_local_file: str): os.makedirs(tmp_dir, exist_ok=True) new_local_path = os.path.join(tmp_dir, new_filename) + logger.info(f"[{task_id}] Copying file to: {new_local_path}") + log_task_progress(task_id, "copy_file", "in_progress", f"Copying file to {new_filename}", file_id=new_record.id) # Copy the file instead of moving it shutil.copy(original_local_file, new_local_path) + log_task_progress(task_id, "copy_file", "success", f"File copied to {new_filename}", file_id=new_record.id) # Update the DB with final local filename new_record.local_filename = new_local_path db.commit() # 2. Check for embedded text (outside the DB session to avoid long open transactions) + logger.info(f"[{task_id}] Checking for embedded text in PDF") + log_task_progress(task_id, "check_text", "in_progress", "Checking for embedded text", file_id=new_record.id) with open(new_local_path, 'rb') as file: pdf_reader = PyPDF2.PdfReader(file) has_text = False @@ -90,19 +112,30 @@ def process_document(original_local_file: str): break if has_text: - print(f"[INFO] PDF {original_local_file} contains embedded text. Processing locally.") + logger.info(f"[{task_id}] PDF {original_local_file} contains embedded text. Processing locally.") + log_task_progress(task_id, "check_text", "success", "Embedded text found, extracting locally", file_id=new_record.id) # Extract text locally + logger.info(f"[{task_id}] Extracting text from PDF") + log_task_progress(task_id, "extract_text", "in_progress", "Extracting text locally", file_id=new_record.id) extracted_text = "" with open(new_local_path, 'rb') as file: pdf_reader = PyPDF2.PdfReader(file) for page in pdf_reader.pages: extracted_text += page.extract_text() + "\n" + + logger.info(f"[{task_id}] Extracted {len(extracted_text)} characters") + log_task_progress(task_id, "extract_text", "success", f"Extracted {len(extracted_text)} characters", file_id=new_record.id) # Call metadata extraction directly - extract_metadata_with_gpt.delay(new_filename, extracted_text) - return {"file": new_local_path, "status": "Text extracted locally"} + logger.info(f"[{task_id}] Queueing metadata extraction") + log_task_progress(task_id, "process_document", "success", "Queued for metadata extraction", file_id=new_record.id) + extract_metadata_with_gpt.delay(new_filename, extracted_text, new_record.id) + return {"file": new_local_path, "status": "Text extracted locally", "file_id": new_record.id} # 3. If no embedded text, queue Azure Document Intelligence processing - process_with_azure_document_intelligence.delay(new_filename) - return {"file": new_local_path, "status": "Queued for OCR"} + logger.info(f"[{task_id}] No embedded text found. Queueing Azure Document Intelligence processing") + log_task_progress(task_id, "check_text", "success", "No embedded text, queuing OCR", file_id=new_record.id) + log_task_progress(task_id, "process_document", "success", "Queued for OCR processing", file_id=new_record.id) + process_with_azure_document_intelligence.delay(new_filename, new_record.id) + return {"file": new_local_path, "status": "Queued for OCR", "file_id": new_record.id} diff --git a/app/tasks/process_with_azure_document_intelligence.py b/app/tasks/process_with_azure_document_intelligence.py index 1dbed541..dd799ccb 100644 --- a/app/tasks/process_with_azure_document_intelligence.py +++ b/app/tasks/process_with_azure_document_intelligence.py @@ -76,7 +76,7 @@ def check_page_rotation(result, filename): return rotation_data @celery.task(base=BaseTaskWithRetry) -def process_with_azure_document_intelligence(filename: str): +def process_with_azure_document_intelligence(filename: str, file_id: int = None): """ Processes a PDF document using Azure Document Intelligence and overlays OCR text onto the local temporary file (stored under /tmp). @@ -88,6 +88,10 @@ def process_with_azure_document_intelligence(filename: str): 3. Saves the OCR-processed PDF locally in the same location as before. 4. Checks for page rotation and triggers page rotation if needed. 5. Triggers downstream metadata extraction. + + Args: + filename: Name of the file to process + file_id: Optional file ID to pass through to subsequent tasks """ try: tmp_file_path = os.path.join(settings.workdir, "tmp", filename) @@ -139,7 +143,7 @@ def process_with_azure_document_intelligence(filename: str): logger.info(f"Extracted text for {filename}: {len(extracted_text)} characters") # Trigger page rotation task if rotation is detected, otherwise proceed to metadata extraction - rotate_pdf_pages.delay(filename, extracted_text, rotation_data) + rotate_pdf_pages.delay(filename, extracted_text, rotation_data, file_id) return {"file": filename, "searchable_pdf": searchable_pdf_path, "cleaned_text": extracted_text} except Exception as e: diff --git a/app/tasks/rotate_pdf_pages.py b/app/tasks/rotate_pdf_pages.py index 6ac38d4e..f96bb5c0 100644 --- a/app/tasks/rotate_pdf_pages.py +++ b/app/tasks/rotate_pdf_pages.py @@ -47,7 +47,7 @@ def determine_rotation_angle(detected_angle): return rotation_value @celery.task(base=BaseTaskWithRetry) -def rotate_pdf_pages(filename: str, extracted_text: str, rotation_data=None): +def rotate_pdf_pages(filename: str, extracted_text: str, rotation_data=None, file_id: int = None): """ Rotates pages in a PDF document based on detected rotation angles. @@ -55,6 +55,7 @@ def rotate_pdf_pages(filename: str, extracted_text: str, rotation_data=None): filename: The name of the file to rotate extracted_text: The extracted text from the document rotation_data: Optional rotation data dictionary {page_index: angle} + file_id: Optional file ID to pass through to subsequent tasks """ try: pdf_path = os.path.join(settings.workdir, "tmp", filename) @@ -64,7 +65,7 @@ def rotate_pdf_pages(filename: str, extracted_text: str, rotation_data=None): # Skip rotation if no rotation data provided if not rotation_data: logger.info(f"No rotation data provided for {filename}, proceeding with metadata extraction") - extract_metadata_with_gpt.delay(filename, extracted_text) + extract_metadata_with_gpt.delay(filename, extracted_text, file_id) return {"file": filename, "status": "no_rotation_needed"} # Standardize rotation_data keys to integers @@ -77,7 +78,7 @@ def rotate_pdf_pages(filename: str, extracted_text: str, rotation_data=None): if not any(abs(angle) > 0 for angle in normalized_rotation_data.values()): logger.info(f"No significant rotations detected in {filename}, proceeding with metadata extraction") - extract_metadata_with_gpt.delay(filename, extracted_text) + extract_metadata_with_gpt.delay(filename, extracted_text, file_id) return {"file": filename, "status": "no_rotation_needed"} logger.info(f"Rotating {len(normalized_rotation_data)} pages in {filename}") @@ -117,7 +118,7 @@ def rotate_pdf_pages(filename: str, extracted_text: str, rotation_data=None): logger.info(f"Detected rotations in {filename} but no rotations were actually applied (angles too small or not multiples of 90°)") # Continue with metadata extraction - extract_metadata_with_gpt.delay(filename, extracted_text) + extract_metadata_with_gpt.delay(filename, extracted_text, file_id) return { "file": filename, @@ -129,5 +130,5 @@ def rotate_pdf_pages(filename: str, extracted_text: str, rotation_data=None): except Exception as e: logger.error(f"Error rotating PDF {filename}: {e}") # Continue with metadata extraction despite rotation failure - extract_metadata_with_gpt.delay(filename, extracted_text) + extract_metadata_with_gpt.delay(filename, extracted_text, file_id) return {"file": filename, "status": "rotation_failed", "error": str(e)} diff --git a/app/tasks/send_to_all.py b/app/tasks/send_to_all.py index ba9eab89..33a9774f 100644 --- a/app/tasks/send_to_all.py +++ b/app/tasks/send_to_all.py @@ -16,6 +16,9 @@ from app.tasks.upload_to_onedrive import upload_to_onedrive from app.tasks.upload_to_s3 import upload_to_s3 from app.utils.config_validator import get_provider_status 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__) @@ -104,8 +107,8 @@ def get_configured_services_from_validator(): return result -@celery.task(base=BaseTaskWithRetry) -def send_to_all_destinations(file_path: str, use_validator=True): +@celery.task(base=BaseTaskWithRetry, bind=True) +def send_to_all_destinations(self, file_path: str, use_validator=True, file_id: int = None): """ Distribute a file to all configured storage destinations. @@ -113,11 +116,29 @@ def send_to_all_destinations(file_path: str, use_validator=True): 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 """ + 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"Sending {file_path} to all configured destinations") + 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 @@ -179,12 +200,13 @@ def send_to_all_destinations(file_path: str, use_validator=True): if use_validator: try: configured_services = get_configured_services_from_validator() - logger.info(f"Configured services according to validator: {configured_services}") + logger.info(f"[{task_id}] Configured services according to validator: {configured_services}") except Exception as e: - logger.warning(f"Failed to get configuration from validator: {str(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"] @@ -192,24 +214,31 @@ def send_to_all_destinations(file_path: str, use_validator=True): is_configured = False if use_validator and service_name in configured_services: is_configured = configured_services[service_name] - logger.debug(f"{service_name} configuration from validator: {is_configured}") + logger.debug(f"[{task_id}] {service_name} configuration from validator: {is_configured}") else: try: is_configured = service["should_upload"]() - logger.debug(f"{service_name} configuration from function: {is_configured}") + logger.debug(f"[{task_id}] {service_name} configuration from function: {is_configured}") except Exception as e: - logger.error(f"Error checking configuration for {service_name}: {str(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"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) 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) except Exception as e: - logger.error(f"Failed to queue {service_name} task: {str(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", 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", diff --git a/frontend/templates/files.html b/frontend/templates/files.html index afe8f4bb..c421fcc3 100644 --- a/frontend/templates/files.html +++ b/frontend/templates/files.html @@ -43,6 +43,20 @@ .delete-btn:hover { background-color: #fed7d7; } + .view-logs-btn { + color: #3182ce; + cursor: pointer; + padding: 0.25rem 0.5rem; + border-radius: 0.25rem; + background: none; + border: none; + display: flex; + align-items: center; + margin-right: 0.5rem; + } + .view-logs-btn:hover { + background-color: #bee3f8; + } .error-message { background-color: #FEE2E2; border: 1px solid #F87171; @@ -76,8 +90,10 @@ background-color: white; border-radius: 0.5rem; padding: 2rem; - max-width: 500px; + max-width: 800px; width: 90%; + max-height: 80vh; + overflow-y: auto; box-shadow: 0 4px 6px rgba(0, 0, 0, 0.1); } .modal-title { @@ -110,6 +126,48 @@ .modal-btn-delete:hover { background-color: #c53030; } + + /* Logs styles */ + .logs-container { + margin-top: 1rem; + } + .log-entry { + padding: 0.75rem; + border-left: 3px solid #e2e8f0; + margin-bottom: 0.5rem; + background-color: #f7fafc; + border-radius: 0.25rem; + } + .log-entry.success { + border-left-color: #48bb78; + background-color: #f0fff4; + } + .log-entry.failure { + border-left-color: #f56565; + background-color: #fff5f5; + } + .log-entry.in_progress { + border-left-color: #4299e1; + background-color: #ebf8ff; + } + .log-step { + font-weight: 600; + color: #2d3748; + } + .log-message { + color: #4a5568; + margin-top: 0.25rem; + } + .log-timestamp { + font-size: 0.875rem; + color: #718096; + margin-top: 0.25rem; + } + .no-logs { + text-align: center; + padding: 2rem; + color: #718096; + } {% endblock %} @@ -147,9 +205,14 @@ {{ file.mime_type }} {{ file.created_at }} - +
+ + +
{% else %} @@ -173,32 +236,104 @@ + + +