| 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): |
| |
| 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 |
|
|
| |
| command_queue = CommandQueue() |
|
|
