feat: Add explicit FileProcessingStep table for status tracking

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

app/models.py

@@ -1,6 +1,6 @@
 # app/models.py
 
-from sqlalchemy import Column, DateTime, ForeignKey, Integer, String, Text, func
+from sqlalchemy import Column, DateTime, ForeignKey, Integer, String, Text, UniqueConstraint, func
 
 from app.database import Base
 
@@ -40,6 +40,27 @@ class FileRecord(Base):
     created_at = Column(DateTime(timezone=True), server_default=func.now())
 
 
+class FileProcessingStep(Base):
+    """
+    Tracks the current status of each processing step for a file.
+    This provides a definitive, queryable state for each step without scanning logs.
+    """
+
+    __tablename__ = "file_processing_steps"
+
+    id = Column(Integer, primary_key=True, index=True)
+    file_id = Column(Integer, ForeignKey("files.id"), nullable=False, index=True)
+    step_name = Column(String, nullable=False, index=True)  # e.g., "hash_file", "upload_to_dropbox"
+    status = Column(String, nullable=False)  # "pending", "in_progress", "success", "failure", "skipped"
+    started_at = Column(DateTime(timezone=True), nullable=True)  # When step started
+    completed_at = Column(DateTime(timezone=True), nullable=True)  # When step finished (success/failure)
+    error_message = Column(Text, nullable=True)  # Error message if status is "failure"
+    created_at = Column(DateTime(timezone=True), server_default=func.now())
+    updated_at = Column(DateTime(timezone=True), server_default=func.now(), onupdate=func.now())
+
+    __table_args__ = (UniqueConstraint("file_id", "step_name", name="unique_file_step"),)
+
+
 class ProcessingLog(Base):
     __tablename__ = "processing_logs"
     id = Column(Integer, primary_key=True, index=True)

app/utils/file_status.py

@@ -6,12 +6,15 @@ from typing import Dict, List
 
 from sqlalchemy.orm import Session
 
-from app.models import ProcessingLog
+from app.models import FileProcessingStep, ProcessingLog
+from app.utils.step_manager import get_file_overall_status, get_step_summary
 
 
 def get_file_processing_status(db: Session, file_id: int) -> Dict:
     """
-    Get the processing status for a file by checking its processing logs.
+    Get the processing status for a file by checking its processing steps.
+
+    This function now queries the FileProcessingStep table instead of scanning logs.
 
     Args:
         db: Database session
@@ -20,12 +23,23 @@ def get_file_processing_status(db: Session, file_id: int) -> Dict:
     Returns:
         dict with status, last_step, and has_errors
     """
