gitea-admin/gh-christianlouis-docuelevate
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge pull request #296 from christianlouis/copilot/enhance-logging-embed-metadata

Browse Source 
feat(api): add diagnostic logging and original_file_path fallback to retry-subtask endpoint
This commit is contained in:
Christian Krakau-Louis
2026-02-13 23:34:45 +01:00
committed by GitHub
parent 84f4d4d9f6 c129badc14
commit 232aa24511
2 changed files with 384 additions and 23 deletions
Download Patch File Download Diff File Expand all files Collapse all files
app/api/files.py
+108 -23
View File
@@ -496,8 +496,21 @@ def _retry_pipeline_step(file_record: FileRecord, step_name: str, db: Session) -
if step_name == "process_document":
# Full reprocessing with duplicate check bypass
if not file_record.local_filename or not os.path.exists(file_record.local_filename):
raise HTTPException(status_code=400, detail="Local file not found on disk. Cannot retry.")
logger.info(
f"Retrying process_document for file {file_id}: local_filename={file_record.local_filename!r}"
)
if not file_record.local_filename:
logger.error(f"process_document retry failed for file {file_id}: local_filename is None")
raise HTTPException(status_code=400, detail="Local file path is None. Cannot retry.")
exists = os.path.exists(file_record.local_filename)
logger.info(f"Checking local_filename: {file_record.local_filename!r}, exists={exists}")
if not exists:
error_message = f"Local file not found on disk. Cannot retry. Path checked: local_filename={file_record.local_filename!r} (exists=False)"
logger.error(f"process_document retry failed for file {file_id}: {error_message}")
raise HTTPException(status_code=400, detail=error_message)
logger.info(f"Found file for process_document retry at: {file_record.local_filename!r}")
task = process_document.delay(
file_record.local_filename, original_filename=file_record.original_filename, file_id=file_id
)
@@ -505,17 +518,42 @@ def _retry_pipeline_step(file_record: FileRecord, step_name: str, db: Session) -
from app.tasks.process_with_azure_document_intelligence import process_with_azure_document_intelligence
# OCR needs the file in workdir/tmp
if not file_record.local_filename or not os.path.exists(file_record.local_filename):
raise HTTPException(status_code=400, detail="Local file not found on disk. Cannot retry OCR.")
logger.info(
f"Retrying process_with_azure_document_intelligence for file {file_id}: "
f"local_filename={file_record.local_filename!r}"
)
if not file_record.local_filename:
logger.error(f"OCR retry failed for file {file_id}: local_filename is None")
raise HTTPException(status_code=400, detail="Local file path is None. Cannot retry OCR.")
exists = os.path.exists(file_record.local_filename)
logger.info(f"Checking local_filename: {file_record.local_filename!r}, exists={exists}")
if not exists:
error_message = f"Local file not found on disk. Cannot retry OCR. Path checked: local_filename={file_record.local_filename!r} (exists=False)"
logger.error(f"OCR retry failed for file {file_id}: {error_message}")
raise HTTPException(status_code=400, detail=error_message)
logger.info(f"Found file for OCR retry at: {file_record.local_filename!r}")
filename = os.path.basename(file_record.local_filename)
task = process_with_azure_document_intelligence.delay(filename, file_id)
elif step_name == "extract_metadata_with_gpt":
from app.tasks.extract_metadata_with_gpt import extract_metadata_with_gpt
if not file_record.local_filename or not os.path.exists(file_record.local_filename):
raise HTTPException(
status_code=400, detail="Local file not found on disk. Cannot retry metadata extraction."
)
logger.info(
f"Retrying extract_metadata_with_gpt for file {file_id}: local_filename={file_record.local_filename!r}"
)
if not file_record.local_filename:
logger.error(f"Metadata extraction retry failed for file {file_id}: local_filename is None")
raise HTTPException(status_code=400, detail="Local file path is None. Cannot retry metadata extraction.")
exists = os.path.exists(file_record.local_filename)
logger.info(f"Checking local_filename: {file_record.local_filename!r}, exists={exists}")
if not exists:
error_message = f"Local file not found on disk. Cannot retry metadata extraction. Path checked: local_filename={file_record.local_filename!r} (exists=False)"
logger.error(f"Metadata extraction retry failed for file {file_id}: {error_message}")
raise HTTPException(status_code=400, detail=error_message)
logger.info(f"Found file for metadata extraction retry at: {file_record.local_filename!r}")
extracted_text = _extract_text_from_pdf(file_record.local_filename)
filename = os.path.basename(file_record.local_filename)
task = extract_metadata_with_gpt.delay(filename, extracted_text, file_id)
@@ -526,30 +564,64 @@ def _retry_pipeline_step(file_record: FileRecord, step_name: str, db: Session) -
# 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.
# Log all database path values at the start
logger.info(
f"Retrying embed_metadata_into_pdf for file {file_id}: "
f"local_filename={file_record.local_filename!r}, "
f"processed_file_path={file_record.processed_file_path!r}, "
f"original_file_path={file_record.original_file_path!r}"
)
# 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>
# 3. Original immutable copy (file_record.original_file_path)
# 4. 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:
checked_paths = []
# Check 1: local_filename (original tmp location)
if file_record.local_filename:
exists = os.path.exists(file_record.local_filename)
checked_paths.append(f"local_filename={file_record.local_filename!r} (exists={exists})")
logger.info(f"Checking local_filename: {file_record.local_filename!r}, exists={exists}")
if exists:
file_path = file_record.local_filename
# Check 2: processed_file_path
if not file_path and file_record.processed_file_path:
exists = os.path.exists(file_record.processed_file_path)
checked_paths.append(f"processed_file_path={file_record.processed_file_path!r} (exists={exists})")
logger.info(f"Checking processed_file_path: {file_record.processed_file_path!r}, exists={exists}")
if exists:
file_path = file_record.processed_file_path
# Check 3: original_file_path (immutable copy in workdir/original/)
if not file_path and file_record.original_file_path:
exists = os.path.exists(file_record.original_file_path)
checked_paths.append(f"original_file_path={file_record.original_file_path!r} (exists={exists})")
logger.info(f"Checking original_file_path: {file_record.original_file_path!r}, exists={exists}")
if exists:
file_path = file_record.original_file_path
# Check 4: Fallback to workdir/tmp/<basename>
if not file_path and 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):
exists = os.path.exists(fallback_path)
checked_paths.append(f"workdir_tmp_fallback={fallback_path!r} (exists={exists})")
logger.info(f"Checking workdir/tmp fallback: {fallback_path!r}, exists={exists}")
if exists:
file_path = fallback_path
if not file_path:
raise HTTPException(
status_code=400,
detail="File not found in tmp or processed directory. Cannot retry metadata embedding."
)
paths_detail = "; ".join(checked_paths)
error_message = f"File not found on disk. Cannot retry metadata embedding. Paths checked: {paths_detail}"
logger.error(f"embed_metadata_into_pdf retry failed for file {file_id}: {error_message}")
raise HTTPException(status_code=400, detail=error_message)
logger.info(f"Found file for embed_metadata_into_pdf retry at: {file_path!r}")
extracted_text = _extract_text_from_pdf(file_path)
# Pass the full path to the task so it can locate the file
task = extract_metadata_task.delay(file_path, extracted_text, file_id)
@@ -642,6 +714,11 @@ def retry_subtask(
)
# Check for processed file (upload tasks work with processed files)
logger.info(
f"Retrying upload task {subtask_name} for file {file_id}: "
f"original_filename={file_record.original_filename!r}, "
f"filehash={file_record.filehash!r}"
)
workdir = settings.workdir
processed_dir = os.path.join(workdir, "processed")
@@ -654,14 +731,22 @@ def retry_subtask(
]
file_path = None
checked_paths = []
for path in potential_paths:
if os.path.exists(path):
exists = os.path.exists(path)
checked_paths.append(f"{path!r} (exists={exists})")
logger.info(f"Checking processed file path: {path!r}, exists={exists}")
if exists:
file_path = path
break
if not file_path:
raise HTTPException(status_code=400, detail="Processed file not found. Cannot retry upload.")
paths_detail = "; ".join(checked_paths)
error_message = f"Processed file not found. Cannot retry upload. Paths checked: {paths_detail}"
logger.error(f"Upload retry failed for file {file_id}: {error_message}")
raise HTTPException(status_code=400, detail=error_message)
logger.info(f"Found processed file for upload retry at: {file_path!r}")
# Queue the specific upload task
upload_task = task_map[subtask_name]
task = upload_task.delay(file_path, file_id)
tests/test_retry_subtask_logging.py
+276
View File
@@ -0,0 +1,276 @@
"""
Tests for enhanced logging and original_file_path fallback in retry-subtask endpoint.
"""
import shutil
from unittest.mock import MagicMock, patch
import pytest
from fastapi.testclient import TestClient
from app.models import FileRecord
@pytest.mark.integration
class TestRetrySubtaskEnhancedLogging:
"""Tests for enhanced logging in retry-subtask endpoint."""
def test_embed_metadata_retry_with_original_file_path(
self, client: TestClient, db_session, sample_pdf_path, tmp_path
):
"""Test that embed_metadata_into_pdf retry uses original_file_path as fallback."""
mock_task = MagicMock()
mock_task.id = "embed-original-fallback-task"
# Create original directory and copy file there
original_dir = tmp_path / "original"
original_dir.mkdir(exist_ok=True)
original_file = original_dir / "original_doc.pdf"
shutil.copy(sample_pdf_path, original_file)
# Create file record with non-existent local_filename and processed_file_path
# but existing original_file_path
file_record = FileRecord(
filehash="embed_original_test",
original_filename="original_doc.pdf",
local_filename="/nonexistent/tmp/doc.pdf", # File no longer in tmp
processed_file_path="/nonexistent/processed/doc.pdf", # Also not in processed
original_file_path=str(original_file), # But exists in original
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")
assert response.status_code == 200
data = response.json()
assert data["status"] == "success"
assert data["subtask_name"] == "embed_metadata_into_pdf"
# Verify the task was called with the original_file_path
mock_extract.delay.assert_called_once()
call_args = mock_extract.delay.call_args
# First argument should be the file path (original_file_path in this case)
assert call_args[0][0] == str(original_file)
def test_embed_metadata_retry_error_includes_all_checked_paths(self, client: TestClient, db_session):
"""Test that error message includes all checked paths with their existence status."""
# Create file record with all non-existent paths
file_record = FileRecord(
filehash="embed_error_paths_test",
original_filename="missing.pdf",
local_filename="/nonexistent/tmp/missing.pdf",
processed_file_path="/nonexistent/processed/missing.pdf",
original_file_path="/nonexistent/original/missing.pdf",
file_size=1024,
mime_type="application/pdf",
)
db_session.add(file_record)
db_session.commit()
db_session.refresh(file_record)
response = client.post(f"/api/files/{file_record.id}/retry-subtask?subtask_name=embed_metadata_into_pdf")
assert response.status_code == 400
error_detail = response.json()["detail"]
# Verify the error message contains diagnostic information
assert "Cannot retry metadata embedding" in error_detail
assert "Paths checked:" in error_detail
assert "local_filename" in error_detail
assert "processed_file_path" in error_detail
assert "original_file_path" in error_detail
assert "workdir_tmp_fallback" in error_detail
assert "exists=False" in error_detail
def test_process_document_error_includes_checked_path(self, client: TestClient, db_session):
"""Test that process_document error message includes checked path."""
file_record = FileRecord(
filehash="process_error_test",
original_filename="missing.pdf",
local_filename="/nonexistent/tmp/missing.pdf",
file_size=1024,
mime_type="application/pdf",
)
db_session.add(file_record)
db_session.commit()
db_session.refresh(file_record)
response = client.post(f"/api/files/{file_record.id}/retry-subtask?subtask_name=process_document")
assert response.status_code == 400
error_detail = response.json()["detail"]
# Verify the error message contains diagnostic information
assert "Cannot retry" in error_detail
assert "Path checked:" in error_detail
assert "local_filename" in error_detail
assert "exists=False" in error_detail
def test_ocr_retry_error_includes_checked_path(self, client: TestClient, db_session):
"""Test that OCR retry error message includes checked path."""
file_record = FileRecord(
filehash="ocr_error_test",
original_filename="missing.pdf",
local_filename="/nonexistent/tmp/missing.pdf",
file_size=1024,
mime_type="application/pdf",
)
db_session.add(file_record)
db_session.commit()
db_session.refresh(file_record)
response = client.post(
f"/api/files/{file_record.id}/retry-subtask?subtask_name=process_with_azure_document_intelligence"
)
assert response.status_code == 400
error_detail = response.json()["detail"]
# Verify the error message contains diagnostic information
assert "Cannot retry OCR" in error_detail
assert "Path checked:" in error_detail
assert "local_filename" in error_detail
assert "exists=False" in error_detail
def test_metadata_extraction_error_includes_checked_path(self, client: TestClient, db_session):
"""Test that metadata extraction retry error message includes checked path."""
file_record = FileRecord(
filehash="metadata_error_test",
original_filename="missing.pdf",
local_filename="/nonexistent/tmp/missing.pdf",
file_size=1024,
mime_type="application/pdf",
)
db_session.add(file_record)
db_session.commit()
db_session.refresh(file_record)
response = client.post(
f"/api/files/{file_record.id}/retry-subtask?subtask_name=extract_metadata_with_gpt"
)
assert response.status_code == 400
error_detail = response.json()["detail"]
# Verify the error message contains diagnostic information
assert "Cannot retry metadata extraction" in error_detail
assert "Path checked:" in error_detail
assert "local_filename" in error_detail
assert "exists=False" in error_detail
def test_upload_retry_error_includes_all_checked_paths(self, client: TestClient, db_session):
"""Test that upload retry error message includes all checked paths."""
file_record = FileRecord(
filehash="upload_error_test",
original_filename="missing.pdf",
local_filename="/nonexistent/tmp/missing.pdf",
file_size=1024,
mime_type="application/pdf",
)
db_session.add(file_record)
db_session.commit()
db_session.refresh(file_record)
response = client.post(f"/api/files/{file_record.id}/retry-subtask?subtask_name=upload_to_dropbox")
assert response.status_code == 400
error_detail = response.json()["detail"]
# Verify the error message contains diagnostic information
assert "Cannot retry upload" in error_detail
assert "Paths checked:" in error_detail
assert "exists=False" in error_detail
def test_embed_metadata_path_order(self, client: TestClient, db_session, sample_pdf_path, tmp_path):
"""Test that embed_metadata_into_pdf checks paths in the correct order."""
mock_task = MagicMock()
mock_task.id = "embed-path-order-task"
# Create all three directories with files
local_file = tmp_path / "local_file.pdf"
processed_dir = tmp_path / "processed"
processed_dir.mkdir(exist_ok=True)
processed_file = processed_dir / "processed_file.pdf"
original_dir = tmp_path / "original"
original_dir.mkdir(exist_ok=True)
original_file = original_dir / "original_file.pdf"
# Copy sample PDF to all locations
shutil.copy(sample_pdf_path, local_file)
shutil.copy(sample_pdf_path, processed_file)
shutil.copy(sample_pdf_path, original_file)
# Create file record with all paths existing
file_record = FileRecord(
filehash="embed_path_order_test",
original_filename="doc.pdf",
local_filename=str(local_file),
processed_file_path=str(processed_file),
original_file_path=str(original_file),
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")
assert response.status_code == 200
# Verify the task was called with the local_filename (first priority)
mock_extract.delay.assert_called_once()
call_args = mock_extract.delay.call_args
assert call_args[0][0] == str(local_file)
def test_embed_metadata_prefers_processed_over_original(
self, client: TestClient, db_session, sample_pdf_path, tmp_path
):
"""Test that when local_filename is missing, processed_file_path is preferred over original_file_path."""
mock_task = MagicMock()
mock_task.id = "embed-processed-priority-task"
# Create processed and original directories with files
processed_dir = tmp_path / "processed"
processed_dir.mkdir(exist_ok=True)
processed_file = processed_dir / "processed_file.pdf"
original_dir = tmp_path / "original"
original_dir.mkdir(exist_ok=True)
original_file = original_dir / "original_file.pdf"
# Copy sample PDF to both locations
shutil.copy(sample_pdf_path, processed_file)
shutil.copy(sample_pdf_path, original_file)
# Create file record with non-existent local_filename but existing processed and original
file_record = FileRecord(
filehash="embed_processed_priority_test",
original_filename="doc.pdf",
local_filename="/nonexistent/tmp/doc.pdf",
processed_file_path=str(processed_file),
original_file_path=str(original_file),
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")
assert response.status_code == 200
# Verify the task was called with the processed_file_path (second priority)
mock_extract.delay.assert_called_once()
call_args = mock_extract.delay.call_args
assert call_args[0][0] == str(processed_file)