From 0d3687253980f84f635c5020f49634758868d599 Mon Sep 17 00:00:00 2001 From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com> Date: Fri, 6 Feb 2026 22:22:04 +0000 Subject: [PATCH 1/5] Initial plan From 1903dc5bcdd098ca9d918dfe7ffdec5a3ef6782f Mon Sep 17 00:00:00 2001 From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com> Date: Fri, 6 Feb 2026 22:30:18 +0000 Subject: [PATCH 2/5] Add comprehensive processing logging system - Added database logging to all major processing tasks - Created API endpoints for retrieving processing logs - Updated frontend to display processing logs per file - Logging includes: process_document, convert_to_pdf, extract_metadata_with_gpt, embed_metadata_into_pdf, finalize_document_storage, send_to_all_destinations Co-authored-by: christianlouis <361235+christianlouis@users.noreply.github.com> --- app/api/__init__.py | 2 + app/api/logs.py | 164 +++++++++++++++++++++++++ app/tasks/convert_to_pdf.py | 35 ++++-- app/tasks/embed_metadata_into_pdf.py | 51 ++++++-- app/tasks/extract_metadata_with_gpt.py | 39 ++++-- app/tasks/finalize_document_storage.py | 27 +++- app/tasks/process_document.py | 49 ++++++-- app/tasks/send_to_all.py | 46 +++++-- frontend/templates/files.html | 155 +++++++++++++++++++++-- 9 files changed, 509 insertions(+), 59 deletions(-) create mode 100644 app/api/logs.py 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 25545c70..21adda06 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__) def unique_filepath(directory, base_filename, extension=".pdf"): """ @@ -39,8 +45,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): """ Embeds extracted metadata into the PDF's standard metadata fields. The mapping is as follows: @@ -54,14 +60,26 @@ 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)}") + + # Get file_id from database + file_id = None # 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", 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") return {"error": "File not found"} + + 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 # Work on a safe copy in a secure temporary directory original_file = local_file_path @@ -75,7 +93,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: @@ -98,7 +117,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]) @@ -110,17 +130,25 @@ 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. + 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) # After triggering final storage, delete the original file if it is in workdir/tmp. @@ -128,19 +156,20 @@ def embed_metadata_into_pdf(local_file_path: str, extracted_text: str, metadata: 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..bd447d5d 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): """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}") + + # Get file_id from database + file_id = 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 + 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) 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..0ac271c8 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,17 +9,36 @@ 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): """ 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)}") + + # Get file_id from database + file_id = None + with SessionLocal() as db: + # Try to find by the processed file path first + file_record = db.query(FileRecord).filter( + FileRecord.local_filename.like(f"%{os.path.basename(original_file)}%") + ).first() + if file_record: + file_id = file_record.id # 2) Enqueue uploads to all destinations (Dropbox, Nextcloud, Paperless) + 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) return { diff --git a/app/tasks/process_document.py b/app/tasks/process_document.py index fa795600..f5eeb6ea 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 + 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) - return {"file": new_local_path, "status": "Text extracted locally"} + return {"file": new_local_path, "status": "Text extracted locally", "file_id": new_record.id} # 3. If no embedded text, queue Azure Document Intelligence processing + 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) - return {"file": new_local_path, "status": "Queued for OCR"} + return {"file": new_local_path, "status": "Queued for OCR", "file_id": new_record.id} diff --git a/app/tasks/send_to_all.py b/app/tasks/send_to_all.py index ba9eab89..afb3ab06 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): """ Distribute a file to all configured storage destinations. @@ -114,10 +117,25 @@ def send_to_all_destinations(file_path: str, use_validator=True): use_validator: Whether to use the config validator to determine enabled services (if False, falls back to individual checks) """ + 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") 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)}") + + # Get file_id from database + file_id = None + with SessionLocal() as db: + file_record = db.query(FileRecord).filter( + FileRecord.local_filename.like(f"%{os.path.basename(file_path)}%") + ).first() + if file_record: + file_id = file_record.id + results = {} # Define service configurations @@ -179,12 +197,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 +211,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) 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/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 @@ + + +