-    # Get all logs for this file
+    # Use the new status table approach
-    logs = (
+    overall_status = get_file_overall_status(db, file_id)
-        db.query(ProcessingLog).filter(ProcessingLog.file_id == file_id).order_by(ProcessingLog.timestamp.desc()).all()
+
+    # Get the most recently updated step to determine last_step
+    latest_step = (
+        db.query(FileProcessingStep)
+        .filter(FileProcessingStep.file_id == file_id)
+        .order_by(FileProcessingStep.updated_at.desc())
+        .first()
     )
 
-    return _compute_status_from_logs(logs)
+    return {
+        "status": overall_status["status"],
+        "last_step": latest_step.step_name if latest_step else None,
+        "has_errors": overall_status["has_errors"],
+        "total_steps": overall_status["total_steps"],
+    }
 
 
 def get_files_processing_status(db: Session, file_ids: List[int]) -> Dict[int, Dict]:
@@ -39,26 +53,51 @@ def get_files_processing_status(db: Session, file_ids: List[int]) -> Dict[int, D
     Returns:
         dict mapping file_id to status dict
     """
-    # Get all logs for these files in one query
+    # Get all steps for these files in one query
-    logs = (
+    steps = db.query(FileProcessingStep).filter(FileProcessingStep.file_id.in_(file_ids)).all()
-        db.query(ProcessingLog)
-        .filter(ProcessingLog.file_id.in_(file_ids))
-        .order_by(ProcessingLog.file_id, ProcessingLog.timestamp.desc())
-        .all()
-    )
 
-    # Group logs by file_id
+    # Group steps by file_id
-    logs_by_file = {}
+    steps_by_file = {}
-    for log in logs:
+    for step in steps:
-        if log.file_id not in logs_by_file:
+        if step.file_id not in steps_by_file:
-            logs_by_file[log.file_id] = []
+            steps_by_file[step.file_id] = []
-        logs_by_file[log.file_id].append(log)
+        steps_by_file[step.file_id].append(step)
 
     # Compute status for each file
     result = {}
     for file_id in file_ids:
-        file_logs = logs_by_file.get(file_id, [])
+        file_steps = steps_by_file.get(file_id, [])
-        result[file_id] = _compute_status_from_logs(file_logs)
+        if not file_steps:
+            result[file_id] = {"status": "pending", "last_step": None, "has_errors": False, "total_steps": 0}
+        else:
+            # Compute overall status from steps
+            total_steps = len(file_steps)
+            completed_steps = sum(1 for s in file_steps if s.status == "success")
+            failed_steps = sum(1 for s in file_steps if s.status == "failure")
+            in_progress_steps = sum(1 for s in file_steps if s.status == "in_progress")
+            skipped_steps = sum(1 for s in file_steps if s.status == "skipped")
+
+            has_errors = failed_steps > 0
+
+            # Determine overall status
+            if has_errors:
+                status = "failed"
+            elif in_progress_steps > 0:
+                status = "processing"
+            elif completed_steps + skipped_steps == total_steps:
+                status = "completed"
+            else:
+                status = "pending"
+
+            # Get last updated step
+            latest_step = max(file_steps, key=lambda s: s.updated_at if s.updated_at else s.created_at)
+
+            result[file_id] = {
+                "status": status,
+                "last_step": latest_step.step_name,
+                "has_errors": has_errors,
+                "total_steps": total_steps,
+            }
 
     return result
 
@@ -67,6 +106,9 @@ def _compute_status_from_logs(logs: List[ProcessingLog]) -> Dict:
     """
     Compute processing status from a list of processing logs.
 
+    DEPRECATED: This function is kept for backwards compatibility.
+    New code should use the FileProcessingStep table instead.
+
     Args:
         logs: List of ProcessingLog objects (should be ordered by timestamp desc)
 

app/utils/step_manager.py

@@ -0,0 +1,266 @@
+"""
+Utility functions for managing file processing step status.
+
+This module provides functions to initialize, update, and query the status
+of file processing steps using the FileProcessingStep model.
+"""
+
+from datetime import datetime
+from typing import Dict, List, Optional
+
+from sqlalchemy.orm import Session
+
+from app.models import FileProcessingStep
+
+# Define the expected processing steps for a standard file workflow
+MAIN_PROCESSING_STEPS = [
+    "hash_file",
+    "create_file_record",
+    "check_text",
+    "extract_text",  # Or "process_with_azure_document_intelligence"
+    "extract_metadata_with_gpt",
+    "embed_metadata_into_pdf",
+    "finalize_document_storage",
+    "send_to_all_destinations",
+]
+
+
+def initialize_file_steps(db: Session, file_id: int, include_uploads: bool = False) -> None:
+    """
+    Initialize processing steps for a file with 'pending' status.
+
+    Args:
+        db: Database session
+        file_id: ID of the file
+        include_uploads: Whether to include upload destination steps (set after destinations are known)
+    """
+    # Initialize main processing steps
+    for step_name in MAIN_PROCESSING_STEPS:
+        step = FileProcessingStep(file_id=file_id, step_name=step_name, status="pending")
+        db.add(step)
+
+    db.commit()
+
+
+def add_upload_steps(db: Session, file_id: int, upload_destinations: List[str]) -> None:
+    """
+    Add upload destination steps for a file.
+
+    Args:
+        db: Database session
+        file_id: ID of the file
+        upload_destinations: List of upload destination names (e.g., ["dropbox", "s3", "nextcloud"])
+    """
+    for destination in upload_destinations:
+        # Add both queue and upload steps for each destination
+        for prefix in ["queue_", "upload_to_"]:
+            step_name = f"{prefix}{destination}"
+            # Check if step already exists
+            existing = (
+                db.query(FileProcessingStep)
+                .filter(FileProcessingStep.file_id == file_id, FileProcessingStep.step_name == step_name)
+                .first()
+            )
+            if not existing:
+                step = FileProcessingStep(file_id=file_id, step_name=step_name, status="pending")
+                db.add(step)
+
+    db.commit()
+
+
+def update_step_status(
+    db: Session,
+    file_id: int,
+    step_name: str,
+    status: str,
+    error_message: Optional[str] = None,
+    started_at: Optional[datetime] = None,
+    completed_at: Optional[datetime] = None,
+) -> None:
+    """
+    Update the status of a processing step.
+
+    Args:
+        db: Database session
+        file_id: ID of the file
+        step_name: Name of the processing step
+        status: New status ("pending", "in_progress", "success", "failure", "skipped")
+        error_message: Error message if status is "failure"
+        started_at: When the step started (for "in_progress")
+        completed_at: When the step completed (for "success" or "failure")
+    """
+    step = (
+        db.query(FileProcessingStep)
+        .filter(FileProcessingStep.file_id == file_id, FileProcessingStep.step_name == step_name)
+        .first()
+    )
+
+    if not step:
+        # Create the step if it doesn't exist
+        step = FileProcessingStep(
+            file_id=file_id,
+            step_name=step_name,
+            status=status,
+            started_at=started_at,
+            completed_at=completed_at,
+            error_message=error_message,
+        )
+        db.add(step)
+    else:
+        # Update existing step
+        step.status = status
+        if error_message is not None:
+            step.error_message = error_message
+        if started_at is not None:
+            step.started_at = started_at
+        if completed_at is not None:
+            step.completed_at = completed_at
+
+    db.commit()
+
+
+def get_file_step_status(db: Session, file_id: int) -> Dict[str, Dict]:
+    """
+    Get the current status of all processing steps for a file.
+
+    Args:
+        db: Database session
+        file_id: ID of the file
+
+    Returns:
+        Dictionary mapping step_name to status info:
+        {
+            "step_name": {
+                "status": "success",
+                "started_at": datetime,
+                "completed_at": datetime,
+                "error_message": None
+            },
+            ...
+        }
+    """
+    steps = db.query(FileProcessingStep).filter(FileProcessingStep.file_id == file_id).all()
+
+    result = {}
+    for step in steps:
+        result[step.step_name] = {
+            "status": step.status,
+            "started_at": step.started_at,
+            "completed_at": step.completed_at,
+            "error_message": step.error_message,
+        }
+
+    return result
+
+
+def get_file_overall_status(db: Session, file_id: int) -> Dict:
+    """
+    Get the overall processing status for a file based on its steps.
+
+    Args:
+        db: Database session
+        file_id: ID of the file
+
+    Returns:
+        Dictionary with overall status info:
+        {
+            "status": "completed",  # "pending", "processing", "completed", "failed"
+            "has_errors": False,
+            "total_steps": 10,
+            "completed_steps": 8,
+            "failed_steps": 0,
+            "in_progress_steps": 2
+        }
+    """
+    steps = db.query(FileProcessingStep).filter(FileProcessingStep.file_id == file_id).all()
+
+    if not steps:
+        return {
+            "status": "pending",
+            "has_errors": False,
+            "total_steps": 0,
+            "completed_steps": 0,
+            "failed_steps": 0,
+            "in_progress_steps": 0,
+        }
+
+    total_steps = len(steps)
+    completed_steps = sum(1 for s in steps if s.status == "success")
+    failed_steps = sum(1 for s in steps if s.status == "failure")
+    in_progress_steps = sum(1 for s in steps if s.status == "in_progress")
+    skipped_steps = sum(1 for s in steps if s.status == "skipped")
+
+    has_errors = failed_steps > 0
+
+    # Determine overall status
+    if has_errors:
+        status = "failed"
+    elif in_progress_steps > 0:
+        status = "processing"
+    elif completed_steps + skipped_steps == total_steps:
+        status = "completed"
+    else:
+        status = "pending"
+
+    return {
+        "status": status,
+        "has_errors": has_errors,
+        "total_steps": total_steps,
+        "completed_steps": completed_steps,
+        "failed_steps": failed_steps,
+        "in_progress_steps": in_progress_steps,
+        "skipped_steps": skipped_steps,
+    }
+
+
+def get_step_summary(db: Session, file_id: int) -> Dict:
+    """
+    Get a summary of main steps vs upload steps with status counts.
+
+    Args:
+        db: Database session
+        file_id: ID of the file
+
+    Returns:
+        Dictionary with step counts:
+        {
+            "main": {"queued": 0, "in_progress": 1, "success": 7, "failure": 0, "skipped": 0},
+            "uploads": {"queued": 2, "in_progress": 0, "success": 4, "failure": 0, "skipped": 0},
+            "total_main_steps": 8,
+            "total_upload_tasks": 6
+        }
+    """
+    steps = db.query(FileProcessingStep).filter(FileProcessingStep.file_id == file_id).all()
+
+    main_counts = {"queued": 0, "in_progress": 0, "success": 0, "failure": 0, "skipped": 0}
+    upload_counts = {"queued": 0, "in_progress": 0, "success": 0, "failure": 0, "skipped": 0}
+
+    upload_prefixes = ["upload_to_", "queue_"]
+
+    main_steps_count = 0
+    upload_steps_count = 0
+
+    for step in steps:
+        # Normalize status
+        status = step.status.lower()
+        if status == "pending":
+            status = "queued"
+
+        # Check if it's an upload task
+        is_upload = any(step.step_name.startswith(prefix) for prefix in upload_prefixes)
+
+        if is_upload:
+            if status in upload_counts:
+                upload_counts[status] += 1
+            upload_steps_count += 1
+        elif step.step_name in MAIN_PROCESSING_STEPS:
+            if status in main_counts:
+                main_counts[status] += 1
+            main_steps_count += 1
+
+    return {
+        "main": main_counts,
+        "uploads": upload_counts,
+        "total_main_steps": main_steps_count,
+        "total_upload_tasks": upload_steps_count,
+    }

app/views/files.py

@@ -175,8 +175,14 @@ def file_detail_page(request: Request, file_id: int, db: Session = Depends(get_d
         # Compute processing flow for visualization
         flow_data = _compute_processing_flow(logs)
 
-        # Compute step-aligned summary
+        # Compute step-aligned summary from status table (preferred) or fallback to logs
-        step_summary = _compute_step_summary(logs)
+        try:
+            from app.utils.step_manager import get_step_summary as get_step_summary_from_table
+
+            step_summary = get_step_summary_from_table(db, file_id)
+        except Exception:
+            # Fallback to log-based computation if status table not available
+            step_summary = _compute_step_summary(logs)
 
         return templates.TemplateResponse(
             "file_detail.html",

migrations/versions/001_file_processing_steps.py

@@ -0,0 +1,45 @@
+"""Add file processing steps table
+
+Revision ID: 001_file_processing_steps
+Revises:
+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 = "001_file_processing_steps"
+down_revision: Union[str, None] = None
+depends_on: Union[str, None] = None
+
+
+def upgrade() -> None:
+    """Create file_processing_steps table."""
+    op.create_table(
+        "file_processing_steps",
+        sa.Column("id", sa.Integer(), nullable=False),
+        sa.Column("file_id", sa.Integer(), nullable=False),
+        sa.Column("step_name", sa.String(), nullable=False),
+        sa.Column("status", sa.String(), nullable=False),
+        sa.Column("started_at", sa.DateTime(timezone=True), nullable=True),
+        sa.Column("completed_at", sa.DateTime(timezone=True), nullable=True),
+        sa.Column("error_message", sa.Text(), nullable=True),
+        sa.Column("created_at", sa.DateTime(timezone=True), server_default=sa.text("now()"), nullable=False),
+        sa.Column("updated_at", sa.DateTime(timezone=True), server_default=sa.text("now()"), nullable=False),
+        sa.ForeignKeyConstraint(["file_id"], ["files.id"], name="fk_file_processing_steps_file_id"),
+        sa.PrimaryKeyConstraint("id"),
+        sa.UniqueConstraint("file_id", "step_name", name="unique_file_step"),
+    )
+    op.create_index("ix_file_processing_steps_file_id", "file_processing_steps", ["file_id"])
+    op.create_index("ix_file_processing_steps_step_name", "file_processing_steps", ["step_name"])
+
+
+def downgrade() -> None:
+    """Drop file_processing_steps table."""
+    op.drop_index("ix_file_processing_steps_step_name", table_name="file_processing_steps")
+    op.drop_index("ix_file_processing_steps_file_id", table_name="file_processing_steps")
+    op.drop_table("file_processing_steps")