copilot-swe-agent[bot]
4fb696e1eb
- Fix _compute_status_from_logs to track latest status per unique step - Fix _compute_step_summary to count only latest status per step - Add comprehensive tests for both fixes - Resolves issue where completed files showed as "Processing" - Resolves issue where metrics showed incorrect counts Co-authored-by: christianlouis <361235+christianlouis@users.noreply.github.com>
106 lines
3.0 KiB
Python
106 lines
3.0 KiB
Python
"""
|
|
Utility functions for file processing status determination.
|
|
"""
|
|
|
|
from typing import Dict, List
|
|
|
|
from sqlalchemy.orm import Session
|
|
|
|
from app.models import ProcessingLog
|
|
|
|
|
|
def get_file_processing_status(db: Session, file_id: int) -> Dict:
|
|
"""
|
|
Get the processing status for a file by checking its processing logs.
|
|
|
|
Args:
|
|
db: Database session
|
|
file_id: ID of the file
|
|
|
|
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()
|
|
)
|
|
|
|
return _compute_status_from_logs(logs)
|
|
|
|
|
|
def get_files_processing_status(db: Session, file_ids: List[int]) -> Dict[int, Dict]:
|
|
"""
|
|
Get processing status for multiple files efficiently.
|
|
|
|
Args:
|
|
db: Database session
|
|
file_ids: List of file IDs
|
|
|
|
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()
|
|
)
|
|
|
|
# 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)
|
|
|
|
# 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)
|
|
|
|
return result
|
|
|
|
|
|
def _compute_status_from_logs(logs: List[ProcessingLog]) -> Dict:
|
|
"""
|
|
Compute processing status from a list of processing logs.
|
|
|
|
Args:
|
|
logs: List of ProcessingLog objects (should be ordered by timestamp desc)
|
|
|
|
Returns:
|
|
dict with status, last_step, has_errors, and total_steps
|
|
"""
|
|
if not logs:
|
|
return {"status": "pending", "last_step": None, "has_errors": False, "total_steps": 0}
|
|
|
|
# Get the latest status for each unique step
|
|
# Since logs are ordered by timestamp desc, the first occurrence is the latest
|
|
latest_by_step = {}
|
|
for log in logs:
|
|
if log.step_name not in latest_by_step:
|
|
latest_by_step[log.step_name] = log
|
|
|
|
# Check for failures in latest statuses
|
|
has_errors = any(log.status == "failure" for log in latest_by_step.values())
|
|
|
|
# Check if any step is currently in progress (based on latest status per step)
|
|
in_progress = any(log.status == "in_progress" for log in latest_by_step.values())
|
|
|
|
# Get the overall latest log
|
|
latest_log = logs[0]
|
|
|
|
# Determine overall status
|
|
if has_errors:
|
|
status = "failed"
|
|
elif in_progress:
|
|
status = "processing"
|
|
elif latest_log.status == "success":
|
|
status = "completed"
|
|
else:
|
|
status = "pending"
|
|
|
|
return {"status": status, "last_step": latest_log.step_name, "has_errors": has_errors, "total_steps": len(logs)}
|