QBittorrent WebAPI script for getting the size of all torrents.

README.md

Requires Python 3.6+

Only 3rd party dependancy is the Requests library. Install it with: python3 -m pip install requests

torrentsize.py

import enum
import getpass
import urllib3
import typing as t

import requests

# disregard HTTPS verification warnings due to common selfhost config issues
urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)


class QBitAPIError(Exception):
    """Exception for errors encountered with the QBittorrent API."""
    pass


class AuthError(QBitAPIError):
    """Exception for issues when authenticating with Qbittorrent API."""
    pass


class Filters(enum.Enum):
    """The available filters for torrents in the client."""

    all = "all"
    downloading = "downloading"
    completed = "completed"
    paused = "paused"
    active = "active"
    inactive = "inactive"
    resumed = "resumed"


class Units(enum.IntEnum):
    """The different units data can be represented."""

    auto = 0
    B = 1
    KiB = 1024
    MiB = KiB * 1024
    GiB = MiB * 1024
    TiB = GiB * 1024
    PiB = TiB * 1024

    @staticmethod
    def humanize(size_bytes: float) -> t.Tuple[float, "Units"]:
        """Convert the given value of bytes into the maximum reasonable unit."""
        for unit in Units:
            if unit.value == 0:
                continue
            new_size = size_bytes/unit
            if abs(new_size) < 1024:
                return round(new_size, 2), unit

        return round(new_size, 2), unit


class QBitAPI:
    """Manage interacting with the QBittorrent API."""

    def __init__(
        self,
        username: t.Optional[str] = None,
        password: t.Optional[str] = None,
        token: t.Optional[str] = None,
        hosted_url: str = "http://localhost:8080"
    ):
        self.username = username
        self.password = password
        self.cookies = {"SID": token} if token else None

        # remove trailing slashes to keep clean for url building
        hosted_url = hosted_url.rstrip("/")
        url = urllib3.util.parse_url(hosted_url)
        port = f":{url.port}" if url.port else ""

        self.host = f"{url.scheme}://{url.host}{port}"
        self.api_url = f"{hosted_url}/api/v2"

        print(f"Connecting to: {self.host}")

        # cached property values
        self._version = None
        self._webapi_version = None
        self._torrents = {}

        if not self.version:
            self._version = None
            print("Your cookie is invalid or expired")
            self.authenticate()

        print(f"QBit {self.version}")
        print(f"QBit WebAPI {self.webapi_version}")

        if float(self.webapi_version) < 2:
            raise QBitAPIError(
                "Your WebAPI version is too old: "
                f"Need v2.x and above but you have v{self.webapi_version}"
            )

    def get(self, endpoint: str, params: t.Dict[str, str] = None) -> requests.Request:
        """Send a request to the given API endpoint, using any existing authentication."""
        if not self.cookies:
            self.authenticate()
        return requests.get(
            f"{self.api_url}/{endpoint}",
            cookies=self.cookies,
            verify=False,
            params=params
        )

    def authenticate(self) -> t.Dict[str, str]:
        """Use provided login credentials to request a new authentication cookie."""
        if not self.username and not self.password:
            self.username = input("Please enter your username: ")
            self.password = getpass.getpass("Please enter your password: ")

        url = f"{self.api_url}/auth/login"
        headers = {"Referer": self.host}
        params = {"username": self.username, "password": self.password}
        r = requests.get(url, headers=headers, verify=False, params=params)

        if r.status_code == 403:
            raise AuthError(
                "You're IP banned due to too many auth failures. "
                "Wait an hour or restart qbittorrent."
            )

        if r.status_code != 200:
            raise AuthError("Something went wrong, and I'm not quite sure what.")

        if "SID" not in r.cookies:
            raise AuthError(
                "Your login credentials were rejected. "
                "Please check your details and try again."
            )

        token = r.cookies["SID"]
        self.cookies = {"SID": token}
        print(f"Your new auth token is: {token}")
        return token

    @property
    def version(self) -> str:
        """Get the QBittorrent client version number."""
        if not self._version:
            self._version = self.get("app/version").text
        return self._version

    @property
    def webapi_version(self) -> str:
        """Get the QBittorrent WebAPI version number."""
        if not self._webapi_version:
            self._webapi_version = self.get("app/webapiVersion").text
        return self._webapi_version

    def torrents(self) -> t.Dict[str, t.Union[int, bool, str, float]]:
        """Get the data of all torrents in the client."""
        r = self.get("torrents/info")
        if not self._torrents:
            self._torrents = r.json()
        return self._torrents

    def total_size(
        self,
        filter_: Filters = Filters.all,
        category: t.Optional[str] = None,
        unit: Units = Units.auto,
        as_str: bool = True
    ):
        """Retrieve the collective size of torrents that match the given filter and category."""
        torrents = self.torrents()
        total_bytes = sum(t["size"] for t in torrents)
        if unit == Units.auto:
            size = Units.humanize(total_bytes)
        else:
            size = (total_bytes / unit, unit)
        if as_str:
            size = f"{size[0]:.2f}{size[1].name}"
        return size

    def torrent_count(self):
        """Get the total count of all torrents."""
        return len(self.torrents())


qapi = QBitAPI(token="1cGoHRlLntZH7O4QGqne1OryPk5Ppp/C")

print(f"Count: {qapi.torrent_count()}")
print(f"Total Size: {qapi.total_size()}")