File size: 2,110 Bytes
ab616bf | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 | import time
import uuid
from typing import List, Dict, Any, Optional
from threading import Lock
from utils.logger import setup_logger
logger = setup_logger("command_queue")
class CommandQueue:
_instance = None
_lock = Lock()
def __new__(cls):
with cls._lock:
if cls._instance is None:
cls._instance = super(CommandQueue, cls).__new__(cls)
cls._instance._init_queue()
return cls._instance
def _init_queue(self):
# session_id -> List[Dict]
self.queues: Dict[str, List[Dict[str, Any]]] = {}
self.lock = Lock()
def push(self, session_id: str, command_type: str, data: Dict[str, Any], priority: str = "medium"):
"""Pushes a new command to the queue for a specific session."""
command = {
"id": str(uuid.uuid4())[:8],
"type": command_type,
"data": data,
"priority": priority,
"timestamp": time.time(),
"status": "pending"
}
with self.lock:
if session_id not in self.queues:
self.queues[session_id] = []
self.queues[session_id].append(command)
target_info = data.get("target") or data.get("text") or data
print(f"[TEST LOG] Command queued: {command_type} ({target_info})")
logger.info(f"Command pushed to queue [{session_id}]: {command_type} (ID: {command['id']})")
def get_pending(self, session_id: str, clear: bool = True) -> List[Dict[str, Any]]:
"""Retrieves and optionally clears pending commands for a session."""
with self.lock:
if session_id not in self.queues or not self.queues[session_id]:
return []
pending = self.queues[session_id]
if clear:
self.queues[session_id] = []
if pending:
print(f"[TEST LOG] {len(pending)} Command(s) popped from queue")
return pending
# Global instance for easy import
command_queue = CommandQueue()
|