diff --git a/app/api/files.py b/app/api/files.py index abdc4ed5..7da4f6ac 100644 --- a/app/api/files.py +++ b/app/api/files.py @@ -392,6 +392,72 @@ def reprocess_single_file(request: Request, file_id: int, db: DbSession): raise HTTPException(status_code=500, detail=f"Error reprocessing file: {str(e)}") +@router.post("/files/{file_id}/reprocess-with-cloud-ocr") +@require_login +def reprocess_with_cloud_ocr(request: Request, file_id: int, db: DbSession): + """ + Reprocess a single file with forced Cloud OCR processing. + + This endpoint forces Azure Document Intelligence OCR processing regardless + of whether the PDF contains embedded text. Useful for documents with + low-quality embedded text or when higher quality OCR is needed. + + Args: + file_id: ID of the file to reprocess with Cloud OCR + + Returns: + Task ID and status information + """ + try: + # Find the file record + file_record = db.query(FileRecord).filter(FileRecord.id == file_id).first() + + if not file_record: + raise HTTPException(status_code=404, detail=f"File with ID {file_id} not found") + + # Prefer using the original_file_path if available, otherwise fall back to local_filename + source_file = None + if file_record.original_file_path and os.path.exists(file_record.original_file_path): + source_file = file_record.original_file_path + logger.info(f"Using original file for Cloud OCR reprocessing: {source_file}") + elif file_record.local_filename and os.path.exists(file_record.local_filename): + source_file = file_record.local_filename + logger.info(f"Using local file for Cloud OCR reprocessing: {source_file}") + else: + raise HTTPException( + status_code=400, + detail="Neither original nor local file found on disk. Cannot reprocess." + ) + + # Queue the file for processing with force_cloud_ocr=True + task = process_document.delay( + source_file, + original_filename=file_record.original_filename, + file_id=file_record.id, + force_cloud_ocr=True + ) + + logger.info( + f"Reprocessing file with Cloud OCR: ID={file_record.id}, " + f"Filename={file_record.original_filename}, TaskID={task.id}" + ) + + return { + "status": "success", + "message": "File queued for Cloud OCR reprocessing", + "file_id": file_record.id, + "filename": file_record.original_filename, + "task_id": task.id, + "force_cloud_ocr": True, + } + + except HTTPException: + raise + except Exception as e: + logger.exception(f"Error reprocessing file {file_id} with Cloud OCR: {str(e)}") + raise HTTPException(status_code=500, detail=f"Error reprocessing file with Cloud OCR: {str(e)}") + + def _extract_text_from_pdf(file_path: str) -> str: """ Extract text from a PDF file using PyPDF2. diff --git a/app/models.py b/app/models.py index 9be67422..9ca3c859 100644 --- a/app/models.py +++ b/app/models.py @@ -30,6 +30,14 @@ class FileRecord(Base): # The name/path we store on disk (e.g. /workdir/tmp/.pdf) local_filename = Column(String, nullable=False) + # Immutable original copy path (e.g. /workdir/original/.pdf) + # This is the first copy made when the file is ingested + original_file_path = Column(String) + + # Processed copy path (e.g. /workdir/processed/2024-01-01_Invoice.pdf) + # This is the final file with embedded metadata before upload + processed_file_path = Column(String) + # Size of the file in bytes file_size = Column(Integer, nullable=False) diff --git a/app/tasks/embed_metadata_into_pdf.py b/app/tasks/embed_metadata_into_pdf.py index ba3b9b3a..fd8ec487 100644 --- a/app/tasks/embed_metadata_into_pdf.py +++ b/app/tasks/embed_metadata_into_pdf.py @@ -16,7 +16,7 @@ from app.database import SessionLocal from app.models import FileRecord from app.tasks.finalize_document_storage import finalize_document_storage from app.tasks.retry_config import BaseTaskWithRetry -from app.utils import log_task_progress +from app.utils import get_unique_filepath_with_counter, log_task_progress from app.utils.filename_utils import sanitize_filename logger = logging.getLogger(__name__) @@ -30,32 +30,35 @@ TMP_SUBDIR = "tmp" PROCESSED_SUBDIR = "processed" -def unique_filepath(directory, base_filename, extension=".pdf"): - """ - Returns a unique filepath in the specified directory. - If 'base_filename.pdf' exists, it will append an underscore and counter. - """ - candidate = os.path.join(directory, base_filename + extension) - if not os.path.exists(candidate): - return candidate - counter = 1 - while True: - candidate = os.path.join(directory, f"{base_filename}_{counter}{extension}") - if not os.path.exists(candidate): - return candidate - counter += 1 - - -def persist_metadata(metadata, final_pdf_path): +def persist_metadata(metadata, final_pdf_path, original_file_path=None, processed_file_path=None): """ Saves the metadata dictionary to a JSON file with the same base name as the final PDF. For example, if final_pdf_path is "/processed/MyFile.pdf", the metadata will be saved as "/processed/MyFile.json". + + Optionally augments the metadata with file path references for traceability. + + Args: + metadata: Dictionary of metadata to save + final_pdf_path: Path to the final PDF file + original_file_path: Optional path to the immutable original file + processed_file_path: Optional path to the processed file + + Returns: + str: Path to the created JSON file """ base, _ = os.path.splitext(final_pdf_path) json_path = base + ".json" + + # Augment metadata with file path references if provided + metadata_with_paths = metadata.copy() + if original_file_path: + metadata_with_paths["original_file_path"] = original_file_path + if processed_file_path: + metadata_with_paths["processed_file_path"] = processed_file_path + with open(json_path, "w", encoding="utf-8") as f: - json.dump(metadata, f, ensure_ascii=False, indent=2) + json.dump(metadata_with_paths, f, ensure_ascii=False, indent=2) return json_path @@ -159,15 +162,15 @@ def embed_metadata_into_pdf(self, local_file_path: str, extracted_text: str, met # Define the final directory based on settings.workdir and ensure it exists. final_dir = os.path.join(settings.workdir, PROCESSED_SUBDIR) os.makedirs(final_dir, exist_ok=True) - # Get a unique filepath in case of collisions. - final_file_path = unique_filepath(final_dir, suggested_filename, extension=".pdf") + # Get a unique filepath in case of collisions using -0001, -0002 suffix format + final_file_path = get_unique_filepath_with_counter(final_dir, suggested_filename, extension=".pdf") logger.info(f"[{task_id}] Moving file to: {final_file_path}") log_task_progress( task_id, "move_to_processed", "in_progress", - f"Moving to processed: {suggested_filename}.pdf", + f"Moving to processed: {os.path.basename(final_file_path)}", file_id=file_id, ) # Move the processed file using shutil.move to handle cross-device moves. @@ -179,10 +182,28 @@ def embed_metadata_into_pdf(self, local_file_path: str, extracted_text: str, met task_id, "move_to_processed", "success", f"Moved to: {os.path.basename(final_file_path)}", file_id=file_id ) + # Get the original_file_path from the database + original_file_path = None + with SessionLocal() as db: + if file_id: + file_record = db.query(FileRecord).filter_by(id=file_id).first() + if file_record: + original_file_path = file_record.original_file_path + # Update the processed_file_path in the database + file_record.processed_file_path = final_file_path + db.commit() + logger.info(f"[{task_id}] Updated database with processed_file_path: {final_file_path}") + # Persist the metadata into a JSON file with the same base name. + # Include file path references for traceability logger.info(f"[{task_id}] Persisting metadata to JSON") log_task_progress(task_id, "save_metadata_json", "in_progress", "Saving metadata JSON", file_id=file_id) - json_path = persist_metadata(metadata, final_file_path) + json_path = persist_metadata( + metadata, + final_file_path, + original_file_path=original_file_path, + processed_file_path=final_file_path + ) logger.info(f"[{task_id}] Metadata persisted to {json_path}") log_task_progress( task_id, "save_metadata_json", "success", f"Saved: {os.path.basename(json_path)}", file_id=file_id diff --git a/app/tasks/process_document.py b/app/tasks/process_document.py index e0a72da1..e214e2e0 100644 --- a/app/tasks/process_document.py +++ b/app/tasks/process_document.py @@ -17,13 +17,13 @@ from app.tasks.process_with_azure_document_intelligence import ( process_with_azure_document_intelligence, ) from app.tasks.retry_config import BaseTaskWithRetry -from app.utils import hash_file, log_task_progress +from app.utils import get_unique_filepath_with_counter, hash_file, log_task_progress logger = logging.getLogger(__name__) @celery.task(base=BaseTaskWithRetry, bind=True) -def process_document(self, original_local_file: str, original_filename: str = None, file_id: int = None): +def process_document(self, original_local_file: str, original_filename: str = None, file_id: int = None, force_cloud_ocr: bool = False): """ Process a document file and trigger appropriate text extraction. @@ -32,14 +32,18 @@ def process_document(self, original_local_file: str, original_filename: str = No original_filename: Optional original filename (if different from path basename) file_id: Optional existing file record ID. When provided, skips duplicate detection and reuses the existing record (used for reprocessing). + force_cloud_ocr: If True, forces Azure Document Intelligence OCR processing + regardless of embedded text quality. Used for re-processing. Steps: 1. Check if we have a FileRecord entry (via SHA-256 hash). If found, skip re-processing. (Skipped when file_id is provided for reprocessing.) 2. If not found, insert a new DB row and continue with the pipeline: - - Copy file to /workdir/tmp + - Save immutable copy to /workdir/original + - Copy file to /workdir/tmp for processing - Check for embedded text. If present, run local GPT extraction - Otherwise, queue Azure Document Intelligence processing + 3. If force_cloud_ocr is True, skip local text extraction and use cloud OCR """ task_id = self.request.id logger.info(f"[{task_id}] Starting document processing: {original_local_file}") @@ -142,16 +146,43 @@ def process_document(self, original_local_file: str, original_filename: str = No file_id=new_record.id, ) - # 1. Generate a UUID-based filename and place it in /workdir/tmp + # 1. Generate a UUID-based filename for storage file_ext = os.path.splitext(original_local_file)[1] file_uuid = str(uuid.uuid4()) new_filename = f"{file_uuid}{file_ext}" + # 2. Save immutable copy to /workdir/original + # This copy serves as the permanent, untouched reference of the ingested file + original_dir = os.path.join(settings.workdir, "original") + os.makedirs(original_dir, exist_ok=True) + + # Use collision-resistant naming with -0001, -0002 suffixes + base_name = os.path.splitext(new_filename)[0] + original_file_path = get_unique_filepath_with_counter(original_dir, base_name, file_ext) + + logger.info(f"[{task_id}] Saving immutable original to: {original_file_path}") + log_task_progress( + task_id, + "save_original", + "in_progress", + f"Saving original to {os.path.basename(original_file_path)}", + file_id=new_record.id, + ) + shutil.copy(original_local_file, original_file_path) + log_task_progress( + task_id, + "save_original", + "success", + f"Original saved: {os.path.basename(original_file_path)}", + file_id=new_record.id, + ) + + # 3. Copy to /workdir/tmp for processing tmp_dir = os.path.join(settings.workdir, "tmp") os.makedirs(tmp_dir, exist_ok=True) new_local_path = os.path.join(tmp_dir, new_filename) - logger.info(f"[{task_id}] Copying file to: {new_local_path}") + logger.info(f"[{task_id}] Copying file to processing area: {new_local_path}") log_task_progress( task_id, "copy_file", @@ -169,14 +200,35 @@ def process_document(self, original_local_file: str, original_filename: str = No file_id=new_record.id, ) - # Update the DB with final local filename + # Update the DB with file paths new_record.local_filename = new_local_path + new_record.original_file_path = original_file_path db.commit() # Store file_id before session closes to avoid DetachedInstanceError file_id = new_record.id # 2. Check for embedded text (outside the DB session to avoid long open transactions) + # Skip local text extraction if force_cloud_ocr is True + if force_cloud_ocr: + logger.info(f"[{task_id}] Force Cloud OCR requested, skipping embedded text check") + log_task_progress( + task_id, + "check_text", + "success", + "Force Cloud OCR requested, queuing OCR", + file_id=file_id, + ) + log_task_progress( + task_id, + "process_document", + "success", + "Queued for forced OCR processing", + file_id=file_id, + ) + process_with_azure_document_intelligence.delay(new_filename, file_id) + return {"file": new_local_path, "status": "Queued for forced OCR", "file_id": file_id} + logger.info(f"[{task_id}] Checking for embedded text in PDF") log_task_progress( task_id, diff --git a/app/utils/__init__.py b/app/utils/__init__.py index 8a4ad1a3..8a8c0d7f 100644 --- a/app/utils/__init__.py +++ b/app/utils/__init__.py @@ -4,7 +4,8 @@ Utility functions and helpers for the document processor application. # Import functions to make them available through the package from app.utils.file_operations import hash_file +from app.utils.filename_utils import get_unique_filepath_with_counter, sanitize_filename from app.utils.logging import log_task_progress # Export all the functions that should be available when importing from app.utils -__all__ = ["hash_file", "log_task_progress"] +__all__ = ["hash_file", "log_task_progress", "get_unique_filepath_with_counter", "sanitize_filename"] diff --git a/app/utils/filename_utils.py b/app/utils/filename_utils.py index 83b8baff..498475c7 100644 --- a/app/utils/filename_utils.py +++ b/app/utils/filename_utils.py @@ -69,6 +69,57 @@ def get_unique_filename(original_path, check_exists_func=None): return new_path +def get_unique_filepath_with_counter(directory, base_filename, extension=".pdf"): + """ + Returns a unique filepath in the specified directory using a numeric counter suffix. + If 'base_filename.pdf' exists, it will append '-0001', '-0002', etc. + + This function implements robust collision handling with zero-padded numeric suffixes + as required for document storage organization. + + Args: + directory (str): Directory path where the file will be stored + base_filename (str): Base name for the file (without extension) + extension (str): File extension including the dot (default: ".pdf") + + Returns: + str: Full path to a unique filename + + Examples: + >>> get_unique_filepath_with_counter("/workdir/original", "2024-01-01_Invoice") + "/workdir/original/2024-01-01_Invoice.pdf" # If doesn't exist + + >>> get_unique_filepath_with_counter("/workdir/original", "2024-01-01_Invoice") + "/workdir/original/2024-01-01_Invoice-0001.pdf" # If original exists + """ + # Try the base filename first + candidate = os.path.join(directory, base_filename + extension) + if not os.path.exists(candidate): + return candidate + + # If base exists, try with counter suffix + counter = 1 + while True: + # Use zero-padded 4-digit counter: -0001, -0002, etc. + suffix = f"-{counter:04d}" + candidate = os.path.join(directory, f"{base_filename}{suffix}{extension}") + if not os.path.exists(candidate): + return candidate + counter += 1 + + # Sanity check to prevent infinite loops (very unlikely to reach) + if counter > 9999: + # Fall back to timestamp + UUID if somehow we have 10000 collisions + timestamp = datetime.now().strftime("%Y%m%d_%H%M%S") + uuid_str = str(uuid.uuid4())[:8] + candidate = os.path.join(directory, f"{base_filename}-{timestamp}-{uuid_str}{extension}") + logger.warning( + f"Exceeded 9999 file collisions for {base_filename}, " + f"using timestamp+UUID: {os.path.basename(candidate)}" + ) + return candidate + + def sanitize_filename(filename): r""" Sanitize a filename to ensure it's valid across different file systems diff --git a/migrations/versions/002_add_file_paths.py b/migrations/versions/002_add_file_paths.py new file mode 100644 index 00000000..e05d9a8b --- /dev/null +++ b/migrations/versions/002_add_file_paths.py @@ -0,0 +1,29 @@ +"""Add original_file_path and processed_file_path to FileRecord + +Revision ID: 002_add_file_paths +Revises: 001_file_processing_steps +Create Date: 2026-02-11 + +""" + +from typing import Sequence, Union + +import sqlalchemy as sa +from alembic import op + +# revision identifiers, used by Alembic. +revision: str = "002_add_file_paths" +down_revision: Union[str, None] = "001_file_processing_steps" +depends_on: Union[str, None] = None + + +def upgrade() -> None: + """Add original_file_path and processed_file_path columns to files table.""" + op.add_column("files", sa.Column("original_file_path", sa.String(), nullable=True)) + op.add_column("files", sa.Column("processed_file_path", sa.String(), nullable=True)) + + +def downgrade() -> None: + """Remove original_file_path and processed_file_path columns from files table.""" + op.drop_column("files", "processed_file_path") + op.drop_column("files", "original_file_path")