From 1bdd4c22dea72f0d1a9d16601d9922c275232acd Mon Sep 17 00:00:00 2001 From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com> Date: Fri, 6 Feb 2026 22:20:44 +0000 Subject: [PATCH 01/10] Initial plan 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 02/10] Initial plan From 91f851238a6adb2c14c166a6cd77b02e12a1664c Mon Sep 17 00:00:00 2001 From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com> Date: Fri, 6 Feb 2026 22:23:14 +0000 Subject: [PATCH 03/10] Replace hardcoded 'tmp' strings with constants in embed_metadata_into_pdf.py Co-authored-by: christianlouis <361235+christianlouis@users.noreply.github.com> --- app/tasks/embed_metadata_into_pdf.py | 10 +++++++--- 1 file changed, 7 insertions(+), 3 deletions(-) diff --git a/app/tasks/embed_metadata_into_pdf.py b/app/tasks/embed_metadata_into_pdf.py index 25545c70..28eb907a 100644 --- a/app/tasks/embed_metadata_into_pdf.py +++ b/app/tasks/embed_metadata_into_pdf.py @@ -12,6 +12,10 @@ from app.tasks.finalize_document_storage import finalize_document_storage # Import the shared Celery instance from app.celery_app import celery +# Directory constants - defined here to avoid hardcoded strings (BAN-B108) +TMP_SUBDIR = "tmp" +PROCESSED_SUBDIR = "processed" + def unique_filepath(directory, base_filename, extension=".pdf"): """ Returns a unique filepath in the specified directory. @@ -56,7 +60,7 @@ def embed_metadata_into_pdf(local_file_path: str, extracted_text: str, metadata: """ # 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)) + 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: @@ -105,7 +109,7 @@ def embed_metadata_into_pdf(local_file_path: str, extracted_text: str, metadata: # Remove any extension and then add .pdf suggested_filename = os.path.splitext(suggested_filename)[0] # Define the final directory based on settings.workdir and ensure it exists. - final_dir = os.path.join(settings.workdir, "processed") + final_dir = os.path.join(settings.workdir, PROCESSED_SUBDIR) os.makedirs(final_dir, exist_ok=True) # Get a unique filepath in case of collisions. final_file_path = unique_filepath(final_dir, suggested_filename, extension=".pdf") @@ -124,7 +128,7 @@ def embed_metadata_into_pdf(local_file_path: str, extracted_text: str, metadata: finalize_document_storage.delay(original_file, final_file_path, metadata) # After triggering final storage, delete the original file if it is in workdir/tmp. - workdir_tmp = os.path.join(settings.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) From 0dd6ab3c76566f0223c0b1cfde27ae68e06e1180 Mon Sep 17 00:00:00 2001 From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com> Date: Fri, 6 Feb 2026 22:24:32 +0000 Subject: [PATCH 04/10] Initial plan From 566079e5e51a6aafd934167a36827b09808d07ba Mon Sep 17 00:00:00 2001 From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com> Date: Fri, 6 Feb 2026 22:24:58 +0000 Subject: [PATCH 05/10] Add clarifying comments explaining the use of constants Co-authored-by: christianlouis <361235+christianlouis@users.noreply.github.com> --- app/tasks/embed_metadata_into_pdf.py | 4 ++++ 1 file changed, 4 insertions(+) diff --git a/app/tasks/embed_metadata_into_pdf.py b/app/tasks/embed_metadata_into_pdf.py index 28eb907a..d35e63c8 100644 --- a/app/tasks/embed_metadata_into_pdf.py +++ b/app/tasks/embed_metadata_into_pdf.py @@ -13,6 +13,10 @@ from app.tasks.finalize_document_storage import finalize_document_storage from app.celery_app import celery # Directory constants - defined here to avoid hardcoded strings (BAN-B108) +# Note: These are application-specific subdirectories within settings.workdir, +# not system temporary directories. The workdir is a configurable path specific +# to this application. For actual temporary file creation, tempfile module is +# used (see line 70: tempfile.NamedTemporaryFile) TMP_SUBDIR = "tmp" PROCESSED_SUBDIR = "processed" 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 06/10] 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 @@ + + + +``` + +### Forms +- Use CSRF protection when needed +- Include proper validation +- Show clear error messages +- Use proper `method` (GET/POST) and `enctype` for file uploads +```html +
+
+ + +
+ +
+``` + +### Accessibility +- Use semantic HTML elements (`nav`, `main`, `article`, `section`) +- Include `alt` text for images +- Use proper heading hierarchy (h1 → h2 → h3) +- Add ARIA labels when needed +- Ensure keyboard navigation works + +### Error Handling +- Display user-friendly error messages +- Use flash messages for feedback +- Show loading states for async operations +```jinja2 +{% with messages = get_flashed_messages(with_categories=true) %} + {% if messages %} + {% for category, message in messages %} +
+ {{ message }} +
+ {% endfor %} + {% endif %} +{% endwith %} +``` + +### URL Generation +- Always use `url_for()` for URLs, never hardcode +- Examples: + - Routes: `{{ url_for('upload_document') }}` + - Static: `{{ url_for('static', path='css/style.css') }}` + - API: `{{ url_for('api_document', document_id=doc.id) }}` + +### Template Variables +- Check if variables exist before using them +- Use filters for formatting +```jinja2 +{% if document %} +

