Merge pull request #248 from christianlouis/copilot/fix-dashboard-file-status
Fix dashboard status tracking with explicit FileProcessingStep table
This commit is contained in:
+22
-1
@@ -1,6 +1,6 @@
|
|||||||
# app/models.py
|
# 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
|
from app.database import Base
|
||||||
|
|
||||||
@@ -40,6 +40,27 @@ class FileRecord(Base):
|
|||||||
created_at = Column(DateTime(timezone=True), server_default=func.now())
|
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):
|
class ProcessingLog(Base):
|
||||||
__tablename__ = "processing_logs"
|
__tablename__ = "processing_logs"
|
||||||
id = Column(Integer, primary_key=True, index=True)
|
id = Column(Integer, primary_key=True, index=True)
|
||||||
|
|||||||
+63
-21
@@ -6,12 +6,15 @@ from typing import Dict, List
|
|||||||
|
|
||||||
from sqlalchemy.orm import Session
|
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:
|
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:
|
Args:
|
||||||
db: Database session
|
db: Database session
|
||||||
@@ -20,12 +23,23 @@ def get_file_processing_status(db: Session, file_id: int) -> Dict:
|
|||||||
Returns:
|
Returns:
|
||||||
dict with status, last_step, and has_errors
|
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]:
|
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:
|
Returns:
|
||||||
dict mapping file_id to status dict
|
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
|
# Compute status for each file
|
||||||
result = {}
|
result = {}
|
||||||
for file_id in file_ids:
|
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
|
return result
|
||||||
|
|
||||||
@@ -67,6 +106,9 @@ def _compute_status_from_logs(logs: List[ProcessingLog]) -> Dict:
|
|||||||
"""
|
"""
|
||||||
Compute processing status from a list of processing logs.
|
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:
|
Args:
|
||||||
logs: List of ProcessingLog objects (should be ordered by timestamp desc)
|
logs: List of ProcessingLog objects (should be ordered by timestamp desc)
|
||||||
|
|
||||||
|
|||||||
@@ -0,0 +1,326 @@
|
|||||||
|
"""
|
||||||
|
Migration utility to backfill FileProcessingStep table from existing ProcessingLog entries.
|
||||||
|
|
||||||
|
This script analyzes historical logs and populates the FileProcessingStep table
|
||||||
|
for files that were processed before the status tracking table was created.
|
||||||
|
"""
|
||||||
|
|
||||||
|
import logging
|
||||||
|
from datetime import datetime
|
||||||
|
from typing import Dict, List
|
||||||
|
|
||||||
|
from sqlalchemy import func
|
||||||
|
from sqlalchemy.orm import Session
|
||||||
|
|
||||||
|
from app.models import FileProcessingStep, FileRecord, ProcessingLog
|
||||||
|
from app.utils.step_manager import MAIN_PROCESSING_STEPS
|
||||||
|
|
||||||
|
logger = logging.getLogger(__name__)
|
||||||
|
|
||||||
|
|
||||||
|
def migrate_logs_to_steps(db: Session, file_id: int, dry_run: bool = False) -> Dict:
|
||||||
|
"""
|
||||||
|
Migrate ProcessingLog entries to FileProcessingStep for a single file.
|
||||||
|
|
||||||
|
Analyzes all logs for the file and creates/updates FileProcessingStep entries
|
||||||
|
based on the latest status per step found in the logs.
|
||||||
|
|
||||||
|
Args:
|
||||||
|
db: Database session
|
||||||
|
file_id: ID of the file to migrate
|
||||||
|
dry_run: If True, don't commit changes, just report what would be done
|
||||||
|
|
||||||
|
Returns:
|
||||||
|
Dictionary with migration results:
|
||||||
|
{
|
||||||
|
"file_id": 123,
|
||||||
|
"steps_created": 5,
|
||||||
|
"steps_updated": 3,
|
||||||
|
"steps_skipped": 2,
|
||||||
|
"errors": []
|
||||||
|
}
|
||||||
|
"""
|
||||||
|
results = {"file_id": file_id, "steps_created": 0, "steps_updated": 0, "steps_skipped": 0, "errors": []}
|
||||||
|
|
||||||
|
try:
|
||||||
|
# Get all logs for this file, ordered by timestamp
|
||||||
|
logs = (
|
||||||
|
db.query(ProcessingLog)
|
||||||
|
.filter(ProcessingLog.file_id == file_id)
|
||||||
|
.order_by(ProcessingLog.timestamp.asc())
|
||||||
|
.all()
|
||||||
|
)
|
||||||
|
|
||||||
|
if not logs:
|
||||||
|
logger.info(f"No logs found for file {file_id}")
|
||||||
|
return results
|
||||||
|
|
||||||
|
# Parse logs to extract latest status per step
|
||||||
|
step_states = _parse_logs_to_step_states(logs)
|
||||||
|
|
||||||
|
# Create or update FileProcessingStep entries
|
||||||
|
for step_name, state in step_states.items():
|
||||||
|
existing_step = (
|
||||||
|
db.query(FileProcessingStep)
|
||||||
|
.filter(FileProcessingStep.file_id == file_id, FileProcessingStep.step_name == step_name)
|
||||||
|
.first()
|
||||||
|
)
|
||||||
|
|
||||||
|
if existing_step:
|
||||||
|
# Update existing step only if it differs
|
||||||
|
if (
|
||||||
|
existing_step.status != state["status"]
|
||||||
|
or existing_step.started_at != state["started_at"]
|
||||||
|
or existing_step.completed_at != state["completed_at"]
|
||||||
|
):
|
||||||
|
logger.info(
|
||||||
|
f"Updating step {step_name} for file {file_id}: {existing_step.status} -> {state['status']}"
|
||||||
|
)
|
||||||
|
existing_step.status = state["status"]
|
||||||
|
existing_step.started_at = state["started_at"]
|
||||||
|
existing_step.completed_at = state["completed_at"]
|
||||||
|
existing_step.error_message = state["error_message"]
|
||||||
|
results["steps_updated"] += 1
|
||||||
|
else:
|
||||||
|
logger.debug(f"Step {step_name} for file {file_id} already up to date")
|
||||||
|
results["steps_skipped"] += 1
|
||||||
|
else:
|
||||||
|
# Create new step
|
||||||
|
logger.info(f"Creating step {step_name} for file {file_id} with status {state['status']}")
|
||||||
|
new_step = FileProcessingStep(
|
||||||
|
file_id=file_id,
|
||||||
|
step_name=step_name,
|
||||||
|
status=state["status"],
|
||||||
|
started_at=state["started_at"],
|
||||||
|
completed_at=state["completed_at"],
|
||||||
|
error_message=state["error_message"],
|
||||||
|
)
|
||||||
|
db.add(new_step)
|
||||||
|
results["steps_created"] += 1
|
||||||
|
|
||||||
|
if not dry_run:
|
||||||
|
db.commit()
|
||||||
|
logger.info(
|
||||||
|
f"Migration complete for file {file_id}: "
|
||||||
|
f"{results['steps_created']} created, "
|
||||||
|
f"{results['steps_updated']} updated, "
|
||||||
|
f"{results['steps_skipped']} skipped"
|
||||||
|
)
|
||||||
|
else:
|
||||||
|
db.rollback()
|
||||||
|
logger.info(f"Dry run complete for file {file_id} (no changes committed)")
|
||||||
|
|
||||||
|
except Exception as e:
|
||||||
|
logger.error(f"Error migrating file {file_id}: {str(e)}")
|
||||||
|
results["errors"].append(str(e))
|
||||||
|
db.rollback()
|
||||||
|
|
||||||
|
return results
|
||||||
|
|
||||||
|
|
||||||
|
def _parse_logs_to_step_states(logs: List[ProcessingLog]) -> Dict[str, Dict]:
|
||||||
|
"""
|
||||||
|
Parse logs to determine the final state of each step.
|
||||||
|
|
||||||
|
Processes logs in chronological order to build a state machine
|
||||||
|
tracking the progression of each step.
|
||||||
|
|
||||||
|
Args:
|
||||||
|
logs: List of ProcessingLog entries ordered by timestamp
|
||||||
|
|
||||||
|
Returns:
|
||||||
|
Dictionary mapping step_name to state dict:
|
||||||
|
{
|
||||||
|
"step_name": {
|
||||||
|
"status": "success",
|
||||||
|
"started_at": datetime,
|
||||||
|
"completed_at": datetime,
|
||||||
|
"error_message": None
|
||||||
|
}
|
||||||
|
}
|
||||||
|
"""
|
||||||
|
step_states = {}
|
||||||
|
|
||||||
|
for log in logs:
|
||||||
|
step_name = log.step_name
|
||||||
|
status = log.status.lower()
|
||||||
|
|
||||||
|
# Initialize step state if first time seeing this step
|
||||||
|
if step_name not in step_states:
|
||||||
|
step_states[step_name] = {
|
||||||
|
"status": "pending",
|
||||||
|
"started_at": None,
|
||||||
|
"completed_at": None,
|
||||||
|
"error_message": None,
|
||||||
|
}
|
||||||
|
|
||||||
|
# Update state based on log entry
|
||||||
|
if status == "in_progress":
|
||||||
|
# Step started
|
||||||
|
if step_states[step_name]["started_at"] is None:
|
||||||
|
step_states[step_name]["started_at"] = log.timestamp
|
||||||
|
step_states[step_name]["status"] = "in_progress"
|
||||||
|
|
||||||
|
elif status == "success":
|
||||||
|
# Step completed successfully
|
||||||
|
if step_states[step_name]["started_at"] is None:
|
||||||
|
# If we never saw in_progress, assume it started around this time
|
||||||
|
step_states[step_name]["started_at"] = log.timestamp
|
||||||
|
step_states[step_name]["status"] = "success"
|
||||||
|
step_states[step_name]["completed_at"] = log.timestamp
|
||||||
|
step_states[step_name]["error_message"] = None
|
||||||
|
|
||||||
|
elif status == "failure":
|
||||||
|
# Step failed
|
||||||
|
if step_states[step_name]["started_at"] is None:
|
||||||
|
step_states[step_name]["started_at"] = log.timestamp
|
||||||
|
step_states[step_name]["status"] = "failure"
|
||||||
|
step_states[step_name]["completed_at"] = log.timestamp
|
||||||
|
step_states[step_name]["error_message"] = log.message
|
||||||
|
|
||||||
|
elif status in ["pending", "queued"]:
|
||||||
|
# Only update if step hasn't started yet
|
||||||
|
if step_states[step_name]["status"] == "pending":
|
||||||
|
step_states[step_name]["status"] = "pending"
|
||||||
|
|
||||||
|
# Note: If we see success/failure after a previous success/failure,
|
||||||
|
# the later one wins (represents retry/reprocessing)
|
||||||
|
|
||||||
|
return step_states
|
||||||
|
|
||||||
|
|
||||||
|
def migrate_all_files(db: Session, batch_size: int = 100, dry_run: bool = False) -> Dict:
|
||||||
|
"""
|
||||||
|
Migrate all files that have logs but no FileProcessingStep entries.
|
||||||
|
|
||||||
|
Args:
|
||||||
|
db: Database session
|
||||||
|
batch_size: Number of files to process in each batch
|
||||||
|
dry_run: If True, don't commit changes
|
||||||
|
|
||||||
|
Returns:
|
||||||
|
Dictionary with overall migration statistics:
|
||||||
|
{
|
||||||
|
"total_files": 150,
|
||||||
|
"files_migrated": 145,
|
||||||
|
"files_failed": 5,
|
||||||
|
"total_steps_created": 1200,
|
||||||
|
"total_steps_updated": 350,
|
||||||
|
"errors": [...]
|
||||||
|
}
|
||||||
|
"""
|
||||||
|
summary = {
|
||||||
|
"total_files": 0,
|
||||||
|
"files_migrated": 0,
|
||||||
|
"files_failed": 0,
|
||||||
|
"total_steps_created": 0,
|
||||||
|
"total_steps_updated": 0,
|
||||||
|
"total_steps_skipped": 0,
|
||||||
|
"errors": [],
|
||||||
|
}
|
||||||
|
|
||||||
|
# Find all files that have logs but no steps
|
||||||
|
files_with_logs = db.query(ProcessingLog.file_id).distinct().all()
|
||||||
|
file_ids_with_logs = {file_id for (file_id,) in files_with_logs if file_id is not None}
|
||||||
|
|
||||||
|
files_with_steps = db.query(FileProcessingStep.file_id).distinct().all()
|
||||||
|
file_ids_with_steps = {file_id for (file_id,) in files_with_steps}
|
||||||
|
|
||||||
|
files_to_migrate = list(file_ids_with_logs - file_ids_with_steps)
|
||||||
|
|
||||||
|
summary["total_files"] = len(files_to_migrate)
|
||||||
|
logger.info(f"Found {len(files_to_migrate)} files to migrate")
|
||||||
|
|
||||||
|
# Process in batches
|
||||||
|
for i in range(0, len(files_to_migrate), batch_size):
|
||||||
|
batch = files_to_migrate[i : i + batch_size]
|
||||||
|
logger.info(
|
||||||
|
f"Processing batch {i // batch_size + 1}: files {i + 1} to {min(i + batch_size, len(files_to_migrate))}"
|
||||||
|
)
|
||||||
|
|
||||||
|
for file_id in batch:
|
||||||
|
result = migrate_logs_to_steps(db, file_id, dry_run=dry_run)
|
||||||
|
|
||||||
|
if result["errors"]:
|
||||||
|
summary["files_failed"] += 1
|
||||||
|
summary["errors"].extend(result["errors"])
|
||||||
|
else:
|
||||||
|
summary["files_migrated"] += 1
|
||||||
|
summary["total_steps_created"] += result["steps_created"]
|
||||||
|
summary["total_steps_updated"] += result["steps_updated"]
|
||||||
|
summary["total_steps_skipped"] += result["steps_skipped"]
|
||||||
|
|
||||||
|
logger.info(
|
||||||
|
f"Migration summary: {summary['files_migrated']} files migrated, "
|
||||||
|
f"{summary['files_failed']} failed, "
|
||||||
|
f"{summary['total_steps_created']} steps created, "
|
||||||
|
f"{summary['total_steps_updated']} steps updated"
|
||||||
|
)
|
||||||
|
|
||||||
|
return summary
|
||||||
|
|
||||||
|
|
||||||
|
def verify_migration(db: Session, file_id: int) -> Dict:
|
||||||
|
"""
|
||||||
|
Verify that migration for a file is correct by comparing logs to steps.
|
||||||
|
|
||||||
|
Args:
|
||||||
|
db: Database session
|
||||||
|
file_id: ID of the file to verify
|
||||||
|
|
||||||
|
Returns:
|
||||||
|
Dictionary with verification results:
|
||||||
|
{
|
||||||
|
"file_id": 123,
|
||||||
|
"is_valid": True,
|
||||||
|
"discrepancies": [],
|
||||||
|
"log_steps": ["step1", "step2"],
|
||||||
|
"table_steps": ["step1", "step2"]
|
||||||
|
}
|
||||||
|
"""
|
||||||
|
result = {"file_id": file_id, "is_valid": True, "discrepancies": [], "log_steps": [], "table_steps": []}
|
||||||
|
|
||||||
|
# Get logs and parse them
|
||||||
|
logs = (
|
||||||
|
db.query(ProcessingLog).filter(ProcessingLog.file_id == file_id).order_by(ProcessingLog.timestamp.asc()).all()
|
||||||
|
)
|
||||||
|
|
||||||
|
if not logs:
|
||||||
|
result["discrepancies"].append("No logs found for file")
|
||||||
|
result["is_valid"] = False
|
||||||
|
return result
|
||||||
|
|
||||||
|
expected_states = _parse_logs_to_step_states(logs)
|
||||||
|
result["log_steps"] = sorted(expected_states.keys())
|
||||||
|
|
||||||
|
# Get actual steps from table
|
||||||
|
actual_steps = db.query(FileProcessingStep).filter(FileProcessingStep.file_id == file_id).all()
|
||||||
|
result["table_steps"] = sorted([s.step_name for s in actual_steps])
|
||||||
|
|
||||||
|
# Compare
|
||||||
|
actual_states = {s.step_name: s for s in actual_steps}
|
||||||
|
|
||||||
|
# Check for missing steps
|
||||||
|
for step_name in expected_states:
|
||||||
|
if step_name not in actual_states:
|
||||||
|
result["discrepancies"].append(f"Step '{step_name}' missing from table")
|
||||||
|
result["is_valid"] = False
|
||||||
|
continue
|
||||||
|
|
||||||
|
# Compare status
|
||||||
|
expected = expected_states[step_name]
|
||||||
|
actual = actual_states[step_name]
|
||||||
|
|
||||||
|
if expected["status"] != actual.status:
|
||||||
|
result["discrepancies"].append(
|
||||||
|
f"Step '{step_name}' status mismatch: " f"expected '{expected['status']}', got '{actual.status}'"
|
||||||
|
)
|
||||||
|
result["is_valid"] = False
|
||||||
|
|
||||||
|
# Check for extra steps in table
|
||||||
|
for step_name in actual_states:
|
||||||
|
if step_name not in expected_states:
|
||||||
|
result["discrepancies"].append(f"Extra step '{step_name}' in table not found in logs")
|
||||||
|
# Not marking as invalid since this might be intentional
|
||||||
|
|
||||||
|
return result
|
||||||
@@ -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,
|
||||||
|
}
|
||||||
+22
-13
@@ -175,8 +175,14 @@ def file_detail_page(request: Request, file_id: int, db: Session = Depends(get_d
|
|||||||
# Compute processing flow for visualization
|
# Compute processing flow for visualization
|
||||||
flow_data = _compute_processing_flow(logs)
|
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(
|
return templates.TemplateResponse(
|
||||||
"file_detail.html",
|
"file_detail.html",
|
||||||
@@ -328,6 +334,9 @@ def _compute_step_summary(logs):
|
|||||||
Compute a step-aligned summary from logs showing queued, success, and failure counts.
|
Compute a step-aligned summary from logs showing queued, success, and failure counts.
|
||||||
|
|
||||||
Returns a dictionary with main step counts and upload branch counts.
|
Returns a dictionary with main step counts and upload branch counts.
|
||||||
|
|
||||||
|
Note: This function is order-independent - it selects the latest status per step
|
||||||
|
based on timestamp, regardless of input log ordering.
|
||||||
"""
|
"""
|
||||||
# Count statuses for main processing steps (not uploads)
|
# Count statuses for main processing steps (not uploads)
|
||||||
main_steps = [
|
main_steps = [
|
||||||
@@ -347,9 +356,9 @@ def _compute_step_summary(logs):
|
|||||||
main_counts = {"queued": 0, "in_progress": 0, "success": 0, "failure": 0}
|
main_counts = {"queued": 0, "in_progress": 0, "success": 0, "failure": 0}
|
||||||
upload_counts = {"queued": 0, "in_progress": 0, "success": 0, "failure": 0}
|
upload_counts = {"queued": 0, "in_progress": 0, "success": 0, "failure": 0}
|
||||||
|
|
||||||
# Track latest status for each step (logs are ordered by timestamp desc)
|
# Track latest status for each step by comparing timestamps (order-independent)
|
||||||
main_steps_seen = {}
|
main_steps_seen = {} # {step_name: (timestamp, status)}
|
||||||
upload_tasks_seen = {}
|
upload_tasks_seen = {} # {step_name: (timestamp, status)}
|
||||||
|
|
||||||
for log in logs:
|
for log in logs:
|
||||||
step_name = log.step_name
|
step_name = log.step_name
|
||||||
@@ -363,21 +372,21 @@ def _compute_step_summary(logs):
|
|||||||
is_upload = any(step_name.startswith(prefix) for prefix in upload_prefixes)
|
is_upload = any(step_name.startswith(prefix) for prefix in upload_prefixes)
|
||||||
|
|
||||||
if is_upload:
|
if is_upload:
|
||||||
# Track latest status for each unique upload task (first seen is latest)
|
# Track latest status for each unique upload task by timestamp
|
||||||
if step_name not in upload_tasks_seen:
|
if step_name not in upload_tasks_seen or log.timestamp > upload_tasks_seen[step_name][0]:
|
||||||
upload_tasks_seen[step_name] = status
|
upload_tasks_seen[step_name] = (log.timestamp, status)
|
||||||
elif step_name in main_steps:
|
elif step_name in main_steps:
|
||||||
# Track latest status for main steps (first seen is latest)
|
# Track latest status for main steps by timestamp
|
||||||
if step_name not in main_steps_seen:
|
if step_name not in main_steps_seen or log.timestamp > main_steps_seen[step_name][0]:
|
||||||
main_steps_seen[step_name] = status
|
main_steps_seen[step_name] = (log.timestamp, status)
|
||||||
|
|
||||||
# Count main step statuses from latest status per step
|
# Count main step statuses from latest status per step
|
||||||
for task_status in main_steps_seen.values():
|
for _, task_status in main_steps_seen.values():
|
||||||
if task_status in main_counts:
|
if task_status in main_counts:
|
||||||
main_counts[task_status] += 1
|
main_counts[task_status] += 1
|
||||||
|
|
||||||
# Count upload task statuses from latest status per task
|
# Count upload task statuses from latest status per task
|
||||||
for task_status in upload_tasks_seen.values():
|
for _, task_status in upload_tasks_seen.values():
|
||||||
if task_status in upload_counts:
|
if task_status in upload_counts:
|
||||||
upload_counts[task_status] += 1
|
upload_counts[task_status] += 1
|
||||||
|
|
||||||
|
|||||||
@@ -0,0 +1,354 @@
|
|||||||
|
# File Processing Status Architecture
|
||||||
|
|
||||||
|
## Overview
|
||||||
|
|
||||||
|
DocuElevate now uses a **dual-table architecture** for tracking file processing:
|
||||||
|
|
||||||
|
1. **`FileProcessingStep`** - Current state tracking (NEW)
|
||||||
|
2. **`ProcessingLog`** - Historical audit trail (EXISTING, PRESERVED)
|
||||||
|
|
||||||
|
This separation provides both **fast status queries** and **complete audit history**.
|
||||||
|
|
||||||
|
## Architecture
|
||||||
|
|
||||||
|
### FileProcessingStep Table (Current State)
|
||||||
|
|
||||||
|
**Purpose**: Definitive source of truth for current processing status
|
||||||
|
|
||||||
|
**Structure**:
|
||||||
|
```sql
|
||||||
|
CREATE TABLE file_processing_steps (
|
||||||
|
id INTEGER PRIMARY KEY,
|
||||||
|
file_id INTEGER NOT NULL,
|
||||||
|
step_name VARCHAR NOT NULL,
|
||||||
|
status VARCHAR NOT NULL, -- 'pending', 'in_progress', 'success', 'failure', 'skipped'
|
||||||
|
started_at TIMESTAMP,
|
||||||
|
completed_at TIMESTAMP,
|
||||||
|
error_message TEXT,
|
||||||
|
created_at TIMESTAMP DEFAULT NOW(),
|
||||||
|
updated_at TIMESTAMP DEFAULT NOW(),
|
||||||
|
UNIQUE (file_id, step_name)
|
||||||
|
)
|
||||||
|
```
|
||||||
|
|
||||||
|
**Key Features**:
|
||||||
|
- One row per step per file
|
||||||
|
- Updated when step status changes
|
||||||
|
- No scanning required - direct lookups
|
||||||
|
- Indexed for fast queries
|
||||||
|
|
||||||
|
### ProcessingLog Table (Audit Trail)
|
||||||
|
|
||||||
|
**Purpose**: Complete historical record of all processing events
|
||||||
|
|
||||||
|
**Structure**: (Unchanged from original)
|
||||||
|
```sql
|
||||||
|
CREATE TABLE processing_logs (
|
||||||
|
id INTEGER PRIMARY KEY,
|
||||||
|
file_id INTEGER,
|
||||||
|
task_id VARCHAR,
|
||||||
|
step_name VARCHAR,
|
||||||
|
status VARCHAR,
|
||||||
|
message VARCHAR,
|
||||||
|
detail TEXT,
|
||||||
|
timestamp TIMESTAMP DEFAULT NOW()
|
||||||
|
)
|
||||||
|
```
|
||||||
|
|
||||||
|
**Key Features**:
|
||||||
|
- Multiple rows per step (one per event)
|
||||||
|
- Never deleted - complete history
|
||||||
|
- Used for debugging, compliance, audit
|
||||||
|
- Detailed error messages and diagnostics
|
||||||
|
|
||||||
|
## How They Work Together
|
||||||
|
|
||||||
|
### When a File is Created
|
||||||
|
|
||||||
|
```python
|
||||||
|
from app.utils.step_manager import initialize_file_steps
|
||||||
|
|
||||||
|
# Create file record
|
||||||
|
file_record = FileRecord(...)
|
||||||
|
db.add(file_record)
|
||||||
|
db.commit()
|
||||||
|
|
||||||
|
# Initialize expected processing steps
|
||||||
|
initialize_file_steps(db, file_record.id)
|
||||||
|
# Creates FileProcessingStep rows with status='pending'
|
||||||
|
```
|
||||||
|
|
||||||
|
### When a Worker Processes a Step
|
||||||
|
|
||||||
|
```python
|
||||||
|
from app.utils.step_manager import update_step_status
|
||||||
|
|
||||||
|
# Worker starts processing
|
||||||
|
update_step_status(
|
||||||
|
db, file_id, "hash_file", "in_progress",
|
||||||
|
started_at=datetime.now()
|
||||||
|
)
|
||||||
|
|
||||||
|
# Worker also logs to ProcessingLog (for history)
|
||||||
|
log = ProcessingLog(
|
||||||
|
file_id=file_id,
|
||||||
|
step_name="hash_file",
|
||||||
|
status="in_progress",
|
||||||
|
message="Starting file hash calculation"
|
||||||
|
)
|
||||||
|
db.add(log)
|
||||||
|
db.commit()
|
||||||
|
|
||||||
|
# Worker completes successfully
|
||||||
|
update_step_status(
|
||||||
|
db, file_id, "hash_file", "success",
|
||||||
|
completed_at=datetime.now()
|
||||||
|
)
|
||||||
|
|
||||||
|
# Also log completion
|
||||||
|
log = ProcessingLog(
|
||||||
|
file_id=file_id,
|
||||||
|
step_name="hash_file",
|
||||||
|
status="success",
|
||||||
|
message="Hash calculated successfully: abc123..."
|
||||||
|
)
|
||||||
|
db.add(log)
|
||||||
|
db.commit()
|
||||||
|
```
|
||||||
|
|
||||||
|
### When Dashboard Queries Status
|
||||||
|
|
||||||
|
```python
|
||||||
|
from app.utils.step_manager import get_file_overall_status, get_step_summary
|
||||||
|
|
||||||
|
# Fast lookup - no log scanning!
|
||||||
|
overall = get_file_overall_status(db, file_id)
|
||||||
|
# Returns: {"status": "processing", "has_errors": False, ...}
|
||||||
|
|
||||||
|
# Get detailed step breakdown
|
||||||
|
summary = get_step_summary(db, file_id)
|
||||||
|
# Returns: {
|
||||||
|
# "main": {"success": 3, "in_progress": 1, "pending": 4},
|
||||||
|
# "uploads": {"success": 2, "queued": 4}
|
||||||
|
# }
|
||||||
|
```
|
||||||
|
|
||||||
|
### When Viewing Processing History
|
||||||
|
|
||||||
|
```python
|
||||||
|
# Logs provide complete timeline
|
||||||
|
logs = db.query(ProcessingLog)\
|
||||||
|
.filter(ProcessingLog.file_id == file_id)\
|
||||||
|
.order_by(ProcessingLog.timestamp.asc())\
|
||||||
|
.all()
|
||||||
|
|
||||||
|
# Shows every event with timestamps, messages, details
|
||||||
|
# Example:
|
||||||
|
# 2024-01-15 10:00:00 | hash_file | in_progress | Starting...
|
||||||
|
# 2024-01-15 10:00:05 | hash_file | success | Hash: abc123
|
||||||
|
# 2024-01-15 10:00:06 | check_text | in_progress | Checking...
|
||||||
|
# ... etc
|
||||||
|
```
|
||||||
|
|
||||||
|
## Migration from Logs
|
||||||
|
|
||||||
|
For existing files with only ProcessingLog entries:
|
||||||
|
|
||||||
|
```python
|
||||||
|
from app.utils.migrate_logs_to_steps import migrate_logs_to_steps, migrate_all_files
|
||||||
|
|
||||||
|
# Migrate single file
|
||||||
|
result = migrate_logs_to_steps(db, file_id)
|
||||||
|
# Parses logs chronologically to reconstruct final state
|
||||||
|
|
||||||
|
# Migrate all files (batch processing)
|
||||||
|
summary = migrate_all_files(db, batch_size=100)
|
||||||
|
# Finds files with logs but no steps, migrates them all
|
||||||
|
|
||||||
|
# Dry run mode (test without committing)
|
||||||
|
result = migrate_logs_to_steps(db, file_id, dry_run=True)
|
||||||
|
```
|
||||||
|
|
||||||
|
The migration utility:
|
||||||
|
1. Reads all logs for a file in chronological order
|
||||||
|
2. Tracks state transitions (pending → in_progress → success/failure)
|
||||||
|
3. Determines final status per step
|
||||||
|
4. Creates FileProcessingStep entries
|
||||||
|
5. Preserves all original logs unchanged
|
||||||
|
|
||||||
|
## Benefits
|
||||||
|
|
||||||
|
### Performance
|
||||||
|
- **Old**: Scan all logs, dedupe, find latest → O(n log n)
|
||||||
|
- **New**: Single table lookup → O(1)
|
||||||
|
|
||||||
|
### Correctness
|
||||||
|
- **Old**: Dependent on log ordering, can miss retries
|
||||||
|
- **New**: Definitive state, always up-to-date
|
||||||
|
|
||||||
|
### Flexibility
|
||||||
|
- **Old**: Hard-coded step list, can't add dynamic steps
|
||||||
|
- **New**: Dynamic upload destinations, custom workflows
|
||||||
|
|
||||||
|
### Debugging
|
||||||
|
- **Old**: Only logs available
|
||||||
|
- **New**: Current state + complete history
|
||||||
|
|
||||||
|
## Example Workflow
|
||||||
|
|
||||||
|
```python
|
||||||
|
# 1. File uploaded
|
||||||
|
file = FileRecord(filehash="abc123", ...)
|
||||||
|
db.add(file)
|
||||||
|
db.commit()
|
||||||
|
|
||||||
|
# 2. Initialize steps
|
||||||
|
initialize_file_steps(db, file.id)
|
||||||
|
# FileProcessingStep: 9 rows with status='pending'
|
||||||
|
|
||||||
|
# 3. Add upload destinations (discovered later)
|
||||||
|
add_upload_steps(db, file.id, ["dropbox", "s3", "nextcloud"])
|
||||||
|
# FileProcessingStep: +6 rows with status='pending'
|
||||||
|
|
||||||
|
# 4. Worker processes each step
|
||||||
|
for step in ["hash_file", "create_file_record", "check_text", ...]:
|
||||||
|
# Start
|
||||||
|
update_step_status(db, file.id, step, "in_progress", started_at=now())
|
||||||
|
log_event(db, file.id, step, "in_progress", "Starting...")
|
||||||
|
|
||||||
|
# Do work...
|
||||||
|
|
||||||
|
# Complete
|
||||||
|
update_step_status(db, file.id, step, "success", completed_at=now())
|
||||||
|
log_event(db, file.id, step, "success", "Completed successfully")
|
||||||
|
|
||||||
|
# 5. Dashboard queries
|
||||||
|
status = get_file_overall_status(db, file.id)
|
||||||
|
# status="completed", all steps done
|
||||||
|
|
||||||
|
# 6. User views history
|
||||||
|
logs = get_processing_logs(db, file.id)
|
||||||
|
# Complete timeline with all events, messages, timestamps
|
||||||
|
```
|
||||||
|
|
||||||
|
## Status Flow
|
||||||
|
|
||||||
|
```
|
||||||
|
pending → in_progress → success
|
||||||
|
↓
|
||||||
|
failure
|
||||||
|
```
|
||||||
|
|
||||||
|
## Main Processing Steps
|
||||||
|
|
||||||
|
Defined in `app/utils/step_manager.py`:
|
||||||
|
```python
|
||||||
|
MAIN_PROCESSING_STEPS = [
|
||||||
|
"hash_file",
|
||||||
|
"create_file_record",
|
||||||
|
"check_text",
|
||||||
|
"extract_text",
|
||||||
|
"extract_metadata_with_gpt",
|
||||||
|
"embed_metadata_into_pdf",
|
||||||
|
"finalize_document_storage",
|
||||||
|
"send_to_all_destinations",
|
||||||
|
]
|
||||||
|
```
|
||||||
|
|
||||||
|
## Upload Steps
|
||||||
|
|
||||||
|
Dynamically created per destination:
|
||||||
|
- `queue_{destination}` - Queued for upload
|
||||||
|
- `upload_to_{destination}` - Actual upload
|
||||||
|
|
||||||
|
Example: Dropbox
|
||||||
|
- `queue_dropbox` - Added to upload queue
|
||||||
|
- `upload_to_dropbox` - Upload to Dropbox
|
||||||
|
|
||||||
|
## API Reference
|
||||||
|
|
||||||
|
### Step Manager Functions
|
||||||
|
|
||||||
|
```python
|
||||||
|
# Initialize steps for new file
|
||||||
|
initialize_file_steps(db: Session, file_id: int) -> None
|
||||||
|
|
||||||
|
# Add upload destination steps
|
||||||
|
add_upload_steps(db: Session, file_id: int, destinations: List[str]) -> None
|
||||||
|
|
||||||
|
# Update step status
|
||||||
|
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
|
||||||
|
|
||||||
|
# Query step status
|
||||||
|
get_file_step_status(db: Session, file_id: int) -> Dict[str, Dict]
|
||||||
|
get_file_overall_status(db: Session, file_id: int) -> Dict
|
||||||
|
get_step_summary(db: Session, file_id: int) -> Dict
|
||||||
|
```
|
||||||
|
|
||||||
|
### Migration Functions
|
||||||
|
|
||||||
|
```python
|
||||||
|
# Migrate single file
|
||||||
|
migrate_logs_to_steps(
|
||||||
|
db: Session,
|
||||||
|
file_id: int,
|
||||||
|
dry_run: bool = False
|
||||||
|
) -> Dict
|
||||||
|
|
||||||
|
# Migrate all files
|
||||||
|
migrate_all_files(
|
||||||
|
db: Session,
|
||||||
|
batch_size: int = 100,
|
||||||
|
dry_run: bool = False
|
||||||
|
) -> Dict
|
||||||
|
|
||||||
|
# Verify migration
|
||||||
|
verify_migration(db: Session, file_id: int) -> Dict
|
||||||
|
```
|
||||||
|
|
||||||
|
## Best Practices
|
||||||
|
|
||||||
|
1. **Always log AND update status**: Workers should do both
|
||||||
|
2. **Logs are append-only**: Never delete or modify logs
|
||||||
|
3. **Status is current state**: FileProcessingStep reflects latest status
|
||||||
|
4. **Initialize early**: Create steps when file is created
|
||||||
|
5. **Add uploads dynamically**: Add upload steps when destinations are known
|
||||||
|
6. **Handle retries**: update_step_status() handles overwriting previous status
|
||||||
|
7. **Preserve history**: Use logs for debugging and compliance
|
||||||
|
|
||||||
|
## Troubleshooting
|
||||||
|
|
||||||
|
### Status shows wrong state
|
||||||
|
- Check FileProcessingStep table directly
|
||||||
|
- Use `get_file_step_status()` to see all steps
|
||||||
|
- Verify workers are calling `update_step_status()`
|
||||||
|
|
||||||
|
### Missing steps
|
||||||
|
- Check if `initialize_file_steps()` was called
|
||||||
|
- For uploads, check if `add_upload_steps()` was called
|
||||||
|
- Use migration utility to backfill from logs
|
||||||
|
|
||||||
|
### Dashboard shows old data
|
||||||
|
- Clear any caching
|
||||||
|
- Verify status table is being queried, not logs
|
||||||
|
- Check `app/views/files.py` is using `get_step_summary_from_table()`
|
||||||
|
|
||||||
|
## Migration Checklist
|
||||||
|
|
||||||
|
For deploying to existing system:
|
||||||
|
|
||||||
|
- [ ] Apply database migration: `alembic upgrade head`
|
||||||
|
- [ ] Run migration utility: `migrate_all_files(db, dry_run=True)` to test
|
||||||
|
- [ ] Run actual migration: `migrate_all_files(db, dry_run=False)`
|
||||||
|
- [ ] Verify: Check a few files with `verify_migration()`
|
||||||
|
- [ ] Update workers to call `update_step_status()`
|
||||||
|
- [ ] Update file creation to call `initialize_file_steps()`
|
||||||
|
- [ ] Monitor dashboard for correct status display
|
||||||
@@ -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")
|
||||||
@@ -231,9 +231,7 @@ class TestSubtaskRetry:
|
|||||||
|
|
||||||
with patch("app.tasks.extract_metadata_with_gpt.extract_metadata_with_gpt") as mock_extract:
|
with patch("app.tasks.extract_metadata_with_gpt.extract_metadata_with_gpt") as mock_extract:
|
||||||
mock_extract.delay.return_value = mock_task
|
mock_extract.delay.return_value = mock_task
|
||||||
response = client.post(
|
response = client.post(f"/api/files/{file_record.id}/retry-subtask?subtask_name=embed_metadata_into_pdf")
|
||||||
f"/api/files/{file_record.id}/retry-subtask?subtask_name=embed_metadata_into_pdf"
|
|
||||||
)
|
|
||||||
assert response.status_code == 200
|
assert response.status_code == 200
|
||||||
data = response.json()
|
data = response.json()
|
||||||
assert data["status"] == "success"
|
assert data["status"] == "success"
|
||||||
@@ -503,21 +501,25 @@ class TestStepSummary:
|
|||||||
|
|
||||||
def test_summary_with_mixed_statuses(self):
|
def test_summary_with_mixed_statuses(self):
|
||||||
"""Test step summary with various statuses."""
|
"""Test step summary with various statuses."""
|
||||||
|
from datetime import datetime, timedelta
|
||||||
|
|
||||||
from app.views.files import _compute_step_summary
|
from app.views.files import _compute_step_summary
|
||||||
|
|
||||||
# Create mock logs
|
# Create mock logs
|
||||||
class MockLog:
|
class MockLog:
|
||||||
def __init__(self, step_name, status):
|
def __init__(self, step_name, status, timestamp):
|
||||||
self.step_name = step_name
|
self.step_name = step_name
|
||||||
self.status = status
|
self.status = status
|
||||||
|
self.timestamp = timestamp
|
||||||
|
|
||||||
|
now = datetime.now()
|
||||||
logs = [
|
logs = [
|
||||||
MockLog("hash_file", "success"),
|
MockLog("hash_file", "success", now - timedelta(minutes=5)),
|
||||||
MockLog("create_file_record", "success"),
|
MockLog("create_file_record", "success", now - timedelta(minutes=4)),
|
||||||
MockLog("extract_metadata_with_gpt", "failure"),
|
MockLog("extract_metadata_with_gpt", "failure", now - timedelta(minutes=3)),
|
||||||
MockLog("upload_to_dropbox", "success"),
|
MockLog("upload_to_dropbox", "success", now - timedelta(minutes=2)),
|
||||||
MockLog("upload_to_s3", "failure"),
|
MockLog("upload_to_s3", "failure", now - timedelta(minutes=1)),
|
||||||
MockLog("upload_to_nextcloud", "in_progress"),
|
MockLog("upload_to_nextcloud", "in_progress", now),
|
||||||
]
|
]
|
||||||
|
|
||||||
summary = _compute_step_summary(logs)
|
summary = _compute_step_summary(logs)
|
||||||
|
|||||||
@@ -7,9 +7,10 @@ This test module verifies that:
|
|||||||
3. Files with completed steps show "completed" not "processing"
|
3. Files with completed steps show "completed" not "processing"
|
||||||
"""
|
"""
|
||||||
|
|
||||||
import pytest
|
|
||||||
from datetime import datetime, timedelta
|
from datetime import datetime, timedelta
|
||||||
|
|
||||||
|
import pytest
|
||||||
|
|
||||||
from app.utils.file_status import _compute_status_from_logs
|
from app.utils.file_status import _compute_status_from_logs
|
||||||
from app.views.files import _compute_step_summary
|
from app.views.files import _compute_step_summary
|
||||||
|
|
||||||
@@ -249,3 +250,82 @@ class TestMetricsCountingBugFixes:
|
|||||||
assert summary["uploads"]["success"] == 4 # 1 upload + 3 queue
|
assert summary["uploads"]["success"] == 4 # 1 upload + 3 queue
|
||||||
assert summary["uploads"]["failure"] == 1 # 1 upload
|
assert summary["uploads"]["failure"] == 1 # 1 upload
|
||||||
assert summary["uploads"]["in_progress"] == 1 # 1 upload
|
assert summary["uploads"]["in_progress"] == 1 # 1 upload
|
||||||
|
|
||||||
|
def test_order_independent_ascending(self):
|
||||||
|
"""
|
||||||
|
Test that _compute_step_summary works correctly with ascending order logs
|
||||||
|
(as used in production by file_detail_page).
|
||||||
|
|
||||||
|
This test ensures the function correctly selects the latest status per step
|
||||||
|
based on timestamp, not position in the list.
|
||||||
|
"""
|
||||||
|
|
||||||
|
class MockLog:
|
||||||
|
def __init__(self, step_name, status, timestamp):
|
||||||
|
self.step_name = step_name
|
||||||
|
self.status = status
|
||||||
|
self.timestamp = timestamp
|
||||||
|
|
||||||
|
now = datetime.now()
|
||||||
|
# Logs ordered by timestamp ASCENDING (oldest first) - like production
|
||||||
|
logs = [
|
||||||
|
# Older in_progress logs (should be ignored)
|
||||||
|
MockLog("hash_file", "in_progress", now - timedelta(minutes=7)),
|
||||||
|
MockLog("create_file_record", "in_progress", now - timedelta(minutes=6)),
|
||||||
|
MockLog("extract_metadata_with_gpt", "in_progress", now - timedelta(minutes=5)),
|
||||||
|
MockLog("upload_to_dropbox", "in_progress", now - timedelta(minutes=4)),
|
||||||
|
# Latest status for each step (all success) - at the end
|
||||||
|
MockLog("hash_file", "success", now - timedelta(minutes=3)),
|
||||||
|
MockLog("create_file_record", "success", now - timedelta(minutes=2)),
|
||||||
|
MockLog("extract_metadata_with_gpt", "success", now - timedelta(minutes=1)),
|
||||||
|
MockLog("upload_to_dropbox", "success", now),
|
||||||
|
]
|
||||||
|
|
||||||
|
summary = _compute_step_summary(logs)
|
||||||
|
|
||||||
|
# Should use latest status (success) not first seen (in_progress)
|
||||||
|
assert summary["total_main_steps"] == 3
|
||||||
|
assert summary["main"]["success"] == 3
|
||||||
|
assert summary["main"]["in_progress"] == 0
|
||||||
|
|
||||||
|
assert summary["total_upload_tasks"] == 1
|
||||||
|
assert summary["uploads"]["success"] == 1
|
||||||
|
assert summary["uploads"]["in_progress"] == 0
|
||||||
|
|
||||||
|
def test_order_independent_mixed(self):
|
||||||
|
"""
|
||||||
|
Test that _compute_step_summary works correctly with randomly ordered logs.
|
||||||
|
|
||||||
|
This ensures the function truly is order-independent.
|
||||||
|
"""
|
||||||
|
|
||||||
|
class MockLog:
|
||||||
|
def __init__(self, step_name, status, timestamp):
|
||||||
|
self.step_name = step_name
|
||||||
|
self.status = status
|
||||||
|
self.timestamp = timestamp
|
||||||
|
|
||||||
|
now = datetime.now()
|
||||||
|
# Logs in mixed order
|
||||||
|
logs = [
|
||||||
|
MockLog("upload_to_s3", "in_progress", now - timedelta(minutes=8)),
|
||||||
|
MockLog("hash_file", "success", now - timedelta(minutes=1)), # Latest for hash_file
|
||||||
|
MockLog("upload_to_dropbox", "in_progress", now - timedelta(minutes=7)),
|
||||||
|
MockLog("hash_file", "in_progress", now - timedelta(minutes=5)), # Older, should be ignored
|
||||||
|
MockLog("upload_to_s3", "failure", now - timedelta(minutes=2)), # Latest for S3
|
||||||
|
MockLog("create_file_record", "in_progress", now - timedelta(minutes=6)),
|
||||||
|
MockLog("upload_to_dropbox", "success", now), # Latest for Dropbox
|
||||||
|
MockLog("create_file_record", "success", now - timedelta(minutes=3)), # Latest for create
|
||||||
|
]
|
||||||
|
|
||||||
|
summary = _compute_step_summary(logs)
|
||||||
|
|
||||||
|
# Should correctly identify latest status for each step
|
||||||
|
assert summary["total_main_steps"] == 2
|
||||||
|
assert summary["main"]["success"] == 2 # hash_file and create_file_record
|
||||||
|
assert summary["main"]["in_progress"] == 0
|
||||||
|
|
||||||
|
assert summary["total_upload_tasks"] == 2
|
||||||
|
assert summary["uploads"]["success"] == 1 # dropbox
|
||||||
|
assert summary["uploads"]["failure"] == 1 # s3
|
||||||
|
assert summary["uploads"]["in_progress"] == 0
|
||||||
|
|||||||
@@ -0,0 +1,331 @@
|
|||||||
|
"""
|
||||||
|
Tests for the FileProcessingStep model and step_manager utilities.
|
||||||
|
|
||||||
|
This test module verifies the new explicit status tracking approach.
|
||||||
|
"""
|
||||||
|
|
||||||
|
from datetime import datetime, timedelta
|
||||||
|
|
||||||
|
import pytest
|
||||||
|
from sqlalchemy import create_engine
|
||||||
|
from sqlalchemy.orm import Session, sessionmaker
|
||||||
|
|
||||||
|
from app.database import Base
|
||||||
|
from app.models import FileProcessingStep, FileRecord
|
||||||
|
from app.utils.step_manager import (
|
||||||
|
MAIN_PROCESSING_STEPS,
|
||||||
|
add_upload_steps,
|
||||||
|
get_file_overall_status,
|
||||||
|
get_file_step_status,
|
||||||
|
get_step_summary,
|
||||||
|
initialize_file_steps,
|
||||||
|
update_step_status,
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.fixture
|
||||||
|
def db_session():
|
||||||
|
"""Create an in-memory SQLite database for testing."""
|
||||||
|
engine = create_engine("sqlite:///:memory:")
|
||||||
|
Base.metadata.create_all(engine)
|
||||||
|
SessionLocal = sessionmaker(bind=engine)
|
||||||
|
session = SessionLocal()
|
||||||
|
|
||||||
|
yield session
|
||||||
|
|
||||||
|
session.close()
|
||||||
|
Base.metadata.drop_all(engine)
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.mark.unit
|
||||||
|
class TestFileProcessingStepModel:
|
||||||
|
"""Test the FileProcessingStep model."""
|
||||||
|
|
||||||
|
def test_create_step(self, db_session: Session):
|
||||||
|
"""Test creating a processing step."""
|
||||||
|
# Create a file record first
|
||||||
|
file_record = FileRecord(
|
||||||
|
filehash="test123", original_filename="test.pdf", local_filename="/tmp/test.pdf", file_size=1024
|
||||||
|
)
|
||||||
|
db_session.add(file_record)
|
||||||
|
db_session.commit()
|
||||||
|
|
||||||
|
# Create a processing step
|
||||||
|
step = FileProcessingStep(file_id=file_record.id, step_name="hash_file", status="success")
|
||||||
|
db_session.add(step)
|
||||||
|
db_session.commit()
|
||||||
|
|
||||||
|
assert step.id is not None
|
||||||
|
assert step.file_id == file_record.id
|
||||||
|
assert step.step_name == "hash_file"
|
||||||
|
assert step.status == "success"
|
||||||
|
|
||||||
|
def test_unique_constraint(self, db_session: Session):
|
||||||
|
"""Test that the unique constraint on (file_id, step_name) works."""
|
||||||
|
# Create a file record
|
||||||
|
file_record = FileRecord(
|
||||||
|
filehash="test456", original_filename="test2.pdf", local_filename="/tmp/test2.pdf", file_size=2048
|
||||||
|
)
|
||||||
|
db_session.add(file_record)
|
||||||
|
db_session.commit()
|
||||||
|
|
||||||
|
# Create first step
|
||||||
|
step1 = FileProcessingStep(file_id=file_record.id, step_name="hash_file", status="in_progress")
|
||||||
|
db_session.add(step1)
|
||||||
|
db_session.commit()
|
||||||
|
|
||||||
|
# Try to create duplicate step - should fail
|
||||||
|
step2 = FileProcessingStep(file_id=file_record.id, step_name="hash_file", status="success")
|
||||||
|
db_session.add(step2)
|
||||||
|
|
||||||
|
with pytest.raises(Exception): # SQLAlchemy will raise an integrity error
|
||||||
|
db_session.commit()
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.mark.unit
|
||||||
|
class TestStepManager:
|
||||||
|
"""Test step_manager utility functions."""
|
||||||
|
|
||||||
|
def test_initialize_file_steps(self, db_session: Session):
|
||||||
|
"""Test initializing processing steps for a file."""
|
||||||
|
# Create a file record
|
||||||
|
file_record = FileRecord(
|
||||||
|
filehash="test789", original_filename="test3.pdf", local_filename="/tmp/test3.pdf", file_size=4096
|
||||||
|
)
|
||||||
|
db_session.add(file_record)
|
||||||
|
db_session.commit()
|
||||||
|
|
||||||
|
# Initialize steps
|
||||||
|
initialize_file_steps(db_session, file_record.id)
|
||||||
|
|
||||||
|
# Verify steps were created
|
||||||
|
steps = db_session.query(FileProcessingStep).filter(FileProcessingStep.file_id == file_record.id).all()
|
||||||
|
|
||||||
|
assert len(steps) == len(MAIN_PROCESSING_STEPS)
|
||||||
|
for step in steps:
|
||||||
|
assert step.status == "pending"
|
||||||
|
assert step.step_name in MAIN_PROCESSING_STEPS
|
||||||
|
|
||||||
|
def test_add_upload_steps(self, db_session: Session):
|
||||||
|
"""Test adding upload destination steps."""
|
||||||
|
# Create a file record
|
||||||
|
file_record = FileRecord(
|
||||||
|
filehash="test101112", original_filename="test4.pdf", local_filename="/tmp/test4.pdf", file_size=8192
|
||||||
|
)
|
||||||
|
db_session.add(file_record)
|
||||||
|
db_session.commit()
|
||||||
|
|
||||||
|
# Add upload steps
|
||||||
|
destinations = ["dropbox", "s3", "nextcloud"]
|
||||||
|
add_upload_steps(db_session, file_record.id, destinations)
|
||||||
|
|
||||||
|
# Verify upload steps were created
|
||||||
|
steps = db_session.query(FileProcessingStep).filter(FileProcessingStep.file_id == file_record.id).all()
|
||||||
|
|
||||||
|
# Should have 2 steps per destination (queue_ and upload_to_)
|
||||||
|
assert len(steps) == len(destinations) * 2
|
||||||
|
|
||||||
|
expected_steps = []
|
||||||
|
for dest in destinations:
|
||||||
|
expected_steps.append(f"queue_{dest}")
|
||||||
|
expected_steps.append(f"upload_to_{dest}")
|
||||||
|
|
||||||
|
for step in steps:
|
||||||
|
assert step.step_name in expected_steps
|
||||||
|
assert step.status == "pending"
|
||||||
|
|
||||||
|
def test_update_step_status_new_step(self, db_session: Session):
|
||||||
|
"""Test updating status creates step if it doesn't exist."""
|
||||||
|
# Create a file record
|
||||||
|
file_record = FileRecord(
|
||||||
|
filehash="test131415", original_filename="test5.pdf", local_filename="/tmp/test5.pdf", file_size=16384
|
||||||
|
)
|
||||||
|
db_session.add(file_record)
|
||||||
|
db_session.commit()
|
||||||
|
|
||||||
|
# Update a step that doesn't exist yet
|
||||||
|
now = datetime.now()
|
||||||
|
update_step_status(db_session, file_record.id, "hash_file", "in_progress", started_at=now)
|
||||||
|
|
||||||
|
# Verify step was created
|
||||||
|
step = (
|
||||||
|
db_session.query(FileProcessingStep)
|
||||||
|
.filter(FileProcessingStep.file_id == file_record.id, FileProcessingStep.step_name == "hash_file")
|
||||||
|
.first()
|
||||||
|
)
|
||||||
|
|
||||||
|
assert step is not None
|
||||||
|
assert step.status == "in_progress"
|
||||||
|
assert step.started_at == now
|
||||||
|
|
||||||
|
def test_update_step_status_existing_step(self, db_session: Session):
|
||||||
|
"""Test updating status of existing step."""
|
||||||
|
# Create a file record and initialize steps
|
||||||
|
file_record = FileRecord(
|
||||||
|
filehash="test161718", original_filename="test6.pdf", local_filename="/tmp/test6.pdf", file_size=32768
|
||||||
|
)
|
||||||
|
db_session.add(file_record)
|
||||||
|
db_session.commit()
|
||||||
|
|
||||||
|
initialize_file_steps(db_session, file_record.id)
|
||||||
|
|
||||||
|
# Update an existing step
|
||||||
|
now = datetime.now()
|
||||||
|
update_step_status(
|
||||||
|
db_session, file_record.id, "hash_file", "success", started_at=now - timedelta(seconds=5), completed_at=now
|
||||||
|
)
|
||||||
|
|
||||||
|
# Verify step was updated
|
||||||
|
step = (
|
||||||
|
db_session.query(FileProcessingStep)
|
||||||
|
.filter(FileProcessingStep.file_id == file_record.id, FileProcessingStep.step_name == "hash_file")
|
||||||
|
.first()
|
||||||
|
)
|
||||||
|
|
||||||
|
assert step.status == "success"
|
||||||
|
assert step.started_at == now - timedelta(seconds=5)
|
||||||
|
assert step.completed_at == now
|
||||||
|
|
||||||
|
def test_get_file_step_status(self, db_session: Session):
|
||||||
|
"""Test retrieving all step statuses for a file."""
|
||||||
|
# Create a file record and initialize steps
|
||||||
|
file_record = FileRecord(
|
||||||
|
filehash="test192021", original_filename="test7.pdf", local_filename="/tmp/test7.pdf", file_size=65536
|
||||||
|
)
|
||||||
|
db_session.add(file_record)
|
||||||
|
db_session.commit()
|
||||||
|
|
||||||
|
initialize_file_steps(db_session, file_record.id)
|
||||||
|
|
||||||
|
# Update some steps
|
||||||
|
now = datetime.now()
|
||||||
|
update_step_status(db_session, file_record.id, "hash_file", "success", completed_at=now)
|
||||||
|
update_step_status(db_session, file_record.id, "create_file_record", "in_progress", started_at=now)
|
||||||
|
update_step_status(db_session, file_record.id, "check_text", "failure", error_message="Failed to check text")
|
||||||
|
|
||||||
|
# Get all step statuses
|
||||||
|
status_map = get_file_step_status(db_session, file_record.id)
|
||||||
|
|
||||||
|
assert len(status_map) == len(MAIN_PROCESSING_STEPS)
|
||||||
|
assert status_map["hash_file"]["status"] == "success"
|
||||||
|
assert status_map["hash_file"]["completed_at"] == now
|
||||||
|
assert status_map["create_file_record"]["status"] == "in_progress"
|
||||||
|
assert status_map["check_text"]["status"] == "failure"
|
||||||
|
assert status_map["check_text"]["error_message"] == "Failed to check text"
|
||||||
|
|
||||||
|
def test_get_file_overall_status_pending(self, db_session: Session):
|
||||||
|
"""Test overall status for a file with pending steps."""
|
||||||
|
file_record = FileRecord(
|
||||||
|
filehash="test222324", original_filename="test8.pdf", local_filename="/tmp/test8.pdf", file_size=131072
|
||||||
|
)
|
||||||
|
db_session.add(file_record)
|
||||||
|
db_session.commit()
|
||||||
|
|
||||||
|
initialize_file_steps(db_session, file_record.id)
|
||||||
|
|
||||||
|
status = get_file_overall_status(db_session, file_record.id)
|
||||||
|
|
||||||
|
assert status["status"] == "pending"
|
||||||
|
assert status["has_errors"] is False
|
||||||
|
assert status["total_steps"] == len(MAIN_PROCESSING_STEPS)
|
||||||
|
assert status["completed_steps"] == 0
|
||||||
|
assert status["in_progress_steps"] == 0
|
||||||
|
|
||||||
|
def test_get_file_overall_status_processing(self, db_session: Session):
|
||||||
|
"""Test overall status for a file with in-progress steps."""
|
||||||
|
file_record = FileRecord(
|
||||||
|
filehash="test252627", original_filename="test9.pdf", local_filename="/tmp/test9.pdf", file_size=262144
|
||||||
|
)
|
||||||
|
db_session.add(file_record)
|
||||||
|
db_session.commit()
|
||||||
|
|
||||||
|
initialize_file_steps(db_session, file_record.id)
|
||||||
|
|
||||||
|
# Mark some steps as complete and one as in progress
|
||||||
|
update_step_status(db_session, file_record.id, "hash_file", "success")
|
||||||
|
update_step_status(db_session, file_record.id, "create_file_record", "success")
|
||||||
|
update_step_status(db_session, file_record.id, "check_text", "in_progress")
|
||||||
|
|
||||||
|
status = get_file_overall_status(db_session, file_record.id)
|
||||||
|
|
||||||
|
assert status["status"] == "processing"
|
||||||
|
assert status["has_errors"] is False
|
||||||
|
assert status["completed_steps"] == 2
|
||||||
|
assert status["in_progress_steps"] == 1
|
||||||
|
|
||||||
|
def test_get_file_overall_status_completed(self, db_session: Session):
|
||||||
|
"""Test overall status for a completed file."""
|
||||||
|
file_record = FileRecord(
|
||||||
|
filehash="test282930", original_filename="test10.pdf", local_filename="/tmp/test10.pdf", file_size=524288
|
||||||
|
)
|
||||||
|
db_session.add(file_record)
|
||||||
|
db_session.commit()
|
||||||
|
|
||||||
|
initialize_file_steps(db_session, file_record.id)
|
||||||
|
|
||||||
|
# Mark all steps as success
|
||||||
|
for step_name in MAIN_PROCESSING_STEPS:
|
||||||
|
update_step_status(db_session, file_record.id, step_name, "success")
|
||||||
|
|
||||||
|
status = get_file_overall_status(db_session, file_record.id)
|
||||||
|
|
||||||
|
assert status["status"] == "completed"
|
||||||
|
assert status["has_errors"] is False
|
||||||
|
assert status["completed_steps"] == len(MAIN_PROCESSING_STEPS)
|
||||||
|
assert status["in_progress_steps"] == 0
|
||||||
|
|
||||||
|
def test_get_file_overall_status_failed(self, db_session: Session):
|
||||||
|
"""Test overall status for a file with failed steps."""
|
||||||
|
file_record = FileRecord(
|
||||||
|
filehash="test313233", original_filename="test11.pdf", local_filename="/tmp/test11.pdf", file_size=1048576
|
||||||
|
)
|
||||||
|
db_session.add(file_record)
|
||||||
|
db_session.commit()
|
||||||
|
|
||||||
|
initialize_file_steps(db_session, file_record.id)
|
||||||
|
|
||||||
|
# Mark some steps as success and one as failure
|
||||||
|
update_step_status(db_session, file_record.id, "hash_file", "success")
|
||||||
|
update_step_status(db_session, file_record.id, "create_file_record", "success")
|
||||||
|
update_step_status(db_session, file_record.id, "check_text", "failure", error_message="OCR failed")
|
||||||
|
|
||||||
|
status = get_file_overall_status(db_session, file_record.id)
|
||||||
|
|
||||||
|
assert status["status"] == "failed"
|
||||||
|
assert status["has_errors"] is True
|
||||||
|
assert status["failed_steps"] == 1
|
||||||
|
|
||||||
|
def test_get_step_summary(self, db_session: Session):
|
||||||
|
"""Test getting step summary with counts."""
|
||||||
|
file_record = FileRecord(
|
||||||
|
filehash="test343536", original_filename="test12.pdf", local_filename="/tmp/test12.pdf", file_size=2097152
|
||||||
|
)
|
||||||
|
db_session.add(file_record)
|
||||||
|
db_session.commit()
|
||||||
|
|
||||||
|
# Initialize main steps and add upload steps
|
||||||
|
initialize_file_steps(db_session, file_record.id)
|
||||||
|
add_upload_steps(db_session, file_record.id, ["dropbox", "s3", "nextcloud"])
|
||||||
|
|
||||||
|
# Update statuses
|
||||||
|
update_step_status(db_session, file_record.id, "hash_file", "success")
|
||||||
|
update_step_status(db_session, file_record.id, "create_file_record", "success")
|
||||||
|
update_step_status(db_session, file_record.id, "check_text", "in_progress")
|
||||||
|
update_step_status(db_session, file_record.id, "upload_to_dropbox", "success")
|
||||||
|
update_step_status(db_session, file_record.id, "upload_to_s3", "failure")
|
||||||
|
update_step_status(db_session, file_record.id, "queue_nextcloud", "in_progress")
|
||||||
|
|
||||||
|
summary = get_step_summary(db_session, file_record.id)
|
||||||
|
|
||||||
|
# Check main step counts
|
||||||
|
assert summary["main"]["success"] == 2
|
||||||
|
assert summary["main"]["in_progress"] == 1
|
||||||
|
assert summary["main"]["queued"] >= 5 # Remaining pending steps
|
||||||
|
|
||||||
|
# Check upload counts
|
||||||
|
assert summary["uploads"]["success"] == 1
|
||||||
|
assert summary["uploads"]["failure"] == 1
|
||||||
|
assert summary["uploads"]["in_progress"] == 1
|
||||||
|
|
||||||
|
assert summary["total_main_steps"] == len(MAIN_PROCESSING_STEPS)
|
||||||
|
assert summary["total_upload_tasks"] == 6 # 3 destinations x 2 steps each
|
||||||
Reference in New Issue
Block a user