diff --git a/app/tasks/process_document.py b/app/tasks/process_document.py index 9eae16d6..d2d7b966 100644 --- a/app/tasks/process_document.py +++ b/app/tasks/process_document.py @@ -233,6 +233,14 @@ def process_document(self, original_local_file: str, original_filename: str = No "Force Cloud OCR requested, queuing OCR", file_id=file_id, ) + # Mark local text extraction as skipped since force_cloud_ocr was requested + log_task_progress( + task_id, + "extract_text", + "skipped", + "Force cloud OCR requested, skipping local extraction", + file_id=file_id, + ) log_task_progress( task_id, "process_document", @@ -292,6 +300,15 @@ def process_document(self, original_local_file: str, original_filename: str = No f"Extracted {len(extracted_text)} characters", file_id=file_id, ) + + # Mark Azure OCR as skipped since we extracted text locally + log_task_progress( + task_id, + "process_with_azure_document_intelligence", + "skipped", + "Local text extraction succeeded, Azure OCR not needed", + file_id=file_id, + ) # Call metadata extraction directly logger.info(f"[{task_id}] Queueing metadata extraction") @@ -318,6 +335,16 @@ def process_document(self, original_local_file: str, original_filename: str = No "No embedded text, queuing OCR", file_id=file_id, ) + + # Mark local text extraction as skipped since we're using Azure OCR + log_task_progress( + task_id, + "extract_text", + "skipped", + "No embedded text, using Azure OCR instead", + file_id=file_id, + ) + log_task_progress( task_id, "process_document", diff --git a/app/tasks/send_to_all.py b/app/tasks/send_to_all.py index d69fc8cd..330a2ef3 100644 --- a/app/tasks/send_to_all.py +++ b/app/tasks/send_to_all.py @@ -18,8 +18,8 @@ from app.tasks.upload_to_paperless import upload_to_paperless from app.tasks.upload_to_s3 import upload_to_s3 from app.tasks.upload_to_sftp import upload_to_sftp from app.tasks.upload_to_webdav import upload_to_webdav -from app.utils import log_task_progress from app.utils.config_validator import get_provider_status +from app.utils.logging import log_task_progress logger = logging.getLogger(__name__) diff --git a/app/utils/file_queries.py b/app/utils/file_queries.py index a6af8b26..5c0457e7 100644 --- a/app/utils/file_queries.py +++ b/app/utils/file_queries.py @@ -2,7 +2,7 @@ Shared file query utilities for filtering files by processing status. This module contains reusable query logic for filtering FileRecord objects -based on their processing status (pending, processing, failed, completed). +based on their processing status using the FileProcessingStep table. """ from typing import Optional @@ -10,24 +10,33 @@ from typing import Optional from sqlalchemy import or_ from sqlalchemy.orm import Query, Session -from app.models import FileRecord, ProcessingLog +from app.models import FileRecord, FileProcessingStep def apply_status_filter(query: Query, db: Session, status: Optional[str]) -> Query: """ - Apply status filter to a FileRecord query. + Apply status filter to a FileRecord query using FileProcessingStep table. This function modifies a SQLAlchemy query to filter files based on their - processing status by examining associated ProcessingLog entries. + processing status by examining associated FileProcessingStep entries. + + Only tracks "real" processing steps that represent user-facing status: + - Main steps: create_file_record, check_text, extract_text, process_with_azure_document_intelligence, + extract_metadata_with_gpt, embed_metadata_into_pdf, finalize_document_storage, + send_to_all_destinations + - Upload steps: queue_*, upload_to_* + + Diagnostic/internal steps (poll_task, upload_file, set_custom_fields, etc.) are ignored + as they may not complete properly and don't affect the actual status. Args: query: The base SQLAlchemy query for FileRecord objects db: Database session for creating subqueries status: Status filter to apply. Valid values: - - "pending": Files with no ProcessingLog entries - - "processing": Files with in_progress logs - - "failed": Files with failure logs - - "completed": Files with success logs but no failures or in_progress + - "pending": Files with no real FileProcessingStep entries + - "processing": Files with in_progress real steps + - "failed": Files with failure real steps + - "completed": Files with all real steps success/skipped - None: No filter applied (returns query unchanged) Returns: @@ -41,32 +50,64 @@ def apply_status_filter(query: Query, db: Session, status: Optional[str]) -> Que if not status: return query - # Subquery to get file IDs matching the status - if status == "pending": - # Files with no logs - subq = db.query(ProcessingLog.file_id).distinct() - query = query.filter(~FileRecord.id.in_(subq)) - elif status == "processing": - # Files with in_progress logs - subq = db.query(ProcessingLog.file_id).filter(ProcessingLog.status == "in_progress").distinct() - query = query.filter(FileRecord.id.in_(subq)) - elif status == "failed": - # Files with failure logs - subq = db.query(ProcessingLog.file_id).filter(ProcessingLog.status == "failure").distinct() - query = query.filter(FileRecord.id.in_(subq)) - elif status == "completed": - # Files with success logs but no failures or in_progress - success_files = db.query(ProcessingLog.file_id).filter(ProcessingLog.status == "success").distinct().subquery() + # Define which steps are "real" status-determining steps + # Only high-level logical steps and actual upload destinations (not queue_* steps) + REAL_STEPS = { + "create_file_record", + "check_text", + "extract_text", + "process_with_azure_document_intelligence", + "extract_metadata_with_gpt", + "embed_metadata_into_pdf", + "finalize_document_storage", + "send_to_all_destinations", + "upload_to_dropbox", + "upload_to_paperless", + "upload_to_google_drive", + "upload_to_ftp", + "upload_to_onedrive", + "upload_to_webdav", + "upload_to_sftp", + "upload_to_nextcloud", + "upload_to_paperless_ngx", + "upload_to_email", + "upload_to_s3", + } - failed_files = ( - db.query(ProcessingLog.file_id) - .filter(or_(ProcessingLog.status == "failure", ProcessingLog.status == "in_progress")) + # Filter to only real steps + real_steps_subq = ( + db.query(FileProcessingStep) + .filter(FileProcessingStep.step_name.in_(REAL_STEPS)) + ) + + if status == "pending": + # Files with no real steps (never started processing) + subq = real_steps_subq.distinct().subquery() + query = query.filter(~FileRecord.id.in_(db.query(subq.c.file_id))) + elif status == "processing": + # Files with in_progress real steps + subq = real_steps_subq.filter(FileProcessingStep.status == "in_progress").distinct().subquery() + query = query.filter(FileRecord.id.in_(db.query(subq.c.file_id))) + elif status == "failed": + # Files with failure real steps + subq = real_steps_subq.filter(FileProcessingStep.status == "failure").distinct().subquery() + query = query.filter(FileRecord.id.in_(db.query(subq.c.file_id))) + elif status == "completed": + # Files where all real steps are either success or skipped (no failures or in_progress) + # Get files that have real steps + files_with_real_steps = real_steps_subq.distinct().subquery() + + # Get files with failures or in_progress on real steps + files_with_issues = ( + real_steps_subq + .filter(or_(FileProcessingStep.status == "failure", FileProcessingStep.status == "in_progress")) .distinct() .subquery() ) - query = query.filter(FileRecord.id.in_(db.query(success_files.c.file_id))).filter( - ~FileRecord.id.in_(db.query(failed_files.c.file_id)) + # Select files with real steps that don't have issues + query = query.filter(FileRecord.id.in_(db.query(files_with_real_steps.c.file_id))).filter( + ~FileRecord.id.in_(db.query(files_with_issues.c.file_id)) ) return query diff --git a/app/utils/file_status.py b/app/utils/file_status.py index bb683303..bfdca39a 100644 --- a/app/utils/file_status.py +++ b/app/utils/file_status.py @@ -46,6 +46,14 @@ def get_files_processing_status(db: Session, file_ids: List[int]) -> Dict[int, D """ Get processing status for multiple files efficiently. + Only counts "real" processing steps that represent user-facing status: + - Main steps: create_file_record, check_text, extract_text, process_with_azure_document_intelligence, + extract_metadata_with_gpt, embed_metadata_into_pdf, finalize_document_storage, + send_to_all_destinations + - Upload steps: upload_to_* + + Diagnostic/internal steps (poll_task, upload_file, set_custom_fields, etc.) are ignored. + Args: db: Database session file_ids: List of file IDs @@ -53,8 +61,35 @@ def get_files_processing_status(db: Session, file_ids: List[int]) -> Dict[int, D Returns: dict mapping file_id to status dict """ - # Get all steps for these files in one query - steps = db.query(FileProcessingStep).filter(FileProcessingStep.file_id.in_(file_ids)).all() + # Define which steps are "real" status-determining steps + REAL_STEPS = { + "create_file_record", + "check_text", + "extract_text", + "process_with_azure_document_intelligence", + "extract_metadata_with_gpt", + "embed_metadata_into_pdf", + "finalize_document_storage", + "send_to_all_destinations", + "upload_to_dropbox", + "upload_to_paperless", + "upload_to_google_drive", + "upload_to_ftp", + "upload_to_onedrive", + "upload_to_webdav", + "upload_to_sftp", + "upload_to_nextcloud", + "upload_to_paperless_ngx", + "upload_to_email", + "upload_to_s3", + } + + # Get all REAL steps for these files in one query + steps = ( + db.query(FileProcessingStep) + .filter(FileProcessingStep.file_id.in_(file_ids), FileProcessingStep.step_name.in_(REAL_STEPS)) + .all() + ) # Group steps by file_id steps_by_file = {} diff --git a/app/utils/logging.py b/app/utils/logging.py index 57db13f7..0d3245ac 100644 --- a/app/utils/logging.py +++ b/app/utils/logging.py @@ -1,9 +1,10 @@ import logging import threading from collections import defaultdict +from datetime import datetime from app.database import SessionLocal -from app.models import ProcessingLog +from app.models import FileProcessingStep, ProcessingLog class TaskLogCollector(logging.Handler): @@ -66,6 +67,8 @@ def log_task_progress(task_id, step_name, status, message=None, file_id=None, de If no explicit detail is provided, automatically drains any buffered worker log output for this task ID and stores it as the detail. + Also updates the FileProcessingStep table for definitive status tracking. + Args: task_id: The Celery task ID step_name: Name of the processing step @@ -83,6 +86,7 @@ def log_task_progress(task_id, step_name, status, message=None, file_id=None, de detail = collected with SessionLocal() as db: + # Log to ProcessingLog (for historical viewing) log_entry = ProcessingLog( task_id=task_id, step_name=step_name, @@ -92,4 +96,37 @@ def log_task_progress(task_id, step_name, status, message=None, file_id=None, de detail=detail, ) db.add(log_entry) + + # Update FileProcessingStep table (for status tracking) if file_id is provided + if file_id and step_name: + # Find or create the step record + step_record = ( + db.query(FileProcessingStep) + .filter(FileProcessingStep.file_id == file_id, FileProcessingStep.step_name == step_name) + .first() + ) + + now = datetime.utcnow() + + if not step_record: + # Create new step record + step_record = FileProcessingStep( + file_id=file_id, + step_name=step_name, + status=status, + started_at=now if status == "in_progress" else None, + completed_at=now if status in ("success", "failure", "skipped") else None, + error_message=message if status == "failure" else None, + ) + db.add(step_record) + else: + # Update existing step record + step_record.status = status + if status == "in_progress" and not step_record.started_at: + step_record.started_at = now + if status in ("success", "failure", "skipped"): + step_record.completed_at = now + if status == "failure": + step_record.error_message = message or detail + db.commit() diff --git a/app/utils/step_manager.py b/app/utils/step_manager.py index fc5b58f5..621a46bd 100644 --- a/app/utils/step_manager.py +++ b/app/utils/step_manager.py @@ -156,6 +156,9 @@ def get_file_step_status(db: Session, file_id: int) -> Dict[str, Dict]: def get_file_overall_status(db: Session, file_id: int) -> Dict: """ Get the overall processing status for a file based on its steps. + + Only considers "real" processing steps that represent user-facing status. + Ignores diagnostic/internal steps like poll_task, upload_file, etc. Args: db: Database session @@ -172,7 +175,28 @@ def get_file_overall_status(db: Session, file_id: int) -> Dict: "in_progress_steps": 2 } """ - steps = db.query(FileProcessingStep).filter(FileProcessingStep.file_id == file_id).all() + # Define which steps are "real" status-determining steps + # Only high-level logical steps, not implementation sub-steps + REAL_MAIN_STEPS = { + "create_file_record", + "check_text", + "extract_text", + "process_with_azure_document_intelligence", + "extract_metadata_with_gpt", + "embed_metadata_into_pdf", + "finalize_document_storage", + "send_to_all_destinations", + } + + all_steps = db.query(FileProcessingStep).filter(FileProcessingStep.file_id == file_id).all() + + # Filter to only real steps + steps = [ + s for s in all_steps + if s.step_name in REAL_MAIN_STEPS + or s.step_name.startswith("queue_") + or s.step_name.startswith("upload_to_") + ] if not steps: return { @@ -217,6 +241,9 @@ def get_step_summary(db: Session, file_id: int) -> Dict: """ Get a summary of main steps vs upload steps with status counts. + Only counts "real" processing steps that represent user-facing status. + Ignores diagnostic/internal steps like poll_task, upload_file, etc. + Args: db: Database session file_id: ID of the file @@ -230,13 +257,24 @@ def get_step_summary(db: Session, file_id: int) -> Dict: "total_upload_tasks": 6 } """ + # Define which steps are "real" status-determining steps + # Only high-level logical steps, not implementation sub-steps + REAL_MAIN_STEPS = { + "create_file_record", + "check_text", + "extract_text", + "process_with_azure_document_intelligence", + "extract_metadata_with_gpt", + "embed_metadata_into_pdf", + "finalize_document_storage", + "send_to_all_destinations", + } + steps = db.query(FileProcessingStep).filter(FileProcessingStep.file_id == file_id).all() main_counts = {"queued": 0, "in_progress": 0, "success": 0, "failure": 0, "skipped": 0} upload_counts = {"queued": 0, "in_progress": 0, "success": 0, "failure": 0, "skipped": 0} - upload_prefixes = ["upload_to_", "queue_"] - main_steps_count = 0 upload_steps_count = 0 @@ -246,14 +284,21 @@ def get_step_summary(db: Session, file_id: int) -> Dict: if status == "pending": status = "queued" - # Check if it's an upload task - is_upload = any(step.step_name.startswith(prefix) for prefix in upload_prefixes) + # Check if it's an upload task (only count actual upload_to_* steps, not queue_* steps) + is_upload = step.step_name.startswith("upload_to_") + + # Only count "real" steps + is_real_step = step.step_name in REAL_MAIN_STEPS or is_upload or step.step_name.startswith("queue_") + + if not is_real_step: + # Skip diagnostic/internal steps like poll_task, upload_file, set_custom_fields, etc. + continue if is_upload: if status in upload_counts: upload_counts[status] += 1 upload_steps_count += 1 - elif step.step_name in MAIN_PROCESSING_STEPS: + elif step.step_name in REAL_MAIN_STEPS: if status in main_counts: main_counts[status] += 1 main_steps_count += 1 @@ -264,3 +309,4 @@ def get_step_summary(db: Session, file_id: int) -> Dict: "total_main_steps": main_steps_count, "total_upload_tasks": upload_steps_count, } + diff --git a/frontend/templates/file_detail.html b/frontend/templates/file_detail.html index 58225f9d..65008692 100644 --- a/frontend/templates/file_detail.html +++ b/frontend/templates/file_detail.html @@ -919,6 +919,62 @@ {% else %} + + {% if step_summary %} +
+
+
+
Status
+
+ {% set main_completed = step_summary.main.success + step_summary.main.skipped %} + {% set uploads_completed = step_summary.uploads.success + step_summary.uploads.skipped %} + {% if step_summary.main.failure > 0 or step_summary.uploads.failure > 0 %} + Failed + {% elif step_summary.main["in_progress"] > 0 or step_summary.uploads["in_progress"] > 0 %} + Processing + {% elif step_summary.total_main_steps > 0 and main_completed == step_summary.total_main_steps and step_summary.main.failure == 0 %} + Completed + {% else %} + Pending + {% endif %} +
+
+
+
+
+
Main Steps
+
+ {% set main_completed = step_summary.main.success + step_summary.main.skipped %} + {% if main_completed == step_summary.total_main_steps and step_summary.main.failure == 0 %} + ✓ {{ main_completed }}/{{ step_summary.total_main_steps }} + {% else %} + {{ main_completed }}/{{ step_summary.total_main_steps }} + {% endif %} +
+
+
+
Uploads
+
+ {% set uploads_completed = step_summary.uploads.success + step_summary.uploads.skipped %} + {% if uploads_completed == step_summary.total_upload_tasks and step_summary.uploads.failure == 0 and step_summary.total_upload_tasks > 0 %} + ✓ {{ uploads_completed }}/{{ step_summary.total_upload_tasks }} + {% else %} + {{ uploads_completed }}/{{ step_summary.total_upload_tasks }} + {% endif %} +
+
+
+
+
+
+ + {% endif %} +

File Information

@@ -1189,9 +1245,10 @@
Main Processing Steps
- {% if step_summary.main.success > 0 %} + {% set main_completed = step_summary.main.success + step_summary.main.skipped %} + {% if main_completed > 0 %}
- {{ step_summary.main.success }} + {{ main_completed }} Success
{% endif %} @@ -1220,9 +1277,10 @@
Upload Destinations
- {% if step_summary.uploads.success > 0 %} + {% set uploads_completed = step_summary.uploads.success + step_summary.uploads.skipped %} + {% if uploads_completed > 0 %}
- {{ step_summary.uploads.success }} + {{ uploads_completed }} Success
{% endif %}