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

Merge pull request #232 from christianlouis/copilot/fix-document-duplication-error

Browse Source 
fix(tasks): resolve metadata embedding file-not-found race condition and add detailed logging to all pipeline steps
This commit is contained in:
Christian Krakau-Louis
2026-02-11 11:40:52 +01:00
committed by GitHub
parent 665450e7e2 2b2394f858
commit b4cf4d5fd9
18 changed files with 664 additions and 125 deletions
Download Patch File Download Diff File Expand all files Collapse all files
app/api/files.py
+5 -22
View File
@@ -456,35 +456,18 @@ def _retry_pipeline_step(file_record: FileRecord, step_name: str, db: Session) -
filename = os.path.basename(file_record.local_filename)
task = extract_metadata_with_gpt.delay(filename, extracted_text, file_id)
elif step_name == "embed_metadata_into_pdf":
from app.tasks.embed_metadata_into_pdf import embed_metadata_into_pdf
# Retrieve the last successful metadata extraction result from processing logs
last_metadata_log = (
db.query(ProcessingLog)
.filter(
ProcessingLog.file_id == file_id,
ProcessingLog.step_name == "extract_metadata_with_gpt",
ProcessingLog.status == "success",
)
.order_by(ProcessingLog.timestamp.desc())
.first()
)
if not last_metadata_log:
raise HTTPException(
status_code=400,
detail="No successful metadata extraction found. Retry extract_metadata_with_gpt first.",
)
from app.tasks.extract_metadata_with_gpt import extract_metadata_with_gpt as extract_metadata_task
# Retrying embed requires re-running metadata extraction first, because
# 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.
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 embedding."
)
extracted_text = _extract_text_from_pdf(file_record.local_filename)
filename = os.path.basename(file_record.local_filename)
# Empty metadata dict: embed_metadata_into_pdf will re-run with the provided text.
# The GPT extraction step must succeed first (validated above) to ensure
# the pipeline can produce new metadata during the subsequent re-extraction.
task = embed_metadata_into_pdf.delay(filename, extracted_text, {}, file_id)
task = extract_metadata_task.delay(filename, extracted_text, file_id)
else:
raise HTTPException(status_code=400, detail=f"Unsupported pipeline step: {step_name}")
app/database.py
+22
View File
@@ -48,11 +48,33 @@ def init_db():
try:
Base.metadata.create_all(bind=engine)
logger.info("Database initialization complete (tables created if not exist).")
# 6. Run lightweight schema migrations for existing databases
_run_schema_migrations(engine)
except exc.SQLAlchemyError as e:
logger.error(f"Error initializing database: {e}")
raise
def _run_schema_migrations(engine):
"""
Apply lightweight schema migrations for columns added after the initial release.
Each migration is idempotent and safe to run multiple times.
"""
from sqlalchemy import inspect, text
inspector = inspect(engine)
# Migration: Add 'detail' column to processing_logs (added for verbose worker log output)
if "processing_logs" in inspector.get_table_names():
columns = [col["name"] for col in inspector.get_columns("processing_logs")]
if "detail" not in columns:
logger.info("Migrating processing_logs: adding 'detail' column")
with engine.begin() as conn:
conn.execute(text("ALTER TABLE processing_logs ADD COLUMN detail TEXT"))
logger.info("Migration complete: 'detail' column added to processing_logs")
def get_db():
"""
Dependency for FastAPI routes or general DB usage.
app/models.py
+2 -1
View File
@@ -1,6 +1,6 @@
# app/models.py
from sqlalchemy import Column, DateTime, ForeignKey, Integer, String, func
from sqlalchemy import Column, DateTime, ForeignKey, Integer, String, Text, func
from app.database import Base
@@ -48,6 +48,7 @@ class ProcessingLog(Base):
step_name = Column(String) # e.g., "OCR", "convert_to_pdf", "upload_s3"
status = Column(String) # "pending", "in_progress", "success", "failure"
message = Column(String, nullable=True) # Error text or success note
detail = Column(Text, nullable=True) # Verbose worker log output for diagnostics
timestamp = Column(DateTime(timezone=True), server_default=func.now())
app/tasks/embed_metadata_into_pdf.py
+20 -3
View File
@@ -98,7 +98,14 @@ def embed_metadata_into_pdf(self, local_file_path: str, extracted_text: str, met
local_file_path = alt_path
else:
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", file_id=file_id)
log_task_progress(
task_id, "embed_metadata_into_pdf", "failure", "File not found", file_id=file_id,
detail=(
f"Local file not found, cannot embed metadata.\n"
f"Tried path: {local_file_path}\n"
f"Also tried: {alt_path}"
),
)
return {"error": "File not found"}
# Work on a safe copy in a secure temporary directory
@@ -184,7 +191,14 @@ def embed_metadata_into_pdf(self, local_file_path: str, extracted_text: str, met
# 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
task_id, "embed_metadata_into_pdf", "success", "Metadata embedded, queuing finalization", file_id=file_id,
detail=(
f"Metadata embedded into PDF successfully.\n"
f"Original file: {original_file}\n"
f"Final file: {final_file_path}\n"
f"Metadata JSON: {json_path}\n"
f"Suggested filename: {suggested_filename}.pdf"
),
)
finalize_document_storage.delay(original_file, final_file_path, metadata, file_id=file_id)
@@ -209,7 +223,10 @@ def embed_metadata_into_pdf(self, local_file_path: str, extracted_text: str, met
except Exception as 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)
log_task_progress(
task_id, "embed_metadata_into_pdf", "failure", f"Exception: {str(e)}", file_id=file_id,
detail=f"Failed to embed metadata into {processed_file}.\nOriginal file: {original_file}\nException: {str(e)}",
)
# Clean up temporary file in case of error
if os.path.exists(processed_file):
try:
app/tasks/extract_metadata_with_gpt.py
+19 -9
View File
@@ -111,13 +111,17 @@ def extract_metadata_with_gpt(self, filename: str, cleaned_text: str, file_id: i
content = completion.choices[0].message.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)
log_task_progress(
task_id, "call_openai", "success", "Received OpenAI response", file_id=file_id,
detail=f"Raw classification response:\n{content}",
)
json_text = extract_json_from_text(content)
if not json_text:
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
task_id, "extract_metadata_with_gpt", "failure", "Invalid JSON in response", file_id=file_id,
detail=f"Could not parse valid JSON from GPT response.\nRaw response:\n{content}",
)
return {}
@@ -126,26 +130,29 @@ def extract_metadata_with_gpt(self, filename: str, cleaned_text: str, file_id: i
# SECURITY: Validate filename format from GPT to prevent path traversal
# The prompt requests filenames with only letters, numbers, periods, and underscores
# Enforce this constraint to prevent malicious filenames
import re
filename = metadata.get("filename", "")
if filename:
suggested_filename = metadata.get("filename", "")
if suggested_filename:
# Check if filename contains only safe characters AND explicitly check for ".."
# Defense in depth: While the regex [\w\-\. ]+ already excludes / and \,
# we explicitly reject ".." to guard against:
# 1. Potential locale-specific \w behavior
# 2. Files literally named ".." which are valid but problematic
# 3. Future code changes that might relax the regex
if not re.match(r'^[\w\-\. ]+$', filename) or ".." in filename:
logger.warning(f"[{task_id}] Invalid filename format from GPT: '{filename}', using fallback")
if not re.match(r'^[\w\-\. ]+$', suggested_filename) or ".." in suggested_filename:
logger.warning(
f"[{task_id}] Invalid filename format from GPT: '{suggested_filename}', using fallback"
)
# Reset to empty to trigger fallback to original filename
metadata["filename"] = ""
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
task_id, "parse_metadata", "success", f"Parsed metadata: {list(metadata.keys())}", file_id=file_id,
detail=f"Extracted metadata:\n{json.dumps(metadata, ensure_ascii=False, indent=2)}",
)
# 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
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
@@ -156,5 +163,8 @@ def extract_metadata_with_gpt(self, filename: str, cleaned_text: str, file_id: i
except Exception as 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)
log_task_progress(
task_id, "extract_metadata_with_gpt", "failure", f"Exception: {str(e)}", file_id=file_id,
detail=f"OpenAI classification failed for {filename}.\nException: {str(e)}",
)
return {}
app/tasks/process_document.py
+10 -1
View File
@@ -52,7 +52,10 @@ def process_document(self, original_local_file: str, original_filename: str = No
if not os.path.exists(original_local_file):
logger.error(f"[{task_id}] File {original_local_file} not found.")
log_task_progress(task_id, "process_document", "failure", "File not found")
log_task_progress(
task_id, "process_document", "failure", "File not found",
detail=f"File not found on disk: {original_local_file}",
)
return {"error": "File not found"}
# 0. Compute the file hash and check for duplicates
@@ -104,6 +107,12 @@ def process_document(self, original_local_file: str, original_filename: str = No
"success",
"Duplicate file detected, skipping",
file_id=existing.id,
detail=(
f"Duplicate file detected.\n"
f"File hash: {filehash}\n"
f"Existing file record ID: {existing.id}\n"
f"Original filename: {original_filename}"
),
)
return {
"status": "duplicate_file",
app/tasks/process_with_azure_document_intelligence.py
+59 -16
View File
@@ -11,6 +11,7 @@ from app.celery_app import celery
from app.config import settings
from app.tasks.retry_config import BaseTaskWithRetry
from app.tasks.rotate_pdf_pages import rotate_pdf_pages
from app.utils import log_task_progress
logger = logging.getLogger(__name__)
@@ -45,41 +46,43 @@ def get_pdf_page_count(file_path):
return None
def check_page_rotation(result, filename):
def check_page_rotation(result, filename, task_id=None):
"""
Checks if pages in the document are rotated and logs the rotation information.
Args:
result: The AnalyzeResult from Azure Document Intelligence API
filename: The name of the file being processed
task_id: Optional Celery task ID for log prefixing
Returns:
dict: Dictionary mapping page indices (integers) to rotation angles
"""
logger.error(f"Checking rotation for document: {filename}")
prefix = f"[{task_id}] " if task_id else ""
logger.info(f"{prefix}Checking rotation for document: {filename}")
rotation_data = {}
if not hasattr(result, "pages") or not result.pages:
logger.error(f"No page information available for rotation check: {filename}")
logger.warning(f"{prefix}No page information available for rotation check: {filename}")
return rotation_data
for i, page in enumerate(result.pages):
if hasattr(page, "angle"):
rotation_angle = page.angle
if rotation_angle != 0:
logger.error(f"Page {i+1} is rotated by {rotation_angle} degrees")
logger.info(f"{prefix}Page {i+1} is rotated by {rotation_angle} degrees")
# Store page index as integer, not string
rotation_data[i] = rotation_angle
else:
logger.error(f"Page {i+1} has no rotation (0 degrees)")
logger.info(f"{prefix}Page {i+1} has no rotation (0 degrees)")
else:
logger.error(f"Page {i+1} rotation information not available")
logger.info(f"{prefix}Page {i+1} rotation information not available")
return rotation_data
@celery.task(base=BaseTaskWithRetry)
def process_with_azure_document_intelligence(filename: str, file_id: int = None):
@celery.task(base=BaseTaskWithRetry, bind=True)
def process_with_azure_document_intelligence(self, filename: str, file_id: int = None):
"""
Processes a PDF document using Azure Document Intelligence and overlays OCR text onto
the local temporary file (stored under <workdir>/tmp).
@@ -96,6 +99,11 @@ def process_with_azure_document_intelligence(filename: str, file_id: int = None)
filename: Name of the file to process
file_id: Optional file ID to pass through to subsequent tasks
"""
task_id = self.request.id
log_task_progress(
task_id, "process_with_azure_document_intelligence", "in_progress",
f"Starting OCR for {filename}", file_id=file_id,
)
try:
tmp_file_path = os.path.join(settings.workdir, "tmp", filename)
if not os.path.exists(tmp_file_path):
@@ -107,7 +115,11 @@ def process_with_azure_document_intelligence(filename: str, file_id: int = None)
error_msg = (
f"File size ({file_size / (1024 * 1024):.2f} MB) exceeds Azure Document Intelligence limit of 500 MB"
)
logger.error(error_msg)
logger.error(f"[{task_id}] {error_msg}")
log_task_progress(
task_id, "validate_file", "failure",
f"File too large: {filename}", file_id=file_id, detail=error_msg,
)
return {"error": error_msg, "file": filename, "status": "Failed - Size limit exceeded"}
# For PDF files, check page count against service limits
@@ -116,12 +128,27 @@ def process_with_azure_document_intelligence(filename: str, file_id: int = None)
page_count = get_pdf_page_count(tmp_file_path)
if page_count is not None and page_count > AZURE_DOC_INTELLIGENCE_LIMITS["max_pages"]:
error_msg = f"PDF page count ({page_count}) exceeds Azure Document Intelligence limit of 2000 pages"
logger.error(error_msg)
logger.error(f"[{task_id}] {error_msg}")
log_task_progress(
task_id, "validate_file", "failure",
f"Too many pages: {filename}", file_id=file_id, detail=error_msg,
)
return {"error": error_msg, "file": filename, "status": "Failed - Page limit exceeded"}
if page_count is None:
logger.warning(f"Could not determine page count for {filename}, proceeding with processing anyway")
logger.warning(
f"[{task_id}] Could not determine page count for {filename}, proceeding with processing anyway"
)
logger.info(f"Processing {filename} with Azure Document Intelligence OCR.")
log_task_progress(
task_id, "validate_file", "success",
f"File validation passed for {filename}", file_id=file_id,
)
logger.info(f"[{task_id}] Processing {filename} with Azure Document Intelligence OCR.")
log_task_progress(
task_id, "call_azure_ocr", "in_progress",
f"Sending {filename} to Azure Document Intelligence", file_id=file_id,
)
# Open and send the document for processing
with open(tmp_file_path, "rb") as f:
@@ -132,23 +159,39 @@ def process_with_azure_document_intelligence(filename: str, file_id: int = None)
operation_id = poller.details["operation_id"]
# Check and log page rotation information
rotation_data = check_page_rotation(result, filename)
rotation_data = check_page_rotation(result, filename, task_id=task_id)
# Retrieve the processed searchable PDF
response = document_intelligence_client.get_analyze_result_pdf(model_id=result.model_id, result_id=operation_id)
searchable_pdf_path = tmp_file_path # Overwrite the original PDF location
with open(searchable_pdf_path, "wb") as writer:
writer.writelines(response)
logger.info(f"Searchable PDF saved at: {searchable_pdf_path}")
logger.info(f"[{task_id}] Searchable PDF saved at: {searchable_pdf_path}")
# Extract raw text content from the result
extracted_text = result.content if result.content else ""
logger.info(f"Extracted text for {filename}: {len(extracted_text)} characters")
logger.info(f"[{task_id}] Extracted text for {filename}: {len(extracted_text)} characters")
log_task_progress(
task_id, "call_azure_ocr", "success",
f"Azure OCR completed for {filename}", file_id=file_id,
detail=f"Extracted {len(extracted_text)} characters, {len(rotation_data)} rotated pages detected",
)
# Trigger page rotation task if rotation is detected, otherwise proceed to metadata extraction
rotate_pdf_pages.delay(filename, extracted_text, rotation_data, file_id)
log_task_progress(
task_id, "process_with_azure_document_intelligence", "success",
f"OCR processing complete for {filename}", file_id=file_id,
detail=f"Searchable PDF saved, {len(extracted_text)} characters extracted",
)
return {"file": filename, "searchable_pdf": searchable_pdf_path, "cleaned_text": extracted_text}
except Exception as e:
logger.error(f"Error processing {filename} with Azure Document Intelligence: {e}")
logger.error(f"[{task_id}] Error processing {filename} with Azure Document Intelligence: {e}")
log_task_progress(
task_id, "process_with_azure_document_intelligence", "failure",
f"OCR failed for {filename}", file_id=file_id, detail=str(e),
)
raise
app/tasks/refine_text_with_gpt.py
+54 -20
View File
@@ -1,38 +1,72 @@
#!/usr/bin/env python3
import logging
import openai
# Import the shared Celery instance
from app.celery_app import celery
from app.config import settings
from app.tasks.retry_config import BaseTaskWithRetry
from app.utils import log_task_progress
logger = logging.getLogger(__name__)
# Initialize OpenAI client dynamically
client = openai.OpenAI(api_key=settings.openai_api_key, base_url=settings.openai_base_url)
@celery.task(base=BaseTaskWithRetry)
def refine_text_with_gpt(filename: str, raw_text: str):
@celery.task(base=BaseTaskWithRetry, bind=True)
def refine_text_with_gpt(self, filename: str, raw_text: str):
"""Uses OpenAI to clean and refine OCR text."""
response = client.chat.completions.create(
model=settings.openai_model,
messages=[
{
"role": "system",
"content": (
"Clean and format the following text. The idea is that the text you see comes from an OCR "
"system and your task is to eliminate OCR errors. Keep the original language when doing so."
),
},
{"role": "user", "content": raw_text},
],
)
task_id = self.request.id
logger.info(f"[{task_id}] Starting OCR text refinement for: {filename}")
log_task_progress(task_id, "refine_text_with_gpt", "in_progress", f"Refining OCR text for {filename}")
cleaned_text = response.choices[0].message.content
try:
log_task_progress(task_id, "call_openai", "in_progress", "Calling OpenAI for text refinement")
# Trigger next task (import locally if needed to avoid circular imports)
from app.tasks.extract_metadata_with_gpt import extract_metadata_with_gpt
response = client.chat.completions.create(
model=settings.openai_model,
messages=[
{
"role": "system",
"content": (
"Clean and format the following text. The idea is that the text you see comes from an OCR "
"system and your task is to eliminate OCR errors. Keep the original language when doing so."
),
},
{"role": "user", "content": raw_text},
],
)
extract_metadata_with_gpt.delay(filename, cleaned_text)
cleaned_text = response.choices[0].message.content
return {"filename": filename, "cleaned_text": cleaned_text}
logger.info(f"[{task_id}] Text refinement complete for {filename}: {len(cleaned_text)} characters")
log_task_progress(
task_id,
"call_openai",
"success",
"Received refined text from OpenAI",
detail=f"Input: {len(raw_text)} chars → Output: {len(cleaned_text)} chars",
)
# Trigger next task (import locally if needed to avoid circular imports)
from app.tasks.extract_metadata_with_gpt import extract_metadata_with_gpt
extract_metadata_with_gpt.delay(filename, cleaned_text)
logger.info(f"[{task_id}] Queueing metadata extraction for {filename}")
log_task_progress(task_id, "refine_text_with_gpt", "success", "Text refined, queuing metadata extraction")
return {"filename": filename, "cleaned_text": cleaned_text}
except Exception as e:
logger.exception(f"[{task_id}] Text refinement failed for {filename}: {e}")
log_task_progress(
task_id,
"refine_text_with_gpt",
"failure",
f"Exception: {str(e)}",
detail=f"Text refinement failed for {filename}.\nException: {str(e)}",
)
raise
app/tasks/rotate_pdf_pages.py
+50 -11
View File
@@ -8,6 +8,7 @@ from app.celery_app import celery
from app.config import settings
from app.tasks.extract_metadata_with_gpt import extract_metadata_with_gpt
from app.tasks.retry_config import BaseTaskWithRetry
from app.utils import log_task_progress
logger = logging.getLogger(__name__)
@@ -48,8 +49,8 @@ def determine_rotation_angle(detected_angle):
return rotation_value
@celery.task(base=BaseTaskWithRetry)
def rotate_pdf_pages(filename: str, extracted_text: str, rotation_data=None, file_id: int = None):
@celery.task(base=BaseTaskWithRetry, bind=True)
def rotate_pdf_pages(self, filename: str, extracted_text: str, rotation_data=None, file_id: int = None):
"""
Rotates pages in a PDF document based on detected rotation angles.
@@ -60,13 +61,24 @@ def rotate_pdf_pages(filename: str, extracted_text: str, rotation_data=None, fil
file_id: Optional file ID to pass through to subsequent tasks
"""
try:
task_id = self.request.id
log_task_progress(
task_id, "rotate_pdf_pages", "in_progress",
f"Checking page rotation for {filename}", file_id=file_id,
)
pdf_path = os.path.join(settings.workdir, "tmp", filename)
if not os.path.exists(pdf_path):
raise FileNotFoundError(f"PDF file not found: {pdf_path}")
# Skip rotation if no rotation data provided
if not rotation_data:
logger.info(f"No rotation data provided for {filename}, proceeding with metadata extraction")
logger.info(
f"[{task_id}] No rotation data provided for {filename}, proceeding with metadata extraction"
)
log_task_progress(
task_id, "rotate_pdf_pages", "success",
"No rotation needed, proceeding to metadata extraction", file_id=file_id,
)
extract_metadata_with_gpt.delay(filename, extracted_text, file_id)
return {"file": filename, "status": "no_rotation_needed"}
@@ -76,14 +88,24 @@ def rotate_pdf_pages(filename: str, extracted_text: str, rotation_data=None, fil
try:
normalized_rotation_data[int(key)] = float(value)
except (ValueError, TypeError):
logger.warning(f"Invalid rotation data key-value: {key}:{value}")
logger.warning(f"[{task_id}] Invalid rotation data key-value: {key}:{value}")
if not any(abs(angle) > 0 for angle in normalized_rotation_data.values()):
logger.info(f"No significant rotations detected in {filename}, proceeding with metadata extraction")
logger.info(
f"[{task_id}] No significant rotations detected in {filename}, proceeding with metadata extraction"
)
log_task_progress(
task_id, "rotate_pdf_pages", "success",
"No rotation needed, proceeding to metadata extraction", file_id=file_id,
)
extract_metadata_with_gpt.delay(filename, extracted_text, file_id)
return {"file": filename, "status": "no_rotation_needed"}
logger.info(f"Rotating {len(normalized_rotation_data)} pages in {filename}")
logger.info(f"[{task_id}] Rotating {len(normalized_rotation_data)} pages in {filename}")
log_task_progress(
task_id, "apply_rotation", "in_progress",
f"Rotating {len(normalized_rotation_data)} pages", file_id=file_id,
)
applied_rotations = {}
# Load the PDF
@@ -104,12 +126,13 @@ def rotate_pdf_pages(filename: str, extracted_text: str, rotation_data=None, fil
# PyPDF2 uses clockwise rotation in 90-degree increments
page.rotate(rotation_angle)
logger.info(
f"Page {page_idx+1} rotated by {rotation_angle}° " f"(from detected {detected_angle}°)"
f"[{task_id}] Page {page_idx+1} rotated by {rotation_angle}° "
f"(from detected {detected_angle}°)"
)
applied_rotations[str(page_idx)] = rotation_angle
else:
logger.info(
f"Page {page_idx+1} had detected angle {detected_angle}° "
f"[{task_id}] Page {page_idx+1} had detected angle {detected_angle}° "
"but determined it doesn't need rotation"
)
@@ -120,16 +143,26 @@ def rotate_pdf_pages(filename: str, extracted_text: str, rotation_data=None, fil
pdf_writer.write(output_file)
if applied_rotations:
logger.info(f"Successfully rotated PDF: {filename} with rotations: {json.dumps(applied_rotations)}")
logger.info(
f"[{task_id}] Successfully rotated PDF: {filename} with rotations: "
f"{json.dumps(applied_rotations)}"
)
else:
logger.info(
f"Detected rotations in {filename} but no rotations were actually applied "
f"[{task_id}] Detected rotations in {filename} but no rotations were actually applied "
"(angles too small or not multiples of 90°)"
)
# Continue with metadata extraction
extract_metadata_with_gpt.delay(filename, extracted_text, file_id)
log_task_progress(
task_id, "rotate_pdf_pages", "success",
f"Rotation complete for {filename}",
file_id=file_id,
detail={"applied_rotations": applied_rotations},
)
return {
"file": filename,
"status": "rotated" if applied_rotations else "no_rotation_needed",
@@ -138,7 +171,13 @@ def rotate_pdf_pages(filename: str, extracted_text: str, rotation_data=None, fil
}
except Exception as e:
logger.error(f"Error rotating PDF {filename}: {e}")
logger.error(f"[{task_id}] Error rotating PDF {filename}: {e}")
log_task_progress(
task_id, "rotate_pdf_pages", "failure",
f"Rotation failed: {str(e)}",
file_id=file_id,
detail={"error": str(e), "filename": filename},
)
# Continue with metadata extraction despite rotation failure
extract_metadata_with_gpt.delay(filename, extracted_text, file_id)
return {"file": filename, "status": "rotation_failed", "error": str(e)}
app/tasks/upload_to_paperless.py
+6 -2
View File
@@ -245,13 +245,17 @@ def upload_to_paperless(self, file_path: str, file_id: int = None):
resp.raise_for_status()
except requests.exceptions.RequestException as exc:
error_msg = f"Failed to upload to Paperless: {exc}"
response_text = getattr(exc.response, "text", "<no response>")
logger.error(
f"[{task_id}] Failed to upload document '%s' to Paperless. Error: %s. Response=%s",
file_path,
exc,
getattr(exc.response, "text", "<no response>"),
response_text,
)
log_task_progress(
task_id, "upload_to_paperless", "failure", error_msg, file_id=file_id,
detail=f"Failed to upload document to Paperless.\nFile: {file_path}\nError: {exc}\nResponse: {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("'")
app/tasks/upload_with_rclone.py
+55 -14
View File
@@ -7,12 +7,13 @@ import subprocess
from app.celery_app import celery
from app.config import settings
from app.tasks.retry_config import BaseTaskWithRetry
from app.utils import log_task_progress
logger = logging.getLogger(__name__)
@celery.task(base=BaseTaskWithRetry)
def upload_with_rclone(file_path: str, destination: str):
@celery.task(base=BaseTaskWithRetry, bind=True)
def upload_with_rclone(self, file_path: str, destination: str):
"""
Uploads a file using rclone to the specified destination.
@@ -21,12 +22,16 @@ def upload_with_rclone(file_path: str, destination: str):
destination: Rclone destination in format "remote:path/to/folder"
e.g. "gdrive:Uploads" or "dropbox:Documents/Uploads"
"""
task_id = self.request.id
if not os.path.exists(file_path):
raise FileNotFoundError(f"File not found: {file_path}")
# Extract filename
filename = os.path.basename(file_path)
log_task_progress(task_id, "upload_with_rclone", "in_progress", f"Uploading {filename} to {destination}")
# Validate destination format to prevent command injection
if ":" not in destination:
raise ValueError(f"Invalid destination format: {destination}. Expected format: remote:path")
@@ -43,9 +48,11 @@ def upload_with_rclone(file_path: str, destination: str):
rclone_config_path = os.path.join(settings.workdir, "rclone.conf")
if not os.path.exists(rclone_config_path):
error_msg = f"Rclone configuration not found at {rclone_config_path}"
logger.error(error_msg)
logger.error(f"[{task_id}] {error_msg}")
raise ValueError(error_msg)
log_task_progress(task_id, "validate_rclone", "success", f"Validated rclone config for {remote}")
try:
# Ensure the remote path exists (create folders if needed)
mkdir_cmd = ["rclone", "mkdir", "--config", rclone_config_path, destination]
@@ -55,6 +62,8 @@ def upload_with_rclone(file_path: str, destination: str):
# Construct the upload command
upload_cmd = ["rclone", "copy", "--config", rclone_config_path, file_path, destination, "--progress"]
log_task_progress(task_id, "rclone_upload", "in_progress", f"Executing rclone copy to {destination}")
# Execute the upload command
result = subprocess.run(upload_cmd, check=True, capture_output=True, text=True)
@@ -66,44 +75,63 @@ def upload_with_rclone(file_path: str, destination: str):
link_result = subprocess.run(link_cmd, capture_output=True, text=True)
public_url = link_result.stdout.strip() if link_result.returncode == 0 else None
except (subprocess.SubprocessError, OSError) as e:
logger.warning(f"Failed to get public link for {filename}: {str(e)}")
logger.warning(f"[{task_id}] Failed to get public link for {filename}: {str(e)}")
public_url = None
logger.info(f"Successfully uploaded {filename} to {destination}")
logger.info(f"[{task_id}] Successfully uploaded {filename} to {destination}")
log_task_progress(
task_id,
"upload_with_rclone",
"success",
f"Uploaded {filename} to {destination}",
detail=f"public_url={public_url}",
)
return {"status": "Completed", "file": file_path, "destination": destination, "public_url": public_url}
else:
error_msg = f"Failed to upload {filename} to {destination}: {result.stderr}"
logger.error(error_msg)
logger.error(f"[{task_id}] {error_msg}")
raise RuntimeError(error_msg)
except subprocess.CalledProcessError as e:
error_msg = f"Rclone error: {e.stderr.decode('utf-8') if hasattr(e.stderr, 'decode') else e.stderr}"
logger.error(error_msg)
logger.error(f"[{task_id}] {error_msg}")
log_task_progress(
task_id, "upload_with_rclone", "failure", f"Rclone command failed for {filename}", detail=error_msg
)
raise RuntimeError(error_msg) from e
except (OSError, ValueError) as e:
error_msg = f"Error uploading {filename} to {destination}: {str(e)}"
logger.error(error_msg)
logger.error(f"[{task_id}] {error_msg}")
log_task_progress(
task_id, "upload_with_rclone", "failure", f"Upload failed for {filename}", detail=error_msg
)
raise RuntimeError(error_msg) from e
@celery.task(base=BaseTaskWithRetry)
def send_to_all_rclone_destinations(file_path: str):
@celery.task(base=BaseTaskWithRetry, bind=True)
def send_to_all_rclone_destinations(self, file_path: str):
"""
Uploads a file to all configured rclone destinations.
Destinations are loaded from the rclone configuration file.
"""
task_id = self.request.id
if not os.path.exists(file_path):
raise FileNotFoundError(f"File not found: {file_path}")
# Extract filename
filename = os.path.basename(file_path)
log_task_progress(
task_id, "send_to_all_rclone_destinations", "in_progress", f"Queueing rclone uploads for {filename}"
)
# Path to rclone config
rclone_config_path = os.path.join(settings.workdir, "rclone.conf")
if not os.path.exists(rclone_config_path):
error_msg = f"Rclone configuration not found at {rclone_config_path}"
logger.error(error_msg)
logger.error(f"[{task_id}] {error_msg}")
raise ValueError(error_msg)
# Get list of configured destinations from rclone
@@ -133,17 +161,30 @@ def send_to_all_rclone_destinations(file_path: str):
if path and not path.endswith("/"):
full_destination += "/"
logger.info(f"Queueing {file_path} for upload to {full_destination}")
logger.info(f"[{task_id}] Queueing {file_path} for upload to {full_destination}")
task = upload_with_rclone.delay(file_path, full_destination)
results[f"rclone_{remote.rstrip(':')}_task_id"] = task.id
log_task_progress(
task_id,
"send_to_all_rclone_destinations",
"success",
f"Queued {len(results)} rclone upload(s)",
)
return {"status": "Queued", "file_path": file_path, "tasks": results}
else:
error_msg = f"Failed to list rclone remotes: {result.stderr}"
logger.error(error_msg)
logger.error(f"[{task_id}] {error_msg}")
raise RuntimeError(error_msg)
except (subprocess.SubprocessError, OSError) as e:
error_msg = f"Error setting up rclone uploads for {filename}: {str(e)}"
logger.error(error_msg)
logger.error(f"[{task_id}] {error_msg}")
log_task_progress(
task_id,
"send_to_all_rclone_destinations",
"failure",
f"Failed to set up rclone uploads for {filename}",
detail=error_msg,
)
raise RuntimeError(error_msg) from e
app/utils/logging.py
+78 -1
View File
@@ -1,11 +1,87 @@
import logging
import threading
from collections import defaultdict
from app.database import SessionLocal
from app.models import ProcessingLog
def log_task_progress(task_id, step_name, status, message=None, file_id=None):
class TaskLogCollector(logging.Handler):
"""
A logging handler that buffers log messages per Celery task ID.
When log_task_progress() is called, it drains the buffered messages
for that task and stores them in the ProcessingLog.detail field.
This captures all logger.info/error/warning output automatically.
"""
def __init__(self):
super().__init__()
self._buffers = defaultdict(list)
self._lock = threading.Lock()
def emit(self, record: logging.LogRecord) -> None:
"""Buffer a log record if it contains a task ID marker like [task-id]."""
try:
msg = self.format(record)
# Extract task_id from messages formatted as "[task_id] ..."
if msg and "[" in msg and "]" in msg:
start = msg.index("[")
end = msg.index("]", start)
task_id = msg[start + 1 : end].strip()
if task_id and len(task_id) >= 8:
with self._lock:
self._buffers[task_id].append(msg)
except (ValueError, IndexError):
pass
def drain(self, task_id: str) -> str:
"""Return and clear all buffered messages for a task ID."""
with self._lock:
messages = self._buffers.pop(task_id, [])
return "\n".join(messages) if messages else ""
# Singleton collector instance
_collector = TaskLogCollector()
_collector.setLevel(logging.DEBUG)
_collector_installed = False
def _ensure_collector_installed() -> None:
"""Install the TaskLogCollector on the root logger (once)."""
global _collector_installed
if not _collector_installed:
root = logging.getLogger()
# Avoid duplicate handlers
if _collector not in root.handlers:
root.addHandler(_collector)
_collector_installed = True
def log_task_progress(task_id, step_name, status, message=None, file_id=None, detail=None):
"""
Logs the progress of a Celery task to the database.
If no explicit detail is provided, automatically drains any buffered
worker log output for this task ID and stores it as the detail.
Args:
task_id: The Celery task ID
step_name: Name of the processing step
status: Current status (pending, in_progress, success, failure)
message: Short summary message
file_id: Optional associated file record ID
detail: Optional verbose log output for diagnostics.
If not provided, buffered logger output is used automatically.
"""
# Auto-capture buffered log output when no explicit detail is given
if not detail and task_id:
_ensure_collector_installed()
collected = _collector.drain(task_id)
if collected:
detail = collected
with SessionLocal() as db:
log_entry = ProcessingLog(
task_id=task_id,
@@ -13,6 +89,7 @@ def log_task_progress(task_id, step_name, status, message=None, file_id=None):
status=status,
message=message,
file_id=file_id,
detail=detail,
)
db.add(log_entry)
db.commit()
frontend/templates/file_detail.html
+54
View File
@@ -256,6 +256,38 @@
max-height: 5000px;
transition: max-height 0.5s ease-in;
}
/* Verbose detail block */
.timeline-detail-toggle {
cursor: pointer;
color: #3182ce;
font-size: 0.75rem;
font-weight: 600;
margin-top: 0.25rem;
display: inline-flex;
align-items: center;
gap: 0.25rem;
}
.timeline-detail-toggle:hover {
color: #2c5aa0;
}
.timeline-detail {
display: none;
margin-top: 0.5rem;
background-color: #1a202c;
color: #e2e8f0;
padding: 0.75rem;
border-radius: 0.375rem;
font-family: monospace;
font-size: 0.75rem;
white-space: pre-wrap;
word-break: break-word;
max-height: 400px;
overflow-y: auto;
}
.timeline-detail.visible {
display: block;
}
.no-logs {
text-align: center;
@@ -590,6 +622,21 @@
toggleIcon.classList.add('fa-chevron-up');
}
}
// Toggle verbose detail for a log entry
function toggleDetail(logId) {
const detail = document.getElementById('detail-' + logId);
const icon = document.getElementById('detail-icon-' + logId);
if (detail.classList.contains('visible')) {
detail.classList.remove('visible');
icon.classList.remove('fa-chevron-up');
icon.classList.add('fa-chevron-right');
} else {
detail.classList.add('visible');
icon.classList.remove('fa-chevron-right');
icon.classList.add('fa-chevron-up');
}
}
</script>
{% endif %}
{% endblock %}
@@ -760,6 +807,13 @@
<span class="timeline-task-id">Task: {{ log.task_id[:16] }}...</span>
{% endif %}
</div>
{% if log.detail %}
<div class="timeline-detail-toggle" onclick="toggleDetail({{ loop.index }})">
<i id="detail-icon-{{ loop.index }}" class="fas fa-chevron-right"></i>
<span>Show worker log detail</span>
</div>
<div id="detail-{{ loop.index }}" class="timeline-detail">{{ log.detail }}</div>
{% endif %}
</div>
</div>
{% endfor %}
tests/test_database.py
+74
View File
@@ -70,3 +70,77 @@ class TestGetDb:
session = next(gen)
# Force the generator to close
gen.close()
@pytest.mark.unit
class TestSchemaMigrations:
"""Tests for schema migration logic."""
def test_processing_log_detail_column_exists(self, db_session):
"""Test that ProcessingLog has the detail column."""
from app.models import ProcessingLog
log = ProcessingLog(
task_id="test-task",
step_name="test_step",
status="success",
message="Short message",
detail="Verbose worker log output\nWith multiple lines",
)
db_session.add(log)
db_session.commit()
db_session.refresh(log)
assert log.detail == "Verbose worker log output\nWith multiple lines"
def test_processing_log_detail_nullable(self, db_session):
"""Test that detail column is nullable (backward compatible)."""
from app.models import ProcessingLog
log = ProcessingLog(
task_id="test-task-2",
step_name="test_step",
status="success",
message="Short message",
)
db_session.add(log)
db_session.commit()
db_session.refresh(log)
assert log.detail is None
def test_migration_adds_detail_column(self, tmp_path):
"""Test that _run_schema_migrations adds detail column to existing tables."""
from sqlalchemy import Column, Integer, String, create_engine, text
from sqlalchemy.orm import sessionmaker
from app.database import _run_schema_migrations
# Create a database with the old schema (no detail column)
db_path = str(tmp_path / "migration_test.db")
engine = create_engine(f"sqlite:///{db_path}")
with engine.begin() as conn:
conn.execute(
text(
"CREATE TABLE processing_logs ("
"id INTEGER PRIMARY KEY, "
"file_id INTEGER, "
"task_id VARCHAR, "
"step_name VARCHAR, "
"status VARCHAR, "
"message VARCHAR, "
"timestamp DATETIME)"
)
)
# Run migrations
_run_schema_migrations(engine)
# Verify detail column was added
from sqlalchemy import inspect
inspector = inspect(engine)
columns = [col["name"] for col in inspector.get_columns("processing_logs")]
assert "detail" in columns
engine.dispose()
tests/test_file_detail_endpoints.py
+16 -6
View File
@@ -213,8 +213,11 @@ class TestSubtaskRetry:
assert data["status"] == "success"
assert data["subtask_name"] == "extract_metadata_with_gpt"
def test_retry_pipeline_step_embed_no_metadata(self, client: TestClient, db_session, sample_pdf_path):
"""Test retrying embed_metadata_into_pdf without prior metadata extraction."""
def test_retry_pipeline_step_embed_metadata(self, client: TestClient, db_session, sample_pdf_path):
"""Test retrying embed_metadata_into_pdf re-triggers metadata extraction."""
mock_task = MagicMock()
mock_task.id = "embed-retry-task"
file_record = FileRecord(
filehash="pipeline_retry4",
original_filename="embed_retry.pdf",
@@ -226,10 +229,17 @@ class TestSubtaskRetry:
db_session.commit()
db_session.refresh(file_record)
# Try to retry embed without a successful metadata extraction log
response = client.post(f"/api/files/{file_record.id}/retry-subtask?subtask_name=embed_metadata_into_pdf")
assert response.status_code == 400
assert "retry extract_metadata_with_gpt first" in response.json()["detail"].lower()
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 extract_metadata_with_gpt.delay was called (which chains into embed)
mock_extract.delay.assert_called_once()
def test_retry_pipeline_step_missing_local_file(self, client: TestClient, db_session):
"""Test retrying a pipeline step when local file is missing."""
tests/test_logging_utils.py
+104 -1
View File
@@ -4,6 +4,8 @@ Tests for app/utils/logging.py
Tests task progress logging functionality.
"""
import logging
import pytest
from unittest.mock import Mock, patch, MagicMock
@@ -42,6 +44,7 @@ class TestTaskLogging:
status="started",
message="Processing document",
file_id=456,
detail=None,
)
# Verify database operations
@@ -67,7 +70,7 @@ class TestTaskLogging:
# Verify called with None for optional parameters
mock_processing_log.assert_called_once_with(
task_id="task-456", step_name="upload", status="completed", message=None, file_id=None
task_id="task-456", step_name="upload", status="completed", message=None, file_id=None, detail=None
)
mock_db.add.assert_called_once()
@@ -127,6 +130,7 @@ class TestTaskLogging:
status="success",
message="Document processed successfully",
file_id=999,
detail=None,
)
mock_db.add.assert_called_once()
@@ -174,3 +178,102 @@ class TestTaskLogging:
assert mock_processing_log.called
mock_db.add.assert_called()
mock_db.commit.assert_called()
@patch("app.utils.logging.SessionLocal")
@patch("app.utils.logging.ProcessingLog")
def test_log_task_progress_with_explicit_detail(self, mock_processing_log, mock_session_local):
"""Test logging with explicit detail preserves it."""
from app.utils.logging import log_task_progress
mock_db = MagicMock()
mock_session_local.return_value.__enter__.return_value = mock_db
mock_log_entry = Mock()
mock_processing_log.return_value = mock_log_entry
log_task_progress(
task_id="task-explicit",
step_name="test_step",
status="success",
message="Short message",
detail="Verbose detail output",
)
mock_processing_log.assert_called_once_with(
task_id="task-explicit",
step_name="test_step",
status="success",
message="Short message",
file_id=None,
detail="Verbose detail output",
)
@pytest.mark.unit
class TestTaskLogCollector:
"""Test the TaskLogCollector handler."""
def test_collector_buffers_log_messages(self):
"""Test that the collector buffers messages by task ID."""
from app.utils.logging import TaskLogCollector
collector = TaskLogCollector()
collector.setFormatter(logging.Formatter("%(message)s"))
logger = logging.getLogger("test_collector")
logger.addHandler(collector)
logger.setLevel(logging.DEBUG)
logger.info("[abc12345-task] Step 1 starting")
logger.info("[abc12345-task] Step 1 complete")
logger.info("[other-task-id] Different task")
result = collector.drain("abc12345-task")
assert "Step 1 starting" in result
assert "Step 1 complete" in result
assert "Different task" not in result
# After drain, buffer should be empty
assert collector.drain("abc12345-task") == ""
# Other task still has its messages
result2 = collector.drain("other-task-id")
assert "Different task" in result2
logger.removeHandler(collector)
def test_collector_ignores_short_ids(self):
"""Test that the collector ignores short bracketed strings."""
from app.utils.logging import TaskLogCollector
collector = TaskLogCollector()
collector.setFormatter(logging.Formatter("%(message)s"))
logger = logging.getLogger("test_short_ids")
logger.addHandler(collector)
logger.setLevel(logging.DEBUG)
logger.info("[OK] short id")
assert collector.drain("OK") == ""
logger.removeHandler(collector)
def test_collector_handles_malformed_brackets(self):
"""Test that the collector handles messages with [ but no ]."""
from app.utils.logging import TaskLogCollector
collector = TaskLogCollector()
collector.setFormatter(logging.Formatter("%(message)s"))
logger = logging.getLogger("test_malformed")
logger.addHandler(collector)
logger.setLevel(logging.DEBUG)
logger.info("[no closing bracket")
logger.info("no brackets at all")
logger.info("")
# Should not raise and should not buffer anything
assert collector.drain("no closing bracket") == ""
logger.removeHandler(collector)
tests/test_ocr_processing.py
+30 -15
View File
@@ -24,7 +24,8 @@ from app.tasks.rotate_pdf_pages import rotate_pdf_pages, determine_rotation_angl
class TestProcessWithAzureDocumentIntelligence:
"""Tests for Azure Document Intelligence OCR processing."""
def test_successful_ocr_processing(self, tmp_path):
@patch("app.tasks.process_with_azure_document_intelligence.log_task_progress")
def test_successful_ocr_processing(self, mock_log, tmp_path):
"""Test successful OCR processing with Azure Document Intelligence."""
# Create tmp directory and test PDF file
tmp_dir = tmp_path / "tmp"
@@ -120,7 +121,8 @@ startxref
assert call_args[0][1] == "This is extracted text from OCR"
assert call_args[0][3] == 1 # file_id
def test_file_not_found_error(self, tmp_path):
@patch("app.tasks.process_with_azure_document_intelligence.log_task_progress")
def test_file_not_found_error(self, mock_log, tmp_path):
"""Test that FileNotFoundError is raised when file doesn't exist."""
with patch(
"app.tasks.process_with_azure_document_intelligence.settings"
@@ -134,7 +136,8 @@ startxref
assert "Local file not found" in str(exc_info.value)
def test_file_size_limit_exceeded(self, tmp_path):
@patch("app.tasks.process_with_azure_document_intelligence.log_task_progress")
def test_file_size_limit_exceeded(self, mock_log, tmp_path):
"""Test file size limit validation (500 MB)."""
# Create tmp directory and test PDF file
tmp_dir = tmp_path / "tmp"
@@ -163,7 +166,8 @@ startxref
assert "Size limit exceeded" in result["status"]
assert "500 MB" in result["error"]
def test_page_count_limit_exceeded(self, tmp_path):
@patch("app.tasks.process_with_azure_document_intelligence.log_task_progress")
def test_page_count_limit_exceeded(self, mock_log, tmp_path):
"""Test page count limit validation (2000 pages)."""
# Create tmp directory and test PDF file
tmp_dir = tmp_path / "tmp"
@@ -190,7 +194,8 @@ startxref
assert "Page limit exceeded" in result["status"]
assert "2000 pages" in result["error"]
def test_page_count_unknown_proceeds_with_processing(self, tmp_path):
@patch("app.tasks.process_with_azure_document_intelligence.log_task_progress")
def test_page_count_unknown_proceeds_with_processing(self, mock_log, tmp_path):
"""Test that processing continues when page count cannot be determined."""
# Create tmp directory and test PDF file
tmp_dir = tmp_path / "tmp"
@@ -238,7 +243,8 @@ startxref
assert "error" not in result
assert result["file"] == test_pdf.name
def test_page_rotation_detection(self, tmp_path):
@patch("app.tasks.process_with_azure_document_intelligence.log_task_progress")
def test_page_rotation_detection(self, mock_log, tmp_path):
"""Test detection and logging of page rotation."""
# Create tmp directory and test PDF
tmp_dir = tmp_path / "tmp"
@@ -295,7 +301,8 @@ startxref
assert 1 not in rotation_data # Page 2 has no rotation
assert 2 in rotation_data and rotation_data[2] == 180
def test_azure_api_error_handling(self, tmp_path):
@patch("app.tasks.process_with_azure_document_intelligence.log_task_progress")
def test_azure_api_error_handling(self, mock_log, tmp_path):
"""Test error handling when Azure API fails."""
# Create tmp directory and test PDF
tmp_dir = tmp_path / "tmp"
@@ -430,7 +437,8 @@ startxref
class TestRefineTextWithGPT:
"""Tests for OpenAI text refinement task."""
def test_successful_text_refinement(self):
@patch("app.tasks.refine_text_with_gpt.log_task_progress")
def test_successful_text_refinement(self, mock_log):
"""Test successful text refinement with OpenAI."""
raw_text = "This is s0me text with OCR err0rs"
filename = "test.pdf"
@@ -477,7 +485,8 @@ class TestRefineTextWithGPT:
filename, "This is some text with OCR errors"
)
def test_openai_api_error(self):
@patch("app.tasks.refine_text_with_gpt.log_task_progress")
def test_openai_api_error(self, mock_log):
"""Test error handling when OpenAI API fails."""
raw_text = "Test text"
filename = "test.pdf"
@@ -549,7 +558,8 @@ class TestRotatePdfPages:
# Round(135/90) = Round(1.5) = 2, so 2 * 90 = 180
assert angle == 180
def test_rotate_pdf_pages_with_rotation(self, tmp_path):
@patch("app.tasks.rotate_pdf_pages.log_task_progress")
def test_rotate_pdf_pages_with_rotation(self, mock_log, tmp_path):
"""Test PDF rotation with rotation data."""
# Create tmp directory and test PDF
tmp_dir = tmp_path / "tmp"
@@ -623,7 +633,8 @@ startxref
test_pdf.name, extracted_text, 1
)
def test_rotate_pdf_pages_no_rotation_needed(self, tmp_path):
@patch("app.tasks.rotate_pdf_pages.log_task_progress")
def test_rotate_pdf_pages_no_rotation_needed(self, mock_log, tmp_path):
"""Test PDF rotation when no rotation is needed."""
# Create tmp directory and test PDF
tmp_dir = tmp_path / "tmp"
@@ -656,7 +667,8 @@ startxref
# Verify metadata extraction was still queued
mock_extract.delay.assert_called_once()
def test_rotate_pdf_pages_zero_rotations(self, tmp_path):
@patch("app.tasks.rotate_pdf_pages.log_task_progress")
def test_rotate_pdf_pages_zero_rotations(self, mock_log, tmp_path):
"""Test PDF rotation when rotation data contains only zero angles."""
# Create tmp directory and test PDF
tmp_dir = tmp_path / "tmp"
@@ -686,7 +698,8 @@ startxref
# Verify no rotation was applied
assert result["status"] == "no_rotation_needed"
def test_rotate_pdf_pages_file_not_found(self, tmp_path):
@patch("app.tasks.rotate_pdf_pages.log_task_progress")
def test_rotate_pdf_pages_file_not_found(self, mock_log, tmp_path):
"""Test error handling when PDF file is not found."""
# Create tmp directory but no PDF
tmp_dir = tmp_path / "tmp"
@@ -717,7 +730,8 @@ startxref
# Verify metadata extraction was still queued
mock_extract.delay.assert_called_once()
def test_rotate_pdf_pages_continues_on_error(self, tmp_path):
@patch("app.tasks.rotate_pdf_pages.log_task_progress")
def test_rotate_pdf_pages_continues_on_error(self, mock_log, tmp_path):
"""Test that metadata extraction continues even if rotation fails."""
# Create tmp directory and invalid PDF
tmp_dir = tmp_path / "tmp"
@@ -751,7 +765,8 @@ startxref
# Verify metadata extraction was still queued
mock_extract.delay.assert_called_once()
def test_rotate_pdf_pages_with_string_keys(self, tmp_path):
@patch("app.tasks.rotate_pdf_pages.log_task_progress")
def test_rotate_pdf_pages_with_string_keys(self, mock_log, tmp_path):
"""Test that rotation data with string keys is properly handled."""
# Create tmp directory and test PDF
tmp_dir = tmp_path / "tmp"
tests/test_rclone_tasks.py
+6 -3
View File
@@ -15,7 +15,8 @@ class TestUploadWithRclone:
with pytest.raises(FileNotFoundError):
upload_with_rclone("/nonexistent/file.pdf", "remote:path")
def test_raises_value_error_invalid_destination(self, tmp_path):
@patch("app.tasks.upload_with_rclone.log_task_progress")
def test_raises_value_error_invalid_destination(self, mock_log, tmp_path):
"""Test raises ValueError for invalid destination format."""
test_file = tmp_path / "test.pdf"
test_file.write_bytes(b"test")
@@ -23,7 +24,8 @@ class TestUploadWithRclone:
with pytest.raises(ValueError, match="Invalid destination format"):
upload_with_rclone(str(test_file), "invalid_destination")
def test_raises_value_error_invalid_remote_name(self, tmp_path):
@patch("app.tasks.upload_with_rclone.log_task_progress")
def test_raises_value_error_invalid_remote_name(self, mock_log, tmp_path):
"""Test raises ValueError for invalid remote name."""
test_file = tmp_path / "test.pdf"
test_file.write_bytes(b"test")
@@ -31,7 +33,8 @@ class TestUploadWithRclone:
with pytest.raises(ValueError, match="Invalid remote name"):
upload_with_rclone(str(test_file), ":path")
def test_raises_value_error_no_config(self, tmp_path):
@patch("app.tasks.upload_with_rclone.log_task_progress")
def test_raises_value_error_no_config(self, mock_log, tmp_path):
"""Test raises ValueError when rclone config not found."""
test_file = tmp_path / "test.pdf"
test_file.write_bytes(b"test")