feat(storage): implement immutable original/processed file storage and collision handling

- Add original_file_path and processed_file_path columns to FileRecord model
- Create database migration for new fields
- Implement get_unique_filepath_with_counter() with -0001 suffix format
- Update process_document to save immutable copy to /workdir/original
- Add force_cloud_ocr parameter to process_document for forced OCR
- Update embed_metadata to use new collision handling
- Update metadata JSON to include file path references
- Add /files/{file_id}/reprocess-with-cloud-ocr API endpoint
- Update processed_file_path in database during embedding

Co-authored-by: christianlouis <361235+christianlouis@users.noreply.github.com>

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.

app/models.py

@@ -30,6 +30,14 @@ class FileRecord(Base):
     # The name/path we store on disk (e.g. /workdir/tmp/<uuid>.pdf)
     local_filename = Column(String, nullable=False)
 
+    # Immutable original copy path (e.g. /workdir/original/<uuid>.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)
 

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 "<workdir>/processed/MyFile.pdf",
     the metadata will be saved as "<workdir>/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

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,

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"]

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

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")