manage/backend/src/media_library_viewer_api/clients/qbittorrent.py

"""Minimal qBittorrent Web API client (read-only: sync/maindata only).

Modeled on :class:`~media_library_viewer_api.clients.jellyfin.JellyfinClient`'s
session pattern. Authentication uses username/password login which stores an
SID cookie in the requests session. The client re-logins transparently on 403.
"""

from __future__ import annotations

import logging
from typing import Any

import requests

from media_library_viewer_api.clients.http_timeout import DEFAULT_READ_TIMEOUT, http_timeout

logger = logging.getLogger(__name__)


class QbittorrentClient:
    """Small wrapper around the qBittorrent Web API.

    Only the endpoints needed by the dashboard widgets are implemented
    (currently just ``/sync/maindata``). All calls share a single
    :class:`requests.Session` that carries the login cookie.
    """

    def __init__(self, base_url: str, username: str, password: str, timeout: float = DEFAULT_READ_TIMEOUT) -> None:
        if not base_url:
            raise ValueError("qBittorrent base_url is required")
        if not username:
            raise ValueError("qBittorrent username is required")

        self.base_url = base_url.rstrip("/")
        if not self.base_url.endswith("/api/v2"):
            self.base_url += "/api/v2"
        self._username = username
        self._password = password
        # timeout is the per-response READ timeout (seconds); connect timeout is fixed at 5s.
        self.timeout = http_timeout(timeout)
        self._session = requests.Session()
        self._logged_in = False

    def _login(self) -> None:
        """POST username/password to ``/auth/login``; store the SID cookie.

        qBittorrent replies with the plain text ``"Ok."`` and a ``SID`` cookie
        on success, ``"Fails."`` on bad credentials, and ``403 Forbidden`` when
        the source IP is banned (too many failed attempts). The ``Referer``
        header is required by qBittorrent's CSRF protection.

        Any other body — in particular an *empty* 200 — means the request did not
        reach qBittorrent's login handler, almost always because ``base_url`` is
        wrong (wrong host/port/path) or a reverse proxy is misrouting
        ``/api/v2/auth/login``. We surface a diagnostic error in that case
        instead of the useless ``"login failed: "`` message.
        """
        resp = self._session.post(
            f"{self.base_url}/auth/login",
            data={"username": self._username, "password": self._password},
            timeout=self.timeout,
            headers={"Referer": self.base_url},
        )
        # 502/503/504 come from the reverse proxy when qBittorrent is down,
        # starting up, or can't answer within the proxy's forwarding timeout
        # (qBittorrent's PBKDF2 password check is intentionally slow, so a
        # flood of concurrent logins can trip this). Surface it clearly rather
        # than as a bare HTTPError.
        if resp.status_code in (502, 503, 504):
            raise RuntimeError(
                f"qBittorrent is unreachable: reverse proxy returned HTTP {resp.status_code} "
                f"for {resp.url}. qBittorrent may be down, starting up, or unable to "
                "answer within the proxy's forwarding timeout."
            )
        resp.raise_for_status()
        body = resp.text.strip()
        # Some reverse proxies forward the SID cookie but mangle the text body;
        # accept either success signal. Guard the cookie read behind an empty
        # body so a mocked response never accidentally reads as success.
        sid_ok = body == "" and bool(resp.cookies.get("SID"))
        if body == "Ok." or sid_ok:
            self._logged_in = True
            logger.info("qBittorrent login successful for %s", self.base_url)
            return
        if body == "Fails.":
            raise RuntimeError(f"qBittorrent login failed (HTTP {resp.status_code}): invalid username or password")
        raise RuntimeError(
            f"Unexpected response from qBittorrent login endpoint (HTTP {resp.status_code}, "
            f"body={body!r}). Expected the text 'Ok.' from /api/v2/auth/login — this usually "
            "means base_url does not reach the qBittorrent Web API (check the URL, path, "
            "and any reverse proxy in front of qBittorrent)."
        )

    def _get(self, path: str, **params: Any) -> dict[str, Any]:
        """GET an endpoint with auto-login on first call and re-login on 403."""
        if not self._logged_in:
            self._login()
        url = f"{self.base_url}{path}"
        resp = self._session.get(url, params=params, timeout=self.timeout)
        if resp.status_code == 403:
            logger.debug("qBittorrent 403 on %s, re-logging in", path)
            self._logged_in = False
            self._login()
            resp = self._session.get(url, params=params, timeout=self.timeout)
        resp.raise_for_status()
        return resp.json()

    def maindata(self) -> dict[str, Any]:
        """Fetch ``/sync/maindata``.

        Returns a dict with ``server_state`` (containing ``dl_info_speed``,
        ``up_info_speed``, etc.) and ``torrents`` (a dict of
        ``{hash: {name, state, progress, size, dlspeed, upspeed, ...}}``).
        """
        return self._get("/sync/maindata")