Merge pull request #294 from christianlouis/copilot/fix-reprocessing-metadata-embed

Fix retry failure when file moved to processed directory
This commit is contained in:
Christian Krakau-Louis
2026-02-13 22:17:26 +01:00
committed by GitHub
3 changed files with 85 additions and 10 deletions
+26 -5
View File
@@ -525,13 +525,34 @@ def _retry_pipeline_step(file_record: FileRecord, step_name: str, db: Session) -
# Retrying embed requires re-running metadata extraction first, because # Retrying embed requires re-running metadata extraction first, because
# embed_metadata_into_pdf needs the actual metadata dict (not empty). # embed_metadata_into_pdf needs the actual metadata dict (not empty).
# Re-trigger extract_metadata_with_gpt which will chain into embed_metadata_into_pdf. # Re-trigger extract_metadata_with_gpt which will chain into embed_metadata_into_pdf.
if not file_record.local_filename or not os.path.exists(file_record.local_filename):
# Check for file in multiple locations:
# 1. Original location in tmp (file_record.local_filename)
# 2. Processed location (file_record.processed_file_path)
# 3. Fallback to workdir/tmp/<basename>
file_path = None
if file_record.local_filename and os.path.exists(file_record.local_filename):
file_path = file_record.local_filename
elif file_record.processed_file_path and os.path.exists(file_record.processed_file_path):
# File has been processed and moved to processed directory
file_path = file_record.processed_file_path
# Try fallback path in workdir/tmp
elif file_record.local_filename:
workdir = settings.workdir
tmp_dir = os.path.join(workdir, "tmp")
fallback_path = os.path.join(tmp_dir, os.path.basename(file_record.local_filename))
if os.path.exists(fallback_path):
file_path = fallback_path
if not file_path:
raise HTTPException( raise HTTPException(
status_code=400, detail="Local file not found on disk. Cannot retry metadata embedding." status_code=400,
detail="File not found in tmp or processed directory. Cannot retry metadata embedding."
) )
extracted_text = _extract_text_from_pdf(file_record.local_filename)
filename = os.path.basename(file_record.local_filename) extracted_text = _extract_text_from_pdf(file_path)
task = extract_metadata_task.delay(filename, extracted_text, file_id) # Pass the full path to the task so it can locate the file
task = extract_metadata_task.delay(file_path, extracted_text, file_id)
else: else:
raise HTTPException(status_code=400, detail=f"Unsupported pipeline step: {step_name}") raise HTTPException(status_code=400, detail=f"Unsupported pipeline step: {step_name}")
+16 -5
View File
@@ -47,17 +47,28 @@ def extract_json_from_text(text):
@celery.task(base=BaseTaskWithRetry, bind=True) @celery.task(base=BaseTaskWithRetry, bind=True)
def extract_metadata_with_gpt(self, filename: str, cleaned_text: str, file_id: int = None): def extract_metadata_with_gpt(self, filename: str, cleaned_text: str, file_id: int = None):
"""Uses OpenAI to classify document metadata.""" """
Uses OpenAI to classify document metadata.
Args:
filename: Can be either a basename (e.g., "file.pdf") or a full path (e.g., "/workdir/processed/file.pdf")
cleaned_text: The extracted text from the document
file_id: Optional file ID for tracking
"""
task_id = self.request.id task_id = self.request.id
logger.info(f"[{task_id}] Starting metadata extraction for: {filename}") logger.info(f"[{task_id}] Starting metadata extraction for: {filename}")
log_task_progress( log_task_progress(
task_id, "extract_metadata_with_gpt", "in_progress", f"Extracting metadata for {filename}", file_id=file_id task_id, "extract_metadata_with_gpt", "in_progress", f"Extracting metadata for {os.path.basename(filename)}", file_id=file_id
) )
# Get file_id from database if not provided # Get file_id from database if not provided
if file_id is None: if file_id is None:
tmp_dir = os.path.join(settings.workdir, "tmp") tmp_dir = os.path.join(settings.workdir, "tmp")
file_path = os.path.join(tmp_dir, filename) # Handle both basename and full path
if os.path.isabs(filename):
file_path = filename
else:
file_path = os.path.join(tmp_dir, filename)
if os.path.exists(file_path): if os.path.exists(file_path):
with SessionLocal() as db: with SessionLocal() as db:
file_record = db.query(FileRecord).filter_by(local_filename=file_path).first() file_record = db.query(FileRecord).filter_by(local_filename=file_path).first()
@@ -162,14 +173,14 @@ def extract_metadata_with_gpt(self, filename: str, cleaned_text: str, file_id: i
) )
# Trigger the next step: embedding metadata into the PDF # Trigger the next step: embedding metadata into the PDF
# Pass the original filename (UUID-based) so embed_metadata_into_pdf can find the file on disk # Pass the filename (can be basename or full path) so embed_metadata_into_pdf can find the file on disk
logger.info(f"[{task_id}] Queueing metadata embedding task") logger.info(f"[{task_id}] Queueing metadata embedding task")
log_task_progress( log_task_progress(
task_id, "extract_metadata_with_gpt", "success", "Metadata extracted, queuing embed task", file_id=file_id task_id, "extract_metadata_with_gpt", "success", "Metadata extracted, queuing embed task", file_id=file_id
) )
embed_metadata_into_pdf.delay(filename, cleaned_text, metadata, file_id) embed_metadata_into_pdf.delay(filename, cleaned_text, metadata, file_id)
return {"s3_file": filename, "metadata": metadata} return {"s3_file": os.path.basename(filename), "metadata": metadata}
except Exception as e: except Exception as e:
logger.exception(f"[{task_id}] OpenAI classification failed for {filename}: {e}") logger.exception(f"[{task_id}] OpenAI classification failed for {filename}: {e}")
+43
View File
@@ -2,6 +2,7 @@
Tests for file detail view improvements including reprocessing and preview endpoints. Tests for file detail view improvements including reprocessing and preview endpoints.
""" """
import shutil
from unittest.mock import MagicMock, patch from unittest.mock import MagicMock, patch
import pytest import pytest
@@ -257,6 +258,48 @@ class TestSubtaskRetry:
assert response.status_code == 400 assert response.status_code == 400
assert "not found on disk" in response.json()["detail"].lower() assert "not found on disk" in response.json()["detail"].lower()
def test_retry_embed_metadata_with_processed_file(self, client: TestClient, db_session, sample_pdf_path, tmp_path):
"""Test retrying embed_metadata_into_pdf when file is in processed directory."""
mock_task = MagicMock()
mock_task.id = "embed-processed-retry-task"
# Create processed directory and copy file there
processed_dir = tmp_path / "processed"
processed_dir.mkdir(exist_ok=True)
processed_file = processed_dir / "processed_doc.pdf"
# Copy the sample PDF to processed directory
shutil.copy(sample_pdf_path, processed_file)
# Create file record with non-existent local_filename but existing processed_file_path
file_record = FileRecord(
filehash="pipeline_retry6",
original_filename="processed_doc.pdf",
local_filename="/nonexistent/tmp/doc.pdf", # File no longer in tmp
processed_file_path=str(processed_file), # But exists in processed
file_size=1024,
mime_type="application/pdf",
)
db_session.add(file_record)
db_session.commit()
db_session.refresh(file_record)
with patch("app.tasks.extract_metadata_with_gpt.extract_metadata_with_gpt") as mock_extract:
mock_extract.delay.return_value = mock_task
response = client.post(f"/api/files/{file_record.id}/retry-subtask?subtask_name=embed_metadata_into_pdf")
# Should succeed because file exists in processed directory
assert response.status_code == 200
data = response.json()
assert data["status"] == "success"
assert data["subtask_name"] == "embed_metadata_into_pdf"
# Verify extract_metadata_with_gpt.delay was called with full path
mock_extract.delay.assert_called_once()
call_args = mock_extract.delay.call_args
# First argument should be the full path to the processed file
assert str(processed_file) in str(call_args[0][0])
@pytest.mark.integration @pytest.mark.integration
class TestFilePreview: class TestFilePreview: