Merge pull request #432 from christianlouis/copilot/fix-ocr-data-extraction-issue
fix(tasks): resolve files stuck in Pending status despite completed processing
This commit is contained in:
@@ -183,13 +183,6 @@ def process_document(
|
|||||||
|
|
||||||
# Not a duplicate (or deduplication disabled) -> insert a new record
|
# Not a duplicate (or deduplication disabled) -> insert a new record
|
||||||
logger.info(f"[{task_id}] Creating new file record in database")
|
logger.info(f"[{task_id}] Creating new file record in database")
|
||||||
if settings.enable_deduplication and settings.show_deduplication_step:
|
|
||||||
log_task_progress(
|
|
||||||
task_id,
|
|
||||||
"check_for_duplicates",
|
|
||||||
"success",
|
|
||||||
"New file - no duplicates found",
|
|
||||||
)
|
|
||||||
log_task_progress(task_id, "create_file_record", "in_progress", "Creating file record")
|
log_task_progress(task_id, "create_file_record", "in_progress", "Creating file record")
|
||||||
new_record = FileRecord(
|
new_record = FileRecord(
|
||||||
filehash=filehash,
|
filehash=filehash,
|
||||||
@@ -213,6 +206,17 @@ def process_document(
|
|||||||
f"File record ID: {new_record.id}",
|
f"File record ID: {new_record.id}",
|
||||||
file_id=new_record.id,
|
file_id=new_record.id,
|
||||||
)
|
)
|
||||||
|
# Update the check_for_duplicates step now that file_id is available.
|
||||||
|
# This must happen after initialize_file_steps() which creates the
|
||||||
|
# step as "pending". The dedup check already passed at this point.
|
||||||
|
if settings.enable_deduplication:
|
||||||
|
log_task_progress(
|
||||||
|
task_id,
|
||||||
|
"check_for_duplicates",
|
||||||
|
"success",
|
||||||
|
"New file - no duplicates found",
|
||||||
|
file_id=new_record.id,
|
||||||
|
)
|
||||||
|
|
||||||
# 1. Generate a UUID-based filename for storage
|
# 1. Generate a UUID-based filename for storage
|
||||||
file_ext = os.path.splitext(original_local_file)[1]
|
file_ext = os.path.splitext(original_local_file)[1]
|
||||||
@@ -330,6 +334,14 @@ def process_document(
|
|||||||
"Non-PDF file detected, converting to PDF",
|
"Non-PDF file detected, converting to PDF",
|
||||||
file_id=file_id,
|
file_id=file_id,
|
||||||
)
|
)
|
||||||
|
# Mark local text extraction as skipped since the file needs PDF conversion first
|
||||||
|
log_task_progress(
|
||||||
|
task_id,
|
||||||
|
"extract_text",
|
||||||
|
"skipped",
|
||||||
|
"Non-PDF file, text extraction deferred to OCR after conversion",
|
||||||
|
file_id=file_id,
|
||||||
|
)
|
||||||
log_task_progress(
|
log_task_progress(
|
||||||
task_id,
|
task_id,
|
||||||
"process_document",
|
"process_document",
|
||||||
|
|||||||
@@ -145,19 +145,26 @@ def get_files_processing_status(db: Session, file_ids: List[int]) -> Dict[int, D
|
|||||||
has_errors = failed_steps > 0
|
has_errors = failed_steps > 0
|
||||||
|
|
||||||
# Determine overall status
|
# Determine overall status
|
||||||
|
#
|
||||||
|
# The pipeline is dynamic: steps may be skipped, added, or
|
||||||
|
# left "pending" depending on the file type and processing path.
|
||||||
|
# The terminal step is the authoritative completion signal.
|
||||||
|
terminal_step = next((s for s in file_steps if s.step_name == TERMINAL_STEP), None)
|
||||||
|
|
||||||
if has_errors:
|
if has_errors:
|
||||||
status = "failed"
|
status = "failed"
|
||||||
elif in_progress_steps > 0:
|
elif in_progress_steps > 0:
|
||||||
status = "processing"
|
status = "processing"
|
||||||
elif completed_steps + skipped_steps == total_steps:
|
elif completed_steps + skipped_steps == total_steps:
|
||||||
# Only mark as completed if the terminal processing step has been
|
if terminal_step is not None:
|
||||||
# recorded. Without this guard, files where later pipeline steps
|
|
||||||
# have not yet started would be falsely marked as "completed".
|
|
||||||
existing_step_names = {s.step_name for s in file_steps}
|
|
||||||
if TERMINAL_STEP in existing_step_names:
|
|
||||||
status = "completed"
|
status = "completed"
|
||||||
else:
|
else:
|
||||||
status = "pending"
|
status = "pending"
|
||||||
|
elif terminal_step is not None and terminal_step.status == "success":
|
||||||
|
# Terminal step succeeded but some intermediate steps are
|
||||||
|
# still "pending" (dynamic pipeline artifacts). The file
|
||||||
|
# is effectively complete.
|
||||||
|
status = "completed"
|
||||||
else:
|
else:
|
||||||
status = "pending"
|
status = "pending"
|
||||||
|
|
||||||
|
|||||||
@@ -251,19 +251,32 @@ def get_file_overall_status(db: Session, file_id: int) -> Dict:
|
|||||||
has_errors = failed_steps > 0
|
has_errors = failed_steps > 0
|
||||||
|
|
||||||
# Determine overall status
|
# Determine overall status
|
||||||
|
#
|
||||||
|
# The processing pipeline is not strictly linear: steps may be skipped,
|
||||||
|
# repeated, or dynamically added depending on the file (e.g. OCR is
|
||||||
|
# skipped when embedded text is found, local extraction is skipped for
|
||||||
|
# non-PDF files, dedup check may not record its result). Because of
|
||||||
|
# this, we use the terminal step as the authoritative signal that the
|
||||||
|
# pipeline finished successfully, rather than requiring every single
|
||||||
|
# intermediate step to be explicitly marked as success/skipped.
|
||||||
|
terminal_step_obj = next((s for s in steps if s.step_name == TERMINAL_STEP), None)
|
||||||
|
|
||||||
if has_errors:
|
if has_errors:
|
||||||
status = "failed"
|
status = "failed"
|
||||||
elif in_progress_steps > 0:
|
elif in_progress_steps > 0:
|
||||||
status = "processing"
|
status = "processing"
|
||||||
elif completed_steps + skipped_steps == total_steps:
|
elif completed_steps + skipped_steps == total_steps:
|
||||||
# Only mark as completed if the terminal processing step has been recorded.
|
# All steps resolved – completed only if terminal step was recorded.
|
||||||
# Without this guard, files where later pipeline steps have not yet started
|
if terminal_step_obj is not None:
|
||||||
# would be falsely marked as "completed" (e.g. only the first 3 steps ran).
|
|
||||||
existing_step_names = {s.step_name for s in steps}
|
|
||||||
if TERMINAL_STEP in existing_step_names:
|
|
||||||
status = "completed"
|
status = "completed"
|
||||||
else:
|
else:
|
||||||
status = "pending"
|
status = "pending"
|
||||||
|
elif terminal_step_obj is not None and terminal_step_obj.status == "success":
|
||||||
|
# The terminal step succeeded but some intermediate steps are still
|
||||||
|
# "pending" (e.g. check_for_duplicates logged without file_id, or
|
||||||
|
# extract_text not marked when OCR path was taken). The pipeline
|
||||||
|
# is effectively complete.
|
||||||
|
status = "completed"
|
||||||
else:
|
else:
|
||||||
status = "pending"
|
status = "pending"
|
||||||
|
|
||||||
|
|||||||
@@ -474,3 +474,54 @@ class TestFileStatusMissingCoverage:
|
|||||||
result = get_files_processing_status(db_session, [file_record.id])
|
result = get_files_processing_status(db_session, [file_record.id])
|
||||||
|
|
||||||
assert result[file_record.id]["status"] == "completed"
|
assert result[file_record.id]["status"] == "completed"
|
||||||
|
|
||||||
|
def test_get_files_processing_status_completed_with_pending_intermediate(self, db_session):
|
||||||
|
"""Test that terminal step success marks file completed even with pending intermediate steps.
|
||||||
|
|
||||||
|
This tests the dynamic pipeline scenario where check_for_duplicates or
|
||||||
|
extract_text might remain pending because the pipeline skipped them.
|
||||||
|
"""
|
||||||
|
from datetime import datetime
|
||||||
|
from unittest.mock import patch
|
||||||
|
|
||||||
|
from app.models import FileProcessingStep
|
||||||
|
from app.utils.file_status import get_files_processing_status
|
||||||
|
|
||||||
|
file_record = FileRecord(
|
||||||
|
filehash="terminal_fallback",
|
||||||
|
original_filename="fallback.pdf",
|
||||||
|
local_filename="/tmp/fallback.pdf",
|
||||||
|
file_size=100,
|
||||||
|
)
|
||||||
|
db_session.add(file_record)
|
||||||
|
db_session.commit()
|
||||||
|
|
||||||
|
now = datetime.now()
|
||||||
|
# Simulate: check_for_duplicates pending, rest succeeded, OCR skipped
|
||||||
|
for step_name, step_status in [
|
||||||
|
("check_for_duplicates", "pending"),
|
||||||
|
("create_file_record", "success"),
|
||||||
|
("check_text", "success"),
|
||||||
|
("extract_text", "success"),
|
||||||
|
("process_with_ocr", "skipped"),
|
||||||
|
("extract_metadata_with_gpt", "success"),
|
||||||
|
("embed_metadata_into_pdf", "success"),
|
||||||
|
("finalize_document_storage", "success"),
|
||||||
|
("send_to_all_destinations", "success"),
|
||||||
|
]:
|
||||||
|
step = FileProcessingStep(
|
||||||
|
file_id=file_record.id,
|
||||||
|
step_name=step_name,
|
||||||
|
status=step_status,
|
||||||
|
created_at=now,
|
||||||
|
updated_at=now,
|
||||||
|
)
|
||||||
|
db_session.add(step)
|
||||||
|
db_session.commit()
|
||||||
|
|
||||||
|
with patch("app.config.settings") as ms:
|
||||||
|
ms.enable_deduplication = True
|
||||||
|
result = get_files_processing_status(db_session, [file_record.id])
|
||||||
|
|
||||||
|
# Terminal step succeeded → file should be completed despite pending dedup step
|
||||||
|
assert result[file_record.id]["status"] == "completed"
|
||||||
|
|||||||
@@ -281,6 +281,67 @@ class TestStepManager:
|
|||||||
assert status["completed_steps"] == len(MAIN_PROCESSING_STEPS)
|
assert status["completed_steps"] == len(MAIN_PROCESSING_STEPS)
|
||||||
assert status["in_progress_steps"] == 0
|
assert status["in_progress_steps"] == 0
|
||||||
|
|
||||||
|
def test_get_file_overall_status_completed_with_pending_intermediate_steps(self, db_session: Session):
|
||||||
|
"""Test that a file is 'completed' when the terminal step succeeds even if some intermediate steps are pending.
|
||||||
|
|
||||||
|
This covers the scenario where steps like check_for_duplicates or
|
||||||
|
extract_text are left in 'pending' because the dynamic pipeline
|
||||||
|
skipped them without explicitly marking them (e.g. dedup log written
|
||||||
|
before file_id was available, or extract_text not marked for non-PDF).
|
||||||
|
"""
|
||||||
|
file_record = FileRecord(
|
||||||
|
filehash="test_terminal_complete",
|
||||||
|
original_filename="terminal.pdf",
|
||||||
|
local_filename="/tmp/terminal.pdf",
|
||||||
|
file_size=1024,
|
||||||
|
)
|
||||||
|
db_session.add(file_record)
|
||||||
|
db_session.commit()
|
||||||
|
|
||||||
|
initialize_file_steps(db_session, file_record.id)
|
||||||
|
|
||||||
|
# Simulate a pipeline where most steps succeed but check_for_duplicates
|
||||||
|
# stays pending (the exact scenario from the bug report).
|
||||||
|
for step_name in MAIN_PROCESSING_STEPS:
|
||||||
|
if step_name == "check_for_duplicates":
|
||||||
|
continue # Leave this as "pending"
|
||||||
|
update_step_status(db_session, file_record.id, step_name, "success")
|
||||||
|
|
||||||
|
# Also add a dynamically created OCR step as "skipped"
|
||||||
|
update_step_status(db_session, file_record.id, "process_with_ocr", "skipped")
|
||||||
|
|
||||||
|
status = get_file_overall_status(db_session, file_record.id)
|
||||||
|
|
||||||
|
# The terminal step (send_to_all_destinations) succeeded, so the
|
||||||
|
# file should be "completed" despite check_for_duplicates being pending.
|
||||||
|
assert status["status"] == "completed"
|
||||||
|
assert status["has_errors"] is False
|
||||||
|
|
||||||
|
def test_get_file_overall_status_pending_without_terminal_step(self, db_session: Session):
|
||||||
|
"""Test that a file stays 'pending' when the terminal step hasn't been recorded."""
|
||||||
|
file_record = FileRecord(
|
||||||
|
filehash="test_no_terminal",
|
||||||
|
original_filename="no_terminal.pdf",
|
||||||
|
local_filename="/tmp/no_terminal.pdf",
|
||||||
|
file_size=1024,
|
||||||
|
)
|
||||||
|
db_session.add(file_record)
|
||||||
|
db_session.commit()
|
||||||
|
|
||||||
|
initialize_file_steps(db_session, file_record.id)
|
||||||
|
|
||||||
|
# Only first few steps completed - terminal step is still pending
|
||||||
|
update_step_status(db_session, file_record.id, "create_file_record", "success")
|
||||||
|
update_step_status(db_session, file_record.id, "check_text", "success")
|
||||||
|
update_step_status(db_session, file_record.id, "extract_text", "success")
|
||||||
|
|
||||||
|
status = get_file_overall_status(db_session, file_record.id)
|
||||||
|
|
||||||
|
# Terminal step hasn't run yet, so file should remain pending
|
||||||
|
assert status["status"] == "pending"
|
||||||
|
assert status["has_errors"] is False
|
||||||
|
assert status["completed_steps"] == 3
|
||||||
|
|
||||||
def test_get_file_overall_status_failed(self, db_session: Session):
|
def test_get_file_overall_status_failed(self, db_session: Session):
|
||||||
"""Test overall status for a file with failed steps."""
|
"""Test overall status for a file with failed steps."""
|
||||||
file_record = FileRecord(
|
file_record = FileRecord(
|
||||||
|
|||||||
Reference in New Issue
Block a user