"""JSON-backed queue store with file locking for thread safety.""" import json import os from datetime import datetime, timezone from pathlib import Path from threading import Lock from models import QueueItem class QueueStore: """Persistent queue backed by a JSON file.""" def __init__(self, store_path: str = None): if store_path is None: store_path = str(Path.home() / ".config" / "youtube_cli" / "queue.json") self.store_path = store_path self._lock = Lock() self._ensure_file() def _ensure_file(self): """Create the store file if it doesn't exist.""" store_dir = Path(self.store_path).parent store_dir.mkdir(parents=True, exist_ok=True) if not Path(self.store_path).exists(): with open(self.store_path, "w") as f: json.dump({}, f) def _load(self) -> dict: """Load queue data from file.""" try: with open(self.store_path, "r") as f: return json.load(f) except (json.JSONDecodeError, ValueError): with open(self.store_path, "w") as f: json.dump({}, f) return {} def _save(self, data: dict): """Save queue data to file.""" with open(self.store_path, "w") as f: json.dump(data, f, indent=2) def add_item(self, item: QueueItem) -> QueueItem: """Add an item to the queue.""" with self._lock: data = self._load() data[item.id] = item.to_dict() self._save(data) return item def get_all(self) -> list: """Get all queue items.""" with self._lock: data = self._load() items = [] for item_id, item_data in data.items(): item = self._dict_to_item(item_data) items.append(item) return items def get_item(self, queue_id: str) -> QueueItem: """Get a specific queue item.""" with self._lock: data = self._load() item_data = data.get(queue_id) if item_data: return self._dict_to_item(item_data) return None def update_item(self, queue_id: str, updates: dict) -> QueueItem: """Update fields of a queue item.""" with self._lock: data = self._load() if queue_id not in data: return None data[queue_id].update(updates) self._save(data) item = self._dict_to_item(data[queue_id]) return item def update_progress(self, queue_id: str, progress: float, speed: str = None, eta: str = None): """Update download progress for a queue item.""" with self._lock: data = self._load() if queue_id in data: data[queue_id]["progress"] = progress if speed: data[queue_id]["speed"] = speed if eta: data[queue_id]["eta"] = eta self._save(data) def update_status(self, queue_id: str, status: str, error_message: str = None, download_path: str = None, file_size: str = None): """Update download status for a queue item.""" with self._lock: data = self._load() if queue_id in data: data[queue_id]["status"] = status if status in ("completed", "failed"): data[queue_id]["completedAt"] = datetime.now(timezone.utc).isoformat() if error_message: data[queue_id]["errorMessage"] = error_message if download_path: data[queue_id]["downloadPath"] = download_path if file_size: data[queue_id]["fileSize"] = file_size if status == "completed": data[queue_id]["progress"] = 100.0 self._save(data) def remove_item(self, queue_id: str) -> bool: """Remove an item from the queue.""" with self._lock: data = self._load() if queue_id in data: del data[queue_id] self._save(data) return True return False def clear_completed(self) -> int: """Clear all completed items. Returns count of removed items.""" with self._lock: data = self._load() completed_ids = [qid for qid, item in data.items() if item["status"] == "completed"] for qid in completed_ids: del data[qid] self._save(data) return len(completed_ids) def clear_failed(self) -> int: """Clear all failed items. Returns count of removed items.""" with self._lock: data = self._load() failed_ids = [qid for qid, item in data.items() if item["status"] == "failed"] for qid in failed_ids: del data[qid] self._save(data) return len(failed_ids) def clear_all(self) -> int: """Clear all items from the queue. Returns count of removed items.""" with self._lock: count = len(self._load()) self._save({}) return count def get_stats(self) -> dict: """Get queue statistics.""" items = self.get_all() return { "total": len(items), "pending": sum(1 for i in items if i.status == "pending"), "downloading": sum(1 for i in items if i.status == "downloading"), "completed": sum(1 for i in items if i.status == "completed"), "failed": sum(1 for i in items if i.status == "failed"), "cancelled": sum(1 for i in items if i.status == "cancelled"), } def reorder_item(self, queue_id: str, direction: str) -> bool: """Reorder a queue item (up/down). Returns True if reordered.""" with self._lock: data = self._load() ids = list(data.keys()) if queue_id not in ids: return False idx = ids.index(queue_id) if direction == "up" and idx > 0: ids[idx], ids[idx - 1] = ids[idx - 1], ids[idx] elif direction == "down" and idx < len(ids) - 1: ids[idx], ids[idx + 1] = ids[idx + 1], ids[idx] else: return False # Rebuild dict in new order new_data = {} for kid in ids: new_data[kid] = data[kid] self._save(new_data) return True def _dict_to_item(self, data: dict) -> QueueItem: """Convert a dictionary to a QueueItem.""" return QueueItem( id=data["id"], video_id=data.get("videoId", ""), title=data.get("title", ""), url=data.get("url", ""), thumbnail=data.get("thumbnail", ""), status=data.get("status", "pending"), progress=data.get("progress", 0.0), category=data.get("category", ""), network_folder=data.get("network_folder"), added_at=data.get("addedAt", data.get("created_at", datetime.now(timezone.utc).isoformat())), completed_at=data.get("completedAt"), error_message=data.get("errorMessage", data.get("message")), download_path=data.get("downloadPath"), file_size=data.get("fileSize"), speed=data.get("speed"), eta=data.get("eta"), item_type=data.get("type", "video"), quality=data.get("quality"), )