Uploaded: {{ document.created_at|datetime }}

+

Size: {{ document.file_size|filesizeformat }}

+{% else %} +

No document found

+{% endif %} +``` + +### Common Components +- Follow existing patterns for headers, footers, navigation +- Reuse template blocks and includes +- Keep components modular +```jinja2 +{% include 'components/navigation.html' %} +{% include 'components/document_card.html' with document=doc %} +``` + +## UI/UX Guidelines +- Maintain consistent spacing using Tailwind's scale (4, 8, 16, etc.) +- Use the existing color palette from the design +- Ensure mobile responsiveness +- Show loading indicators for long operations +- Provide feedback for user actions (success/error messages) +- Keep the interface clean and minimal + +## Performance +- Optimize images (compress, use appropriate formats) +- Minimize JavaScript bundle size +- Use lazy loading for images when appropriate +- Cache static assets diff --git a/.github/instructions/python-backend.instructions.md b/.github/instructions/python-backend.instructions.md new file mode 100644 index 00000000..3626507b --- /dev/null +++ b/.github/instructions/python-backend.instructions.md @@ -0,0 +1,164 @@ +--- +applyTo: "app/**/*.py" +--- + +# Python Backend Instructions + +These instructions apply to all Python code in the `app/` directory. + +## Code Style +- Use **Black** formatter with 120 character line length +- Use **isort** with Black profile for import organization +- Follow **flake8** rules (ignore E203, W503 as per `.pre-commit-config.yaml`) +- All functions must have type hints for parameters and return values +- Use `from typing import Optional, Dict, List, Any, Union` as needed + +## Import Order (isort with Black profile) +```python +# Standard library imports +import os +from pathlib import Path +from typing import Optional, Dict, List + +# Third-party imports +from fastapi import APIRouter, Depends, HTTPException +from sqlalchemy.orm import Session + +# Local application imports +from app.config import settings +from app.database import get_db +from app.models import Document, User +``` + +## Function Definitions +```python +def process_document( + file_path: Path, + user_id: int, + metadata: Optional[Dict[str, Any]] = None +) -> DocumentMetadata: + """ + Process a document and extract metadata. + + Args: + file_path: Path to the document file + user_id: ID of the user uploading the document + metadata: Optional additional metadata + + Returns: + DocumentMetadata object with extracted information + + Raises: + FileNotFoundError: If file doesn't exist + ProcessingError: If processing fails + """ + pass +``` + +## FastAPI Endpoints +- Use dependency injection for DB sessions and auth +- Return Pydantic models for automatic validation +- Use proper status codes from `fastapi.status` +- Add detailed docstrings for OpenAPI docs +```python +from fastapi import APIRouter, Depends, status +from sqlalchemy.orm import Session + +router = APIRouter(prefix="/api/documents", tags=["documents"]) + +@router.post("/", status_code=status.HTTP_201_CREATED) +async def create_document( + file: UploadFile, + db: Session = Depends(get_db), + current_user: User = Depends(get_current_user) +) -> DocumentResponse: + """Create and process a new document.""" + pass +``` + +## Error Handling +- Use custom exceptions from the application +- Log errors with context using `logging.getLogger(__name__)` +- Return user-friendly error messages +- Never expose internal details in production errors +```python +import logging + +logger = logging.getLogger(__name__) + +try: + result = process_file(file_path) +except FileNotFoundError: + logger.error(f"File not found: {file_path}") + raise HTTPException(status_code=404, detail="File not found") +except Exception as e: + logger.exception(f"Error processing file: {str(e)}") + raise HTTPException(status_code=500, detail="Processing failed") +``` + +## Database Operations +- Use SQLAlchemy ORM, never raw SQL with user input +- Use `get_db()` dependency for sessions +- Always commit in try/except blocks +```python +from sqlalchemy.orm import Session +from app.database import get_db + +def create_document(db: Session, document_data: dict) -> Document: + """Create a new document in the database.""" + db_document = Document(**document_data) + try: + db.add(db_document) + db.commit() + db.refresh(db_document) + return db_document + except Exception as e: + db.rollback() + raise +``` + +## Celery Tasks +- Define in `app/tasks/` directory +- Use descriptive names: `module.action` +- Set retry policies +- Log progress and errors +```python +from celery import shared_task +import logging + +logger = logging.getLogger(__name__) + +@shared_task(bind=True, max_retries=3) +def process_ocr(self, document_id: int) -> Dict[str, Any]: + """Process OCR for a document.""" + try: + # Processing logic + logger.info(f"Processing OCR for document {document_id}") + return {"status": "success"} + except Exception as exc: + logger.exception(f"OCR processing failed for {document_id}") + raise self.retry(exc=exc, countdown=60) +``` + +## Configuration +- All settings in `app/config.py` using Pydantic Settings +- Use environment variables, never hardcode values +- Provide defaults when sensible +```python +from pydantic_settings import BaseSettings + +class Settings(BaseSettings): + openai_api_key: str + max_file_size: int = 10485760 # 10MB default + + class Config: + env_file = ".env" +``` + +## Security +- Never commit secrets +- Validate all user inputs +- Use parameterized queries +- Sanitize file paths +- Check file permissions +- Review SECURITY_AUDIT.md for guidelines diff --git a/.github/instructions/testing.instructions.md b/.github/instructions/testing.instructions.md new file mode 100644 index 00000000..37b3961b --- /dev/null +++ b/.github/instructions/testing.instructions.md @@ -0,0 +1,245 @@ +--- +applyTo: "tests/**/*.py" +--- + +# Testing Instructions + +These instructions apply to all test files in the `tests/` directory. + +## Test Organization +- Mirror the structure of `app/` directory in `tests/` +- Name test files with `test_` prefix (e.g., `test_api.py`) +- Group related tests in classes with `Test` prefix +- Use descriptive test function names: `test___` + +## Pytest Configuration +- Configuration in `pytest.ini` +- Run tests: `pytest -v` +- With coverage: `pytest --cov=app --cov-report=term-missing` +- Run specific markers: `pytest -m unit` or `pytest -m integration` + +## Test Markers +Use pytest markers to categorize tests: +```python +import pytest + +@pytest.mark.unit +def test_document_validation(): + """Test document validation logic.""" + pass + +@pytest.mark.integration +def test_document_upload_api(): + """Test document upload endpoint.""" + pass + +@pytest.mark.slow +def test_large_file_processing(): + """Test processing of large files.""" + pass + +@pytest.mark.requires_external +def test_openai_integration(): + """Test OpenAI API integration.""" + pass +``` + +Available markers: +- `unit` - Unit tests for individual functions/methods +- `integration` - Integration tests for API endpoints and workflows +- `slow` - Tests that take significant time to run +- `security` - Security-related tests +- `requires_external` - Tests requiring external services (OpenAI, Azure, etc.) +- `requires_db` - Tests requiring database +- `requires_redis` - Tests requiring Redis + +## Fixtures +Use pytest fixtures for test setup and teardown: +```python +import pytest +from sqlalchemy import create_engine +from sqlalchemy.orm import sessionmaker +from app.database import Base + +@pytest.fixture +def db_session(): + """Provide a database session for tests.""" + engine = create_engine("sqlite:///:memory:") + Base.metadata.create_all(engine) + Session = sessionmaker(bind=engine) + session = Session() + + yield session + + session.close() + Base.metadata.drop_all(engine) + +@pytest.fixture +def sample_document(): + """Provide a sample document for tests.""" + return { + "filename": "test.pdf", + "content_type": "application/pdf", + "size": 1024 + } +``` + +## API Testing with FastAPI +Use `TestClient` from FastAPI: +```python +from fastapi.testclient import TestClient +from app.main import app + +client = TestClient(app) + +def test_upload_document(): + """Test document upload endpoint.""" + with open("tests/fixtures/sample.pdf", "rb") as f: + response = client.post( + "/api/documents/upload", + files={"file": ("test.pdf", f, "application/pdf")} + ) + + assert response.status_code == 201 + assert "id" in response.json() +``` + +## Async Testing +For async code, use `pytest-asyncio`: +```python +import pytest +import httpx + +@pytest.mark.asyncio +async def test_async_document_processing(): + """Test async document processing.""" + async with httpx.AsyncClient(app=app, base_url="http://test") as client: + response = await client.get("/api/documents/1") + assert response.status_code == 200 +``` + +## Mocking External Services +Always mock external services in tests: +```python +from unittest.mock import Mock, patch + +@pytest.mark.unit +def test_openai_metadata_extraction(mocker): + """Test metadata extraction with mocked OpenAI.""" + mock_response = { + "document_type": "invoice", + "amount": 100.00, + "date": "2024-01-01" + } + + mocker.patch( + "app.utils.openai_client.extract_metadata", + return_value=mock_response + ) + + result = extract_document_metadata("test.pdf") + assert result["document_type"] == "invoice" + +@pytest.mark.unit +def test_azure_ocr_processing(mocker): + """Test OCR with mocked Azure service.""" + mock_text = "Sample extracted text" + + mocker.patch( + "app.utils.azure_client.extract_text", + return_value=mock_text + ) + + result = perform_ocr("test.pdf") + assert result == mock_text +``` + +## Database Testing +```python +@pytest.mark.requires_db +def test_create_document(db_session): + """Test document creation in database.""" + from app.models import Document + + doc = Document( + filename="test.pdf", + user_id=1, + file_path="/tmp/test.pdf" + ) + db_session.add(doc) + db_session.commit() + + assert doc.id is not None + assert doc.filename == "test.pdf" +``` + +## Test Coverage Goals +- Aim for **80% code coverage** for all new code +- Focus on critical paths and error handling +- Test both success and failure scenarios +- Don't test third-party library code + +## Test Structure +Follow the Arrange-Act-Assert pattern: +```python +def test_document_validation(): + """Test that invalid documents are rejected.""" + # Arrange + invalid_document = { + "filename": "", # Empty filename + "size": -1 # Invalid size + } + + # Act + result = validate_document(invalid_document) + + # Assert + assert result.is_valid is False + assert "filename" in result.errors + assert "size" in result.errors +``` + +## Parameterized Tests +Use `pytest.mark.parametrize` for multiple test cases: +```python +@pytest.mark.parametrize("filename,expected", [ + ("document.pdf", True), + ("image.jpg", True), + ("script.exe", False), + ("", False), +]) +def test_allowed_file_types(filename, expected): + """Test file type validation.""" + result = is_allowed_file(filename) + assert result == expected +``` + +## Test Data +- Place test fixtures in `tests/fixtures/` directory +- Use small sample files for testing +- Don't commit large test files +- Clean up test files in teardown + +## Error Testing +Always test error conditions: +```python +def test_missing_file_raises_error(): + """Test that missing files raise appropriate error.""" + with pytest.raises(FileNotFoundError): + process_document("/nonexistent/file.pdf") + +def test_invalid_api_request(): + """Test API error handling.""" + response = client.post("/api/documents/", json={}) + assert response.status_code == 422 # Validation error +``` + +## Best Practices +- Test one thing per test function +- Use descriptive test names +- Keep tests independent (no dependencies between tests) +- Use fixtures for common setup +- Mock external dependencies +- Test edge cases and error conditions +- Keep tests fast (use mocks for slow operations) +- Clean up resources after tests From 2189802ee8c0fdc5b507e27f5e24ede68825ba53 Mon Sep 17 00:00:00 2001 From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com> Date: Sat, 7 Feb 2026 14:02:29 +0000 Subject: [PATCH 10/10] Add database logging to upload tasks - 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> --- app/tasks/send_to_all.py | 2 +- app/tasks/upload_to_dropbox.py | 42 +++++++++++++++++++++++--------- app/tasks/upload_to_nextcloud.py | 37 ++++++++++++++++++++-------- app/tasks/upload_to_paperless.py | 40 ++++++++++++++++++++++++------ 4 files changed, 90 insertions(+), 31 deletions(-) diff --git a/app/tasks/send_to_all.py b/app/tasks/send_to_all.py index 1f284c82..33a9774f 100644 --- a/app/tasks/send_to_all.py +++ b/app/tasks/send_to_all.py @@ -228,7 +228,7 @@ def send_to_all_destinations(self, file_path: str, use_validator=True, file_id: logger.info(f"[{task_id}] Queueing {file_path} for {service_name} upload") log_task_progress(task_id, f"queue_{service_name}", "in_progress", f"Queueing upload to {service_name}", file_id=file_id) try: - task = service["upload_func"].delay(file_path) + task = service["upload_func"].delay(file_path, file_id) results[f"{service_name}_task_id"] = task.id queued_count += 1 log_task_progress(task_id, f"queue_{service_name}", "success", f"Queued for {service_name}", file_id=file_id) diff --git a/app/tasks/upload_to_dropbox.py b/app/tasks/upload_to_dropbox.py index 03415953..d7930115 100644 --- a/app/tasks/upload_to_dropbox.py +++ b/app/tasks/upload_to_dropbox.py @@ -9,6 +9,9 @@ from app.config import settings from app.tasks.retry_config import BaseTaskWithRetry from app.celery_app import celery from app.utils.filename_utils import get_unique_filename, sanitize_filename, extract_remote_path +from app.utils import log_task_progress +from app.database import SessionLocal +from app.models import FileRecord logger = logging.getLogger(__name__) @@ -99,21 +102,31 @@ def get_dropbox_client(): logger.error(f"Error creating Dropbox client: {str(e)}") raise -@celery.task(base=BaseTaskWithRetry) -def upload_to_dropbox(file_path: str): +@celery.task(base=BaseTaskWithRetry, bind=True) +def upload_to_dropbox(self, file_path: str, file_id: int = None): """ Upload a file to Dropbox. + + Args: + file_path: Path to the file to upload + file_id: Optional file ID to associate with logs """ + task_id = self.request.id + logger.info(f"[{task_id}] Starting Dropbox upload: {file_path}") + log_task_progress(task_id, "upload_to_dropbox", "in_progress", f"Uploading to Dropbox: {os.path.basename(file_path)}", file_id=file_id) + if not os.path.exists(file_path): error_msg = f"File not found: {file_path}" - logger.error(error_msg) + logger.error(f"[{task_id}] {error_msg}") + log_task_progress(task_id, "upload_to_dropbox", "failure", error_msg, file_id=file_id) raise FileNotFoundError(error_msg) # Check if Dropbox is properly configured if not (hasattr(settings, 'dropbox_app_key') and settings.dropbox_app_key and hasattr(settings, 'dropbox_app_secret') and settings.dropbox_app_secret and hasattr(settings, 'dropbox_refresh_token') and settings.dropbox_refresh_token): - logger.info("Dropbox upload skipped: Missing configuration") + logger.info(f"[{task_id}] Dropbox upload skipped: Missing configuration") + log_task_progress(task_id, "upload_to_dropbox", "success", "Skipped: Not configured", file_id=file_id) return {"status": "Skipped", "reason": "Dropbox settings not configured"} filename = os.path.basename(file_path) @@ -144,7 +157,8 @@ def upload_to_dropbox(file_path: str): dropbox_path = get_unique_filename(remote_full_path, check_exists_in_dropbox) # Upload the file - logger.info(f"Uploading {filename} to Dropbox at {dropbox_path}") + logger.info(f"[{task_id}] Uploading {filename} to Dropbox at {dropbox_path}") + log_task_progress(task_id, "upload_file", "in_progress", f"Uploading to {dropbox_path}", file_id=file_id) with open(file_path, 'rb') as file_data: # Use files_upload_session for large files to avoid timeouts file_size = os.path.getsize(file_path) @@ -179,7 +193,8 @@ def upload_to_dropbox(file_path: str): mode=dropbox.files.WriteMode.overwrite ) - logger.info(f"Successfully uploaded {filename} to Dropbox at {dropbox_path}") + logger.info(f"[{task_id}] Successfully uploaded {filename} to Dropbox at {dropbox_path}") + log_task_progress(task_id, "upload_to_dropbox", "success", f"Uploaded to Dropbox: {dropbox_path}", file_id=file_id) return { "status": "Completed", "file_path": file_path, @@ -187,14 +202,17 @@ def upload_to_dropbox(file_path: str): } except AuthError: - error_msg = f"[ERROR] Authentication failed while uploading {filename} to Dropbox. Check token." - logger.error(error_msg) + error_msg = f"Authentication failed while uploading {filename} to Dropbox. Check token." + logger.error(f"[{task_id}] {error_msg}") + log_task_progress(task_id, "upload_to_dropbox", "failure", error_msg, file_id=file_id) raise Exception(error_msg) except ApiError as e: - error_msg = f"[ERROR] Failed to upload {filename} to Dropbox: {e}" - logger.error(error_msg) + error_msg = f"Failed to upload {filename} to Dropbox: {e}" + logger.error(f"[{task_id}] {error_msg}") + log_task_progress(task_id, "upload_to_dropbox", "failure", error_msg, file_id=file_id) raise Exception(error_msg) except Exception as e: - error_msg = f"[ERROR] Unexpected error uploading {filename} to Dropbox: {e}" - logger.error(error_msg) + error_msg = f"Unexpected error uploading {filename} to Dropbox: {e}" + logger.error(f"[{task_id}] {error_msg}") + log_task_progress(task_id, "upload_to_dropbox", "failure", error_msg, file_id=file_id) raise Exception(error_msg) diff --git a/app/tasks/upload_to_nextcloud.py b/app/tasks/upload_to_nextcloud.py index 2a3cebc0..3dd98a22 100644 --- a/app/tasks/upload_to_nextcloud.py +++ b/app/tasks/upload_to_nextcloud.py @@ -8,17 +8,29 @@ from app.config import settings from app.celery_app import celery from app.tasks.retry_config import BaseTaskWithRetry from app.utils.filename_utils import get_unique_filename, sanitize_filename, extract_remote_path +from app.utils import log_task_progress +from app.database import SessionLocal +from app.models import FileRecord logger = logging.getLogger(__name__) -@celery.task(base=BaseTaskWithRetry) -def upload_to_nextcloud(file_path: str): +@celery.task(base=BaseTaskWithRetry, bind=True) +def upload_to_nextcloud(self, file_path: str, file_id: int = None): """ Upload a file to Nextcloud WebDAV. + + Args: + file_path: Path to the file to upload + file_id: Optional file ID to associate with logs """ + task_id = self.request.id + logger.info(f"[{task_id}] Starting Nextcloud upload: {file_path}") + log_task_progress(task_id, "upload_to_nextcloud", "in_progress", f"Uploading to Nextcloud: {os.path.basename(file_path)}", file_id=file_id) + if not os.path.exists(file_path): error_msg = f"File not found: {file_path}" - logger.error(error_msg) + logger.error(f"[{task_id}] {error_msg}") + log_task_progress(task_id, "upload_to_nextcloud", "failure", error_msg, file_id=file_id) raise FileNotFoundError(error_msg) # For Nextcloud, we need to check for 'nextcloud_upload_url' instead of 'nextcloud_url' @@ -26,7 +38,8 @@ def upload_to_nextcloud(file_path: str): if not (getattr(settings, 'nextcloud_upload_url', None) and getattr(settings, 'nextcloud_username', None) and getattr(settings, 'nextcloud_password', None)): - logger.info("Nextcloud upload skipped: Missing configuration") + logger.info(f"[{task_id}] Nextcloud upload skipped: Missing configuration") + log_task_progress(task_id, "upload_to_nextcloud", "success", "Skipped: Not configured", file_id=file_id) return {"status": "Skipped", "reason": "Nextcloud settings not configured"} filename = os.path.basename(file_path) @@ -99,7 +112,8 @@ def upload_to_nextcloud(file_path: str): ) # Upload the file - logger.info(f"Uploading {filename} to Nextcloud at {full_url}") + logger.info(f"[{task_id}] Uploading {filename} to Nextcloud at {full_url}") + log_task_progress(task_id, "upload_file", "in_progress", f"Uploading to {remote_path}", file_id=file_id) with open(file_path, 'rb') as file_data: response = requests.put( full_url, @@ -110,7 +124,8 @@ def upload_to_nextcloud(file_path: str): ) if response.status_code in (201, 204): # Created or No Content - logger.info(f"Successfully uploaded {filename} to Nextcloud at {remote_path}") + logger.info(f"[{task_id}] Successfully uploaded {filename} to Nextcloud at {remote_path}") + log_task_progress(task_id, "upload_to_nextcloud", "success", f"Uploaded to Nextcloud: {remote_path}", file_id=file_id) return { "status": "Completed", "file_path": file_path, @@ -118,11 +133,13 @@ def upload_to_nextcloud(file_path: str): "response_code": response.status_code } else: - error_msg = f"[ERROR] Failed to upload {filename} to Nextcloud: {response.status_code} - {response.text}" - logger.error(error_msg) + error_msg = f"Failed to upload {filename} to Nextcloud: {response.status_code} - {response.text}" + logger.error(f"[{task_id}] {error_msg}") + log_task_progress(task_id, "upload_to_nextcloud", "failure", error_msg, file_id=file_id) raise Exception(error_msg) except Exception as e: - error_msg = f"[ERROR] Failed to upload {filename} to Nextcloud: {str(e)}" - logger.error(error_msg) + error_msg = f"Failed to upload {filename} to Nextcloud: {str(e)}" + logger.error(f"[{task_id}] {error_msg}") + log_task_progress(task_id, "upload_to_nextcloud", "failure", error_msg, file_id=file_id) raise Exception(error_msg) diff --git a/app/tasks/upload_to_paperless.py b/app/tasks/upload_to_paperless.py index bd9b5485..dc8af68b 100644 --- a/app/tasks/upload_to_paperless.py +++ b/app/tasks/upload_to_paperless.py @@ -10,6 +10,9 @@ from typing import Dict, Any from app.config import settings from app.tasks.retry_config import BaseTaskWithRetry from app.celery_app import celery +from app.utils import log_task_progress +from app.database import SessionLocal +from app.models import FileRecord logger = logging.getLogger(__name__) @@ -81,12 +84,24 @@ def poll_task_for_document_id(task_id: str) -> int: f"Task {task_id} didn't reach SUCCESS within {POLL_MAX_ATTEMPTS} attempts." ) -@celery.task(base=BaseTaskWithRetry) -def upload_to_paperless(file_path: str): - """Uploads a file to Paperless-ngx.""" +@celery.task(base=BaseTaskWithRetry, bind=True) +def upload_to_paperless(self, file_path: str, file_id: int = None): + """ + Uploads a file to Paperless-ngx. + + Args: + file_path: Path to the file to upload + file_id: Optional file ID to associate with logs + """ + task_id = self.request.id + logger.info(f"[{task_id}] Starting Paperless upload: {file_path}") + log_task_progress(task_id, "upload_to_paperless", "in_progress", f"Uploading to Paperless: {os.path.basename(file_path)}", file_id=file_id) if not os.path.exists(file_path): - raise FileNotFoundError(f"File not found: {file_path}") + error_msg = f"File not found: {file_path}" + logger.error(f"[{task_id}] {error_msg}") + log_task_progress(task_id, "upload_to_paperless", "failure", error_msg, file_id=file_id) + raise FileNotFoundError(error_msg) # Extract filename filename = os.path.basename(file_path) @@ -94,10 +109,13 @@ def upload_to_paperless(file_path: str): # Check if Paperless settings are configured if not settings.paperless_host or not settings.paperless_ngx_api_token: error_msg = "Paperless-ngx credentials are not fully configured" - logger.error(error_msg) + logger.error(f"[{task_id}] {error_msg}") + log_task_progress(task_id, "upload_to_paperless", "failure", error_msg, file_id=file_id) raise ValueError(error_msg) # Upload the PDF + logger.info(f"[{task_id}] Posting document to Paperless") + log_task_progress(task_id, "post_document", "in_progress", "Posting to Paperless API", file_id=file_id) post_url = _paperless_api_url("/api/documents/post_document/") with open(file_path, "rb") as f: files = { @@ -110,18 +128,24 @@ def upload_to_paperless(file_path: str): resp = requests.post(post_url, headers=_get_headers(), files=files, data=data) resp.raise_for_status() except requests.exceptions.RequestException as exc: + error_msg = f"Failed to upload to Paperless: {exc}" logger.error( - "Failed to upload document '%s' to Paperless. Error: %s. Response=%s", + f"[{task_id}] Failed to upload document '%s' to Paperless. Error: %s. Response=%s", file_path, exc, getattr(exc.response, "text", "") ) + log_task_progress(task_id, "upload_to_paperless", "failure", error_msg, file_id=file_id) raise raw_task_id = resp.text.strip().strip('"').strip("'") - logger.info(f"Received Paperless task ID: {raw_task_id}") + logger.info(f"[{task_id}] Received Paperless task ID: {raw_task_id}") + log_task_progress(task_id, "post_document", "success", f"Task ID: {raw_task_id}", file_id=file_id) # Poll tasks until success/fail => get doc_id + logger.info(f"[{task_id}] Polling for document ID") + log_task_progress(task_id, "poll_task", "in_progress", "Waiting for Paperless processing", file_id=file_id) doc_id = poll_task_for_document_id(raw_task_id) - logger.info(f"Document {file_path} successfully ingested => ID={doc_id}") + logger.info(f"[{task_id}] Document {file_path} successfully ingested => ID={doc_id}") + log_task_progress(task_id, "upload_to_paperless", "success", f"Uploaded to Paperless: Doc ID {doc_id}", file_id=file_id) return { "status": "Completed",