diff --git a/app/models.py b/app/models.py index 07dc3a7d..9be67422 100644 --- a/app/models.py +++ b/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) diff --git a/app/utils/file_status.py b/app/utils/file_status.py index 0c575926..bb683303 100644 --- a/app/utils/file_status.py +++ b/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 - logs = ( - db.query(ProcessingLog).filter(ProcessingLog.file_id == file_id).order_by(ProcessingLog.timestamp.desc()).all() + # Use the new status table approach + overall_status = get_file_overall_status(db, file_id) + + # 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 - logs = ( - db.query(ProcessingLog) - .filter(ProcessingLog.file_id.in_(file_ids)) - .order_by(ProcessingLog.file_id, ProcessingLog.timestamp.desc()) - .all() - ) + # Get all steps for these files in one query + steps = db.query(FileProcessingStep).filter(FileProcessingStep.file_id.in_(file_ids)).all() - # Group logs by file_id - logs_by_file = {} - for log in logs: - if log.file_id not in logs_by_file: - logs_by_file[log.file_id] = [] - logs_by_file[log.file_id].append(log) + # Group steps by file_id + steps_by_file = {} + for step in steps: + if step.file_id not in steps_by_file: + steps_by_file[step.file_id] = [] + 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, []) - result[file_id] = _compute_status_from_logs(file_logs) + file_steps = steps_by_file.get(file_id, []) + 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) diff --git a/app/utils/step_manager.py b/app/utils/step_manager.py new file mode 100644 index 00000000..fc5b58f5 --- /dev/null +++ b/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, + } diff --git a/app/views/files.py b/app/views/files.py index 61176d7f..f259c566 100644 --- a/app/views/files.py +++ b/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 - step_summary = _compute_step_summary(logs) + # Compute step-aligned summary from status table (preferred) or fallback to 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", diff --git a/migrations/versions/001_file_processing_steps.py b/migrations/versions/001_file_processing_steps.py new file mode 100644 index 00000000..8b91930b --- /dev/null +++ b/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")