INNER CODE UNIT · Python
emit_queue_stats
project-sekai-ctf/sekaictf-2025 · misc/sekai-bank-transaction/challenge/poc-tester/poc-web/src/app.py:36
def emit_queue_stats():
with queue_lock:
pending = len([q for q in queue if q.status == Status.PENDING_QUEUE])
processing = len([q for q in queue if q.status not in [Status.COMPLETED, Status.ERROR, Status.PENDING_QUEUE]])
queue_size = pending + processing
socketio.emit('queue_stats', {'queue_size': queue_size})
def emit_status_update(queue_item, update_stats=True):
status_data = {
'id': queue_item.id,
'status': queue_item.status.value if hasattr(queue_item.status, 'value') else queue_item.status,
'error': queue_item.error,
'created_at': queue_item.created_at.isoformat() if queue_item.created_at else None,
'completed_at': queue_item.completed_at.isoformat() if queue_item.completed_at else None,
'duration': queue_item.duration,
'challenge': queue_item.client.CHALLENGE_NAME
}