from typing import Any, Dict, Optional import httpx import logging import time from .base import ApiClient, _operation_error_message from ..services.operation_progress import finish_remote_call, start_remote_call def _torrent_state_text(state: Any) -> str: normalized = str(state or "").strip().lower() if normalized in {"uploading", "stalledup", "forcedup", "queuedup", "pausedup", "stoppedup", "completed"}: return "finished" if "pause" in normalized or normalized == "stoppeddl": return "paused" if "stall" in normalized: return "waiting for data" if normalized.startswith("queued"): return "waiting in the queue" if normalized == "metadl": return "getting the download details" if normalized in {"checkingdl", "checkingup", "checkingresumedata"}: return "checking the downloaded files" if "downloading" in normalized or normalized in {"forcedl", "forceddl"}: return "downloading" if "upload" in normalized: return "downloaded and sharing with others" if normalized in {"completed", "missingfiles"}: return "finished" if normalized == "completed" else "missing files" if "error" in normalized: return "unable to continue" return "present" def _torrent_result_message(result: Any) -> str: torrents = result if isinstance(result, list) else [] if not torrents: return "qBittorrent found no matching downloads." first = next((item for item in torrents if isinstance(item, dict)), {}) if len(torrents) == 1: progress = first.get("progress") progress_text = ( f" — {max(0, min(100, round(progress * 100)))}% complete" if isinstance(progress, (int, float)) else "" ) state_text = _torrent_state_text(first.get("state")) return f'{"Downloading" if state_text == "downloading" else "The download is " + state_text}{progress_text}.' active = sum( 1 for item in torrents if isinstance(item, dict) and _torrent_state_text(item.get("state")) == "downloading" ) return f"qBittorrent found {len(torrents)} matching downloads; {active} are actively downloading." def _torrent_action_message(path: str) -> str: normalized_path = path.lower() if normalized_path.endswith("/resume") or normalized_path.endswith("/start"): return "qBittorrent accepted the request to resume the download." if normalized_path.endswith("/add"): return "qBittorrent accepted the release and added it to the download queue." return "qBittorrent accepted the requested download action." class QBittorrentClient(ApiClient): def __init__(self, base_url: Optional[str], username: Optional[str], password: Optional[str]): super().__init__(base_url, None) self.username = username self.password = password self.logger = logging.getLogger(__name__) def configured(self) -> bool: return bool(self.base_url and self.username and self.password) async def _login(self, client: httpx.AsyncClient) -> None: if not self.base_url or not self.username or not self.password: raise RuntimeError("qBittorrent not configured") response = await client.post( f"{self.base_url}/api/v2/auth/login", data={"username": self.username, "password": self.password}, headers={"Referer": self.base_url}, ) response.raise_for_status() text = response.text.strip().lower() has_session_cookie = any(name.upper().startswith("QBT_SID") for name in client.cookies.keys()) if text not in {"ok.", ""} or (text == "" and not has_session_cookie): raise RuntimeError("qBittorrent login failed") async def _get(self, path: str, params: Optional[Dict[str, Any]] = None) -> Optional[Any]: if not self.base_url: return None started_at = time.perf_counter() operation_event_id = start_remote_call("qBittorrent", "Checking qBittorrent for matching downloads…") try: async with httpx.AsyncClient(timeout=10.0) as client: await self._login(client) response = await client.get(f"{self.base_url}{path}", params=params) response.raise_for_status() result = response.json() duration_ms = round((time.perf_counter() - started_at) * 1000, 2) finish_remote_call( operation_event_id, success=True, status_code=response.status_code, message=_torrent_result_message(result), ) return result except Exception as exc: duration_ms = round((time.perf_counter() - started_at) * 1000, 2) status_code = exc.response.status_code if isinstance(exc, httpx.HTTPStatusError) else None finish_remote_call( operation_event_id, success=False, status_code=status_code, message=_operation_error_message("qBittorrent", status_code), ) raise async def _get_text(self, path: str, params: Optional[Dict[str, Any]] = None) -> Optional[str]: if not self.base_url: return None started_at = time.perf_counter() operation_event_id = start_remote_call("qBittorrent") try: async with httpx.AsyncClient(timeout=10.0) as client: await self._login(client) response = await client.get(f"{self.base_url}{path}", params=params) response.raise_for_status() result = response.text.strip() duration_ms = round((time.perf_counter() - started_at) * 1000, 2) finish_remote_call( operation_event_id, success=True, status_code=response.status_code, message=f"Connected to qBittorrent{f' version {result}' if result else ''}.", ) return result except Exception as exc: duration_ms = round((time.perf_counter() - started_at) * 1000, 2) status_code = exc.response.status_code if isinstance(exc, httpx.HTTPStatusError) else None finish_remote_call( operation_event_id, success=False, status_code=status_code, message=_operation_error_message("qBittorrent", status_code), ) raise async def _post_form(self, path: str, data: Dict[str, Any]) -> None: if not self.base_url: return None started_at = time.perf_counter() operation_event_id = start_remote_call("qBittorrent") try: async with httpx.AsyncClient(timeout=10.0) as client: await self._login(client) response = await client.post(f"{self.base_url}{path}", data=data) response.raise_for_status() duration_ms = round((time.perf_counter() - started_at) * 1000, 2) finish_remote_call( operation_event_id, success=True, status_code=response.status_code, message=_torrent_action_message(path), ) except Exception as exc: duration_ms = round((time.perf_counter() - started_at) * 1000, 2) status_code = exc.response.status_code if isinstance(exc, httpx.HTTPStatusError) else None finish_remote_call( operation_event_id, success=False, status_code=status_code, message=_operation_error_message("qBittorrent", status_code), ) raise async def is_webui_reachable(self) -> bool: if not self.base_url: return False try: async with httpx.AsyncClient(timeout=10.0, follow_redirects=True) as client: response = await client.get(self.base_url) response.raise_for_status() return True except httpx.HTTPError: return False async def get_torrents(self) -> Optional[Any]: return await self._get("/api/v2/torrents/info") async def get_torrents_by_hashes(self, hashes: str) -> Optional[Any]: return await self._get("/api/v2/torrents/info", params={"hashes": hashes}) async def get_torrents_by_category(self, category: str) -> Optional[Any]: return await self._get("/api/v2/torrents/info", params={"category": category}) async def get_torrents_by_tag(self, tag: str) -> Optional[Any]: return await self._get("/api/v2/torrents/info", params={"tag": tag}) async def get_app_version(self) -> Optional[Any]: return await self._get_text("/api/v2/app/version") async def resume_torrents(self, hashes: str) -> None: try: await self._post_form("/api/v2/torrents/resume", data={"hashes": hashes}) except httpx.HTTPStatusError as exc: if exc.response is not None and exc.response.status_code == 404: await self._post_form("/api/v2/torrents/start", data={"hashes": hashes}) return raise async def add_torrent_url( self, url: str, category: Optional[str] = None, tags: Optional[str] = None ) -> None: url_host = None if isinstance(url, str) and "://" in url: url_host = url.split("://", 1)[-1].split("/", 1)[0] self.logger.warning( "qBittorrent add_torrent_url invoked: category=%s host=%s", category, url_host or "unknown", ) data: Dict[str, Any] = {"urls": url} if category: data["category"] = category if tags: data["tags"] = tags await self._post_form("/api/v2/torrents/add", data=data)