from __future__ import annotations

import json
import logging
import pathlib
import string
import time
import types
from typing import Any, BinaryIO, Iterable, List, TypeVar, Union
from urllib.parse import quote

import requests
import requests.auth
import requests.exceptions
from typing_extensions import Literal, Self, TypedDict, deprecated

from transmission_rpc.constants import DEFAULT_TIMEOUT, LOGGER, RpcMethod
from transmission_rpc.error import (
    TransmissionAuthError,
    TransmissionConnectError,
    TransmissionError,
    TransmissionTimeoutError,
)
from transmission_rpc.session import Session, SessionStats
from transmission_rpc.torrent import Torrent
from transmission_rpc.types import Group, _Timeout
from transmission_rpc.utils import _try_read_torrent, get_torrent_arguments

_hex_chars = frozenset(string.hexdigits.lower())

_TorrentID = Union[int, str]
_TorrentIDs = Union[_TorrentID, List[_TorrentID], None]

_header_session_id = "x-transmission-session-id"


class ResponseData(TypedDict):
    arguments: Any
    tag: int
    result: str


def ensure_location_str(s: str | pathlib.Path) -> str:
    if isinstance(s, pathlib.Path):
        if s.is_absolute():
            return str(s)

        raise ValueError(
            "using relative `pathlib.Path` as remote path is not supported in v4.",
        )

    return str(s)


def _parse_torrent_id(raw_torrent_id: int | str) -> int | str:
    if isinstance(raw_torrent_id, int):
        if raw_torrent_id >= 0:
            return raw_torrent_id
    elif isinstance(raw_torrent_id, str):
        if len(raw_torrent_id) != 40 or (set(raw_torrent_id) - _hex_chars):
            raise ValueError(f"torrent ids {raw_torrent_id} is not valid torrent id, should be a hex str for sha1 hash")
        return raw_torrent_id
    raise ValueError(f"{raw_torrent_id} is not valid torrent id")


def _parse_torrent_ids(args: Any) -> str | list[str | int]:
    if args is None:
        return []
    if isinstance(args, int):
        return [_parse_torrent_id(args)]
    if isinstance(args, str):
        if args == "recently-active":
            return args
        return [_parse_torrent_id(args)]
    if isinstance(args, (list, tuple)):
        return [_parse_torrent_id(item) for item in args]
    raise ValueError(f"Invalid torrent id {args}")


class Client:
    def __init__(
        self,
        *,
        protocol: Literal["http", "https"] = "http",
        username: str | None = None,
        password: str | None = None,
        host: str = "127.0.0.1",
        port: int = 9091,
        path: str = "/transmission/rpc",
        timeout: float = DEFAULT_TIMEOUT,
        logger: logging.Logger = LOGGER,
    ):
        """

        Parameters:
            protocol:
            username:
            password:
            host:
            port:
            path: rpc request target path, default ``/transmission/rpc``
            timeout:
            logger:
        """
        if isinstance(logger, logging.Logger):
            self.logger = logger
        else:
            raise TypeError(
                "logger must be instance of `logging.Logger`, default: logging.getLogger('transmission-rpc')"
            )
        self._query_timeout: _Timeout = timeout

        username = quote(username or "", safe="$-_.+!*'(),;&=", encoding="utf8") if username else ""
        password = ":" + quote(password or "", safe="$-_.+!*'(),;&=", encoding="utf8") if password else ""
        auth = f"{username}{password}@" if (username or password) else ""

        if path == "/transmission/":
            path = "/transmission/rpc"

        url = f"{protocol}://{auth}{host}:{port}{path}"
        self._url = str(url)
        self.__raw_session: dict[str, Any] = {}
        self.__session_id = "0"
        self.__server_version: str = "(unknown)"
        self.__protocol_version: int = 17  # default 17
        self._http_session = requests.Session()
        self._http_session.trust_env = False
        self.__semver_version = None
        self.get_session()
        self.__torrent_get_arguments = get_torrent_arguments(self.__protocol_version)

    @property
    @deprecated("do not use internal property")
    def url(self) -> str:
        return self._url

    @property
    @deprecated("do not use internal property, use `get_torrent_arguments(rpc_version)` if you need")
    def torrent_get_arguments(self) -> list[str]:
        return self.__torrent_get_arguments

    @property
    @deprecated("do not use internal property, use `.get_session()` instead")
    def raw_session(self) -> dict[str, Any]:
        return self.__raw_session

    @property
    @deprecated("do not use internal property")
    def session_id(self) -> str:
        return self.__session_id

    @property
    @deprecated("do not use internal property, use `.get_session().version` instead")
    def server_version(self) -> str:
        return self.__server_version

    @property
    def timeout(self) -> _Timeout:
        """
        Get current timeout for HTTP queries.
        """
        return self._query_timeout

    @timeout.setter
    def timeout(self, value: _Timeout) -> None:
        """
        Set timeout for HTTP queries.
        """
        if isinstance(value, (tuple, list)):
            if len(value) != 2:
                raise ValueError("timeout tuple can only include 2 numbers elements")
            for v in value:
                if not isinstance(v, (float, int)):
                    raise TypeError("element of timeout tuple can only be int or float")
            self._query_timeout = (value[0], value[1])  # for type checker
        elif value is None:
            self._query_timeout = DEFAULT_TIMEOUT
        else:
            self._query_timeout = float(value)

    @timeout.deleter
    def timeout(self) -> None:
        """
        Reset the HTTP query timeout to the default.
        """
        self._query_timeout = DEFAULT_TIMEOUT

    @property
    def _http_header(self) -> dict[str, str]:
        return {_header_session_id: self.__session_id}

    def _http_query(self, query: dict[str, Any], timeout: _Timeout | None = None) -> str:
        """
        Query Transmission through HTTP.
        """
        request_count = 0
        if timeout is None:
            timeout = self.timeout
        while True:
            if request_count >= 3:
                raise TransmissionError("too much request, try enable logger to see what happened")
            self.logger.debug(
                {
                    "url": self._url,
                    "headers": self._http_header,
                    "data": query,
                    "timeout": timeout,
                }
            )

            request_count += 1
            try:
                r = self._http_session.post(
                    self._url,
                    headers=self._http_header,
                    json=query,
                    timeout=timeout,
                )
            except requests.exceptions.Timeout as e:
                raise TransmissionTimeoutError("timeout when connection to transmission daemon") from e
            except requests.exceptions.ConnectionError as e:
                raise TransmissionConnectError(f"can't connect to transmission daemon: {e!s}") from e

            self.logger.debug(r.text)
            if r.status_code in {401, 403}:
                self.logger.debug(r.request.headers)
                raise TransmissionAuthError("transmission daemon require auth", original=r)

            if _header_session_id in r.headers:
                self.__session_id = r.headers["x-transmission-session-id"]

            if r.status_code != 409:
                return r.text

    def _request(
        self,
        method: RpcMethod,
        arguments: dict[str, Any] | None = None,
        ids: _TorrentIDs | None = None,
        require_ids: bool = False,
        timeout: _Timeout | None = None,
    ) -> dict[str, Any]:
        """
        Send json-rpc request to Transmission using http POST
        """
        if not isinstance(method, str):
            raise TypeError("request takes method as string")
        if arguments is None:
            arguments = {}
        if not isinstance(arguments, dict):
            raise TypeError("request takes arguments should be dict")

        ids = _parse_torrent_ids(ids)
        if len(ids) > 0:
            arguments["ids"] = ids
        elif require_ids:
            raise ValueError("request require ids")

        query = {"method": method, "arguments": arguments}

        start = time.monotonic()
        try:
            http_data = self._http_query(query, timeout)
        finally:
            elapsed = time.monotonic() - start
            self.logger.debug("http request took %.3f s", elapsed)

        try:
            data: ResponseData = json.loads(http_data)
        except json.JSONDecodeError as error:
            self.logger.exception("Error:")
            self.logger.exception('Request: "%s"', query)
            self.logger.exception('HTTP data: "%s"', http_data)
            raise TransmissionError(
                "failed to parse response as json", method=method, argument=arguments, rawResponse=http_data
            ) from error

        if self.logger.isEnabledFor(logging.DEBUG):
            self.logger.debug(json.dumps(data, indent=2))

        if "result" not in data:
            raise TransmissionError(
                "Query failed, response data missing without result.",
                method=method,
                argument=arguments,
                response=data,
                rawResponse=http_data,
            )

        if data["result"] != "success":
            raise TransmissionError(
                f'Query failed with result "{data["result"]}".',
                method=method,
                argument=arguments,
                response=data,
                rawResponse=http_data,
            )

        res = data["arguments"]

        results = {}
        if method == RpcMethod.TorrentGet:
            return res
        if method == RpcMethod.TorrentAdd:
            item = None
            if "torrent-added" in res:
                item = res["torrent-added"]
            elif "torrent-duplicate" in res:
                item = res["torrent-duplicate"]
            if item:
                results[item["id"]] = Torrent(fields=item)
            else:
                raise TransmissionError(
                    "Invalid torrent-add response.",
                    method=method,
                    argument=arguments,
                    response=data,
                    rawResponse=http_data,
                )
        elif method == RpcMethod.SessionGet:
            self.__raw_session.update(res)
        elif method == RpcMethod.SessionStats:
            # older versions of T has the return data in "session-stats"
            if "session-stats" in res:
                return res["session-stats"]
            return res
        elif method in (
            RpcMethod.PortTest,
            RpcMethod.BlocklistUpdate,
            RpcMethod.FreeSpace,
            RpcMethod.TorrentRenamePath,
        ):
            return res
        else:
            return res

        return results

    def _update_server_version(self) -> None:
        """Decode the Transmission version string, if available."""
        self.__semver_version = self.__raw_session.get("rpc-version-semver")
        self.__server_version = self.__raw_session["version"]
        self.__protocol_version = self.__raw_session["rpc-version"]

    @property
    @deprecated("use .get_session().rpc_version_semver instead")
    def semver_version(self) -> str | None:
        """Get the Transmission daemon RPC version.

        .. deprecated:: 7.0.5
            Use ``.get_session().rpc_version_semver`` instead
        """
        return self.__semver_version

    @property
    @deprecated("use .get_session().rpc_version instead")
    def rpc_version(self) -> int:
        """Get the Transmission daemon RPC version.

        .. deprecated:: 7.0.5
            Use ``.get_session().rpc_version`` instead
        """
        return self.__protocol_version

    def _rpc_version_warning(self, required_version: int) -> None:
        """
        Add a warning to the log if the Transmission RPC version is lower then the provided version.
        """
        if self.__protocol_version < required_version:
            self.logger.warning(
                "Using feature not supported by server. RPC version for server %d, feature introduced in %d.",
                self.__protocol_version,
                required_version,
            )

    def add_torrent(
        self,
        torrent: BinaryIO | str | bytes | pathlib.Path,
        timeout: _Timeout | None = None,
        *,
        download_dir: str | None = None,
        files_unwanted: list[int] | None = None,
        files_wanted: list[int] | None = None,
        paused: bool | None = None,
        peer_limit: int | None = None,
        priority_high: list[int] | None = None,
        priority_low: list[int] | None = None,
        priority_normal: list[int] | None = None,
        cookies: str | None = None,
        labels: Iterable[str] | None = None,
        bandwidthPriority: int | None = None,
    ) -> Torrent:
        """
        Add torrent to transfers list. ``torrent`` can be:

        - ``http://``, ``https://`` or  ``magnet:`` URL
        - torrent file-like object in **binary mode**
        - bytes of torrent content
        - ``pathlib.Path`` for local torrent file, will be read and encoded as base64.

        Warnings:
            base64 string or ``file://`` protocol URL are not supported in v4.

        Parameters:
            torrent:
                torrent to add
            timeout:
                request timeout
            bandwidthPriority:
                Priority for this transfer.
            cookies:
                One or more HTTP cookie(s).
            download_dir:
                The directory where the downloaded contents will be saved in.
            files_unwanted:
                A list of file id's that shouldn't be downloaded.
            files_wanted:
                A list of file id's that should be downloaded.
            paused:
                If ``True``, does not start the transfer when added.
                Magnet url will always start to downloading torrents.
            peer_limit:
                Maximum number of peers allowed.
            priority_high:
                A list of file id's that should have high priority.
            priority_low:
                A list of file id's that should have low priority.
            priority_normal:
                A list of file id's that should have normal priority.
            labels:
                Array of string labels.
                Add in rpc 17.
        """
        if torrent is None:
            raise ValueError("add_torrent requires data or a URI.")

        if labels is not None:
            self._rpc_version_warning(17)

        kwargs: dict[str, Any] = remove_unset_value(
            {
                "download-dir": download_dir,
                "files-unwanted": files_unwanted,
                "files-wanted": files_wanted,
                "paused": paused,
                "peer-limit": peer_limit,
                "priority-high": priority_high,
                "priority-low": priority_low,
                "priority-normal": priority_normal,
                "bandwidthPriority": bandwidthPriority,
                "cookies": cookies,
                "labels": list_or_none(_single_str_as_list(labels)),
            }
        )

        torrent_data = _try_read_torrent(torrent)
        if torrent_data is None:
            kwargs["filename"] = torrent
        else:
            if not torrent_data:
                raise ValueError("Torrent metadata is empty")
            kwargs["metainfo"] = torrent_data

        return next(iter(self._request(RpcMethod.TorrentAdd, kwargs, timeout=timeout).values()))

    def remove_torrent(self, ids: _TorrentIDs, delete_data: bool = False, timeout: _Timeout | None = None) -> None:
        """
        remove torrent(s) with provided id(s).

        Local data will be removed by transmission daemon if ``delete_data`` is set to ``True``.
        """
        self._request(
            RpcMethod.TorrentRemove,
            {"delete-local-data": delete_data},
            ids,
            True,
            timeout=timeout,
        )

    def start_torrent(self, ids: _TorrentIDs, bypass_queue: bool = False, timeout: _Timeout | None = None) -> None:
        """Start torrent(s) with provided id(s)"""
        method = RpcMethod.TorrentStart
        if bypass_queue:
            method = RpcMethod.TorrentStartNow
        self._request(method, {}, ids, True, timeout=timeout)

    def start_all(self, bypass_queue: bool = False, timeout: _Timeout | None = None) -> None:
        """Start all torrents respecting the queue order"""
        method = RpcMethod.TorrentStart
        if bypass_queue:
            method = RpcMethod.TorrentStartNow
        torrent_list = sorted(self.get_torrents(), key=lambda t: t.queue_position)
        self._request(
            method,
            {},
            ids=[x.id for x in torrent_list],
            require_ids=True,
            timeout=timeout,
        )

    def stop_torrent(self, ids: _TorrentIDs, timeout: _Timeout | None = None) -> None:
        """stop torrent(s) with provided id(s)"""
        self._request(RpcMethod.TorrentStop, {}, ids, True, timeout=timeout)

    def verify_torrent(self, ids: _TorrentIDs, timeout: _Timeout | None = None) -> None:
        """verify torrent(s) with provided id(s)"""
        self._request(RpcMethod.TorrentVerify, {}, ids, True, timeout=timeout)

    def reannounce_torrent(self, ids: _TorrentIDs, timeout: _Timeout | None = None) -> None:
        """Reannounce torrent(s) with provided id(s)"""
        self._request(RpcMethod.TorrentReannounce, {}, ids, True, timeout=timeout)

    def get_torrent(
        self,
        torrent_id: _TorrentID,
        arguments: Iterable[str] | None = None,
        timeout: _Timeout | None = None,
    ) -> Torrent:
        """
        Get information for torrent with provided id.
        ``arguments`` contains a list of field names to be returned, when ``arguments=None`` (default),
        all fields are requested. See the Torrent class for more information.

        new argument ``format`` in rpc_version 16 is unnecessarily
        and this lib can't handle table response, So it's unsupported.

        Returns a Torrent object with the requested fields.

        Note:
            It's recommended that you only fetch arguments you need,
            this could improve response speed.

            For example, fetch all fields from transmission daemon with 1500 torrents would take ~5s,
            but is only ~0.2s if to fetch 6 fields.

        Parameters:
            torrent_id:
                torrent id can be an int or a torrent ``info_hash`` (``hashString`` property of the ``Torrent`` object).

            arguments:
                fetched torrent arguments, in most cases you don't need to set this,
                transmission-rpc will fetch all torrent fields it supported.

            timeout:
                requests timeout

        Raises:
            KeyError: torrent with given ``torrent_id`` not found
        """
        if arguments:
            arguments = list(set(arguments) | {"id", "hashString"})
        else:
            arguments = self.__torrent_get_arguments
        torrent_id = _parse_torrent_id(torrent_id)
        if torrent_id is None:
            raise ValueError("Invalid id")

        result = self._request(
            RpcMethod.TorrentGet,
            {"fields": arguments},
            torrent_id,
            require_ids=True,
            timeout=timeout,
        )

        for torrent in result["torrents"]:
            if torrent.get("hashString") == torrent_id or torrent.get("id") == torrent_id:
                return Torrent(fields=torrent)
        raise KeyError("Torrent not found in result")

    def get_torrents(
        self,
        ids: _TorrentIDs | None = None,
        arguments: Iterable[str] | None = None,
        timeout: _Timeout | None = None,
    ) -> list[Torrent]:
        """
        Get information for torrents with provided ids. For more information see :py:meth:`Client.get_torrent`.

        Returns a list of Torrent object.
        """
        if arguments:
            arguments = list(set(arguments) | {"id", "hashString"})
        else:
            arguments = self.__torrent_get_arguments
        return [
            Torrent(fields=x)
            for x in self._request(RpcMethod.TorrentGet, {"fields": arguments}, ids, timeout=timeout)["torrents"]
        ]

    def get_recently_active_torrents(
        self, arguments: Iterable[str] | None = None, timeout: _Timeout | None = None
    ) -> tuple[list[Torrent], list[int]]:
        """
        Get information for torrents for recently active torrent. If you want to get recently-removed
        torrents. you should use this method.

        Returns:
            active_torrents, removed_torrents
                list of recently active torrents and list of torrent-id of recently-removed torrents.
        """
        if arguments:
            arguments = list(set(arguments) | {"id", "hashString"})
        else:
            arguments = self.__torrent_get_arguments

        result = self._request(RpcMethod.TorrentGet, {"fields": arguments}, "recently-active", timeout=timeout)

        return [Torrent(fields=x) for x in result["torrents"]], result["removed"]

    def change_torrent(
        self,
        ids: _TorrentIDs,
        timeout: _Timeout | None = None,
        *,
        bandwidth_priority: int | None = None,
        download_limit: int | None = None,
        download_limited: bool | None = None,
        upload_limit: int | None = None,
        upload_limited: bool | None = None,
        files_unwanted: Iterable[int] | None = None,
        files_wanted: Iterable[int] | None = None,
        honors_session_limits: bool | None = None,
        location: str | None = None,
        peer_limit: int | None = None,
        priority_high: Iterable[int] | None = None,
        priority_low: Iterable[int] | None = None,
        priority_normal: Iterable[int] | None = None,
        queue_position: int | None = None,
        seed_idle_limit: int | None = None,
        seed_idle_mode: int | None = None,
        seed_ratio_limit: float | None = None,
        seed_ratio_mode: int | None = None,
        tracker_add: Iterable[str] | None = None,
        labels: Iterable[str] | None = None,
        group: str | None = None,
        tracker_list: Iterable[Iterable[str]] | None = None,
        tracker_replace: Iterable[tuple[int, str]] | None = None,
        tracker_remove: Iterable[int] | None = None,
        **kwargs: Any,
    ) -> None:
        """Change torrent parameters for the torrent(s) with the supplied id's.

        Parameters:
            ids: torrent(s) to change.
            timeout: requesst timeout.
            honors_session_limits: true if session upload limits are honored.
            location: new location of the torrent's content
            peer_limit: maximum number of peers
            queue_position: position of this torrent in its queue [0...n)
            files_wanted: Array of file id to download.
            files_unwanted: Array of file id to not download.
            download_limit: maximum download speed (KBps)
            download_limited: true if ``download_limit`` is honored
            upload_limit: maximum upload speed (KBps)
            upload_limited: true if ``upload_limit`` is honored
            bandwidth_priority: Priority for this transfer.
            priority_high: list of file id to set high download priority
            priority_low: list of file id to set low download priority
            priority_normal: list of file id to set normal download priority
            seed_ratio_limit: Seed inactivity limit in minutes.
            seed_ratio_mode: Torrent seed ratio mode
                Valid options are :py:class:`transmission_rpc.RatioLimitMode`
            seed_idle_limit: torrent-level seeding ratio
            seed_idle_mode: Seed inactivity mode.
                Valid options are :py:class:`transmission_rpc.IdleMode`
            labels: Array of string labels. Add in rpc 16.
            group: The name of this torrent's bandwidth group. Add in rpc 17.
            tracker_list:
                A ``Iterable[Iterable[str]]``, each ``Iterable[str]`` for a tracker tier.

                Add in rpc 17.

                Example: ``[['https://tracker1/announce', 'https://tracker2/announce'],
                ['https://backup1.example.com/announce'], ['https://backup2.example.com/announce']]``.

            tracker_add:
                Array of string with announce URLs to add.

                Warnings:
                    since transmission daemon 4.0.0, this argument is deprecated, use ``tracker_list`` instead.

            tracker_remove:
                Array of ids of trackers to remove.

                Warnings:
                    since transmission daemon 4.0.0, this argument is deprecated, use ``tracker_list`` instead.

            tracker_replace:
                Array of (id, url) tuples where the announcement URL should be replaced.

                Warning:
                    since transmission daemon 4.0.0, this argument is deprecated, use ``tracker_list`` instead.

        Warnings:
            ``kwargs`` is for the future features not supported yet, it's not compatibility promising.
            It will be bypassed to request arguments **as-is**, the underline in the key will not be replaced, so you should use kwargs like ``{'a-argument': 'value'}``
        """
        if labels is not None:
            self._rpc_version_warning(16)

        if tracker_list is not None:
            self._rpc_version_warning(17)

        if group is not None:
            self._rpc_version_warning(17)

        args: dict[str, Any] = remove_unset_value(
            {
                "bandwidthPriority": bandwidth_priority,
                "downloadLimit": download_limit,
                "downloadLimited": download_limited,
                "uploadLimit": upload_limit,
                "uploadLimited": upload_limited,
                "files-unwanted": list_or_none(files_unwanted),
                "files-wanted": list_or_none(files_wanted),
                "honorsSessionLimits": honors_session_limits,
                "location": location,
                "peer-limit": peer_limit,
                "priority-high": list_or_none(priority_high),
                "priority-low": list_or_none(priority_low),
                "priority-normal": list_or_none(priority_normal),
                "queuePosition": queue_position,
                "seedIdleLimit": seed_idle_limit,
                "seedIdleMode": seed_idle_mode,
                "seedRatioLimit": seed_ratio_limit,
                "seedRatioMode": seed_ratio_mode,
                "trackerAdd": tracker_add,
                "trackerRemove": tracker_remove,
                "trackerReplace": tracker_replace,
                "labels": list_or_none(_single_str_as_list(labels)),
                "trackerList": None if tracker_list is None else "\n\n".join("\n".join(tier) for tier in tracker_list),
                "group": group,
            }
        )

        args.update(kwargs)

        if args:
            self._request(RpcMethod.TorrentSet, args, ids, True, timeout=timeout)
        else:
            raise ValueError("No arguments to set")

    def move_torrent_data(
        self,
        ids: _TorrentIDs,
        location: str | pathlib.Path,
        timeout: _Timeout | None = None,
        *,
        move: bool = True,
    ) -> None:
        """
        Move torrent data to the new location.

        See Also:
            `RPC Spec: moving-a-torrent <https://github.com/transmission/transmission/blob/main/docs/rpc-spec.md#36-moving-a-torrent>`_
        """
        args = {"location": ensure_location_str(location), "move": bool(move)}
        self._request(RpcMethod.TorrentSetLocation, args, ids, True, timeout=timeout)

    def rename_torrent_path(
        self,
        torrent_id: _TorrentID,
        location: str,
        name: str,
        timeout: _Timeout | None = None,
    ) -> tuple[str, str]:
        """
        Warnings:
            This method can only be called on single torrent.

        Warnings:
            This is not the method to move torrent data directory,

        See Also:
            `RPC Spec: renaming-a-torrents-path <https://github.com/transmission/transmission/blob/main/docs/rpc-spec.md#37-renaming-a-torrents-path>`_
        """
        self._rpc_version_warning(15)
        torrent_id = _parse_torrent_id(torrent_id)

        name = name.strip()  # https://github.com/trim21/transmission-rpc/issues/185

        result = self._request(
            RpcMethod.TorrentRenamePath,
            {"path": ensure_location_str(location), "name": name},
            torrent_id,
            True,
            timeout=timeout,
        )

        return result["path"], result["name"]

    def queue_top(self, ids: _TorrentIDs, timeout: _Timeout | None = None) -> None:
        """
        Move transfer to the top of the queue.

        https://github.com/transmission/transmission/blob/main/docs/rpc-spec.md#46-queue-movement-requests
        """
        self._request(RpcMethod.QueueMoveTop, ids=ids, require_ids=True, timeout=timeout)

    def queue_bottom(self, ids: _TorrentIDs, timeout: _Timeout | None = None) -> None:
        """
        Move transfer to the bottom of the queue.

        https://github.com/transmission/transmission/blob/main/docs/rpc-spec.md#46-queue-movement-requests
        """
        self._request(RpcMethod.QueueMoveBottom, ids=ids, require_ids=True, timeout=timeout)

    def queue_up(self, ids: _TorrentIDs, timeout: _Timeout | None = None) -> None:
        """Move transfer up in the queue."""
        self._request(RpcMethod.QueueMoveUp, ids=ids, require_ids=True, timeout=timeout)

    def queue_down(self, ids: _TorrentIDs, timeout: _Timeout | None = None) -> None:
        """Move transfer down in the queue."""
        self._request(RpcMethod.QueueMoveDown, ids=ids, require_ids=True, timeout=timeout)

    def get_session(self, timeout: _Timeout | None = None) -> Session:
        """
        Get session parameters. See the Session class for more information.
        """
        self._request(RpcMethod.SessionGet, timeout=timeout)
        self._update_server_version()
        return Session(fields=self.__raw_session)

    def set_session(
        self,
        timeout: _Timeout | None = None,
        *,
        alt_speed_down: int | None = None,
        alt_speed_enabled: bool | None = None,
        alt_speed_time_begin: int | None = None,
        alt_speed_time_day: int | None = None,
        alt_speed_time_enabled: bool | None = None,
        alt_speed_time_end: int | None = None,
        alt_speed_up: int | None = None,
        blocklist_enabled: bool | None = None,
        blocklist_url: str | None = None,
        cache_size_mb: int | None = None,
        dht_enabled: bool | None = None,
        default_trackers: Iterable[str] | None = None,
        download_dir: str | None = None,
        download_queue_enabled: bool | None = None,
        download_queue_size: int | None = None,
        encryption: Literal["required", "preferred", "tolerated"] | None = None,
        idle_seeding_limit: int | None = None,
        idle_seeding_limit_enabled: bool | None = None,
        incomplete_dir: str | None = None,
        incomplete_dir_enabled: bool | None = None,
        lpd_enabled: bool | None = None,
        peer_limit_global: int | None = None,
        peer_limit_per_torrent: int | None = None,
        peer_port: int | None = None,
        peer_port_random_on_start: bool | None = None,
        pex_enabled: bool | None = None,
        port_forwarding_enabled: bool | None = None,
        queue_stalled_enabled: bool | None = None,
        queue_stalled_minutes: int | None = None,
        rename_partial_files: bool | None = None,
        script_torrent_done_enabled: bool | None = None,
        script_torrent_done_filename: str | None = None,
        seed_queue_enabled: bool | None = None,
        seed_queue_size: int | None = None,
        seed_ratio_limit: float | None = None,
        seed_ratio_limited: bool | None = None,
        speed_limit_down: int | None = None,
        speed_limit_down_enabled: bool | None = None,
        speed_limit_up: int | None = None,
        speed_limit_up_enabled: bool | None = None,
        start_added_torrents: bool | None = None,
        trash_original_torrent_files: bool | None = None,
        utp_enabled: bool | None = None,
        script_torrent_done_seeding_filename: str | None = None,
        script_torrent_done_seeding_enabled: bool | None = None,
        script_torrent_added_enabled: bool | None = None,
        script_torrent_added_filename: str | None = None,
        **kwargs: Any,
    ) -> None:
        """
        Set session parameters.

        Parameters:
            timeout
                request timeout
            alt_speed_down:
                max global download speed (KBps)
            alt_speed_enabled:
                true means use the alt speeds
            alt_speed_time_begin:
                Time when alternate speeds should be enabled. Minutes after midnight.
            alt_speed_time_day:
                Enables alternate speeds scheduling these days.
            alt_speed_time_enabled:
                Enables alternate speeds scheduling.
            alt_speed_time_end:
                Time when alternate speeds should be disabled. Minutes after midnight.
            alt_speed_up:
                Alternate session upload speed limit (in Kib/s).
            blocklist_enabled:
                Enables the block list
            blocklist_url:
                Location of the block list. Updated with blocklist-update.
            cache_size_mb:
                The maximum size of the disk cache in MB
            default_trackers:
                list of default trackers to use on public torrents.
            dht_enabled:
                Enables DHT.
            download_dir:
                Set the session download directory.
            download_queue_enabled:
                Enables download queue.
            download_queue_size:
                Number of slots in the download queue.
            encryption:
                Set the session encryption mode, one of ``required``, ``preferred`` or ``tolerated``.
            idle_seeding_limit:
                The default seed inactivity limit in minutes.
            idle_seeding_limit_enabled:
                Enables the default seed inactivity limit
            incomplete_dir:
                The path to the directory of incomplete transfer data.
            incomplete_dir_enabled:
                Enables the incomplete transfer data directory,
                Otherwise data for incomplete transfers are stored in the download target.
            lpd_enabled:
                Enables local peer discovery for public torrents.
            peer_limit_global:
                Maximum number of peers.
            peer_limit_per_torrent:
                Maximum number of peers per transfer.
            peer_port:
                Peer port.
            peer_port_random_on_start:
                Enables randomized peer port on start of Transmission.
            pex_enabled:
                Allowing PEX in public torrents.
            port_forwarding_enabled:
                Enables port forwarding.
            queue_stalled_enabled:
                Enable tracking of stalled transfers.
            queue_stalled_minutes:
                Number of minutes of idle that marks a transfer as stalled.
            rename_partial_files:
                Appends ".part" to incomplete files

            seed_queue_enabled:
                Enables upload queue.
            seed_queue_size:
                Number of slots in the upload queue.
            seed_ratio_limit:
                Seed ratio limit. 1.0 means 1:1 download and upload ratio.
            seed_ratio_limited:
                Enables seed ration limit.
            speed_limit_down:
                Download speed limit (in Kib/s).
            speed_limit_down_enabled:
                Enables download speed limiting.
            speed_limit_up:
                Upload speed limit (in Kib/s).
            speed_limit_up_enabled:
                Enables upload speed limiting.
            start_added_torrents:
                Added torrents will be started right away.
            trash_original_torrent_files:
                The .torrent file of added torrents will be deleted.
            utp_enabled:
                Enables Micro Transport Protocol (UTP).
            script_torrent_done_enabled:
                Whether to call the "done" script.
            script_torrent_done_filename:
                Filename of the script to run when the transfer is done.
            script_torrent_added_filename:
                filename of the script to run
            script_torrent_added_enabled:
                whether or not to call the ``added`` script
            script_torrent_done_seeding_enabled:
                whether or not to call the ``seeding-done`` script
            script_torrent_done_seeding_filename:
                filename of the script to run

        Warnings:
            ``kwargs`` is pass the arguments not supported yet future, it's not compatibility promising.
            transmission-rpc will merge ``kwargs`` in rpc arguments **as-is**
        """

        if encryption is not None and encryption not in ["required", "preferred", "tolerated"]:
            raise ValueError("Invalid encryption value")

        if default_trackers is not None:
            self._rpc_version_warning(17)
        if script_torrent_done_seeding_filename is not None:
            self._rpc_version_warning(17)
        if script_torrent_done_seeding_enabled is not None:
            self._rpc_version_warning(17)
        if script_torrent_added_enabled is not None:
            self._rpc_version_warning(17)
        if script_torrent_added_filename is not None:
            self._rpc_version_warning(17)

        args: dict[str, Any] = remove_unset_value(
            {
                "alt-speed-down": alt_speed_down,
                "alt-speed-enabled": alt_speed_enabled,
                "alt-speed-time-begin": alt_speed_time_begin,
                "alt-speed-time-day": alt_speed_time_day,
                "alt-speed-time-enabled": alt_speed_time_enabled,
                "alt-speed-time-end": alt_speed_time_end,
                "alt-speed-up": alt_speed_up,
                "blocklist-enabled": blocklist_enabled,
                "blocklist-url": blocklist_url,
                "cache-size-mb": cache_size_mb,
                "dht-enabled": dht_enabled,
                "download-dir": download_dir,
                "download-queue-enabled": download_queue_enabled,
                "download-queue-size": download_queue_size,
                "idle-seeding-limit-enabled": idle_seeding_limit_enabled,
                "idle-seeding-limit": idle_seeding_limit,
                "incomplete-dir": incomplete_dir,
                "incomplete-dir-enabled": incomplete_dir_enabled,
                "lpd-enabled": lpd_enabled,
                "peer-limit-global": peer_limit_global,
                "peer-limit-per-torrent": peer_limit_per_torrent,
                "peer-port-random-on-start": peer_port_random_on_start,
                "peer-port": peer_port,
                "pex-enabled": pex_enabled,
                "port-forwarding-enabled": port_forwarding_enabled,
                "queue-stalled-enabled": queue_stalled_enabled,
                "queue-stalled-minutes": queue_stalled_minutes,
                "rename-partial-files": rename_partial_files,
                "script-torrent-done-enabled": script_torrent_done_enabled,
                "script-torrent-done-filename": script_torrent_done_filename,
                "seed-queue-enabled": seed_queue_enabled,
                "seed-queue-size": seed_queue_size,
                "seedRatioLimit": seed_ratio_limit,
                "seedRatioLimited": seed_ratio_limited,
                "speed-limit-down": speed_limit_down,
                "speed-limit-down-enabled": speed_limit_down_enabled,
                "speed-limit-up": speed_limit_up,
                "speed-limit-up-enabled": speed_limit_up_enabled,
                "start-added-torrents": start_added_torrents,
                "trash-original-torrent-files": trash_original_torrent_files,
                "utp-enabled": utp_enabled,
                "encryption": encryption,
                "script-torrent-added-filename": script_torrent_added_filename,
                "script-torrent-done-seeding-filename": script_torrent_done_seeding_filename,
                "script-torrent-done-seeding-enabled": script_torrent_done_seeding_enabled,
                "script-torrent-added-enabled": script_torrent_added_enabled,
                "default-trackers": "\n".join(default_trackers) if default_trackers is not None else None,
            }
        )

        args.update(kwargs)

        if args:
            self._request(RpcMethod.SessionSet, args, timeout=timeout)

    def blocklist_update(self, timeout: _Timeout | None = None) -> int | None:
        """Update block list. Returns the size of the block list."""
        result = self._request(RpcMethod.BlocklistUpdate, timeout=timeout)
        return result.get("blocklist-size")

    def port_test(self, timeout: _Timeout | None = None) -> bool | None:
        """
        Tests to see if your incoming peer port is accessible from the
        outside world.
        """
        result = self._request(RpcMethod.PortTest, timeout=timeout)
        return result.get("port-is-open")

    def free_space(self, path: str | pathlib.Path, timeout: _Timeout | None = None) -> int | None:
        """
        Get the amount of free space (in bytes) at the provided location.
        """
        self._rpc_version_warning(15)
        path = ensure_location_str(path)
        result: dict[str, Any] = self._request(RpcMethod.FreeSpace, {"path": path}, timeout=timeout)
        if result["path"] == path:
            return result["size-bytes"]
        return None

    def session_stats(self, timeout: _Timeout | None = None) -> SessionStats:
        """Get session statistics"""
        result = self._request(RpcMethod.SessionStats, timeout=timeout)
        return SessionStats(fields=result)

    def set_group(
        self,
        name: str,
        *,
        timeout: _Timeout | None = None,
        honors_session_limits: bool | None = None,
        speed_limit_down_enabled: bool | None = None,
        speed_limit_down: int | None = None,
        speed_limit_up_enabled: bool | None = None,
        speed_limit_up: int | None = None,
    ) -> None:
        """create or update a Bandwidth group.

        :param name: Bandwidth group name
        :param honors_session_limits: true if session upload limits are honored
        :param speed_limit_down_enabled: true means enabled
        :param speed_limit_down: 	max global download speed (KBps)
        :param speed_limit_up_enabled: 	true means enabled
        :param speed_limit_up: max global upload speed (KBps)
        :param timeout: request timeout
        """

        self._rpc_version_warning(17)
        arguments: dict[str, Any] = remove_unset_value(
            {
                "name": name,
                "honorsSessionLimits": honors_session_limits,
                "speed-limit-down": speed_limit_down,
                "speed-limit-up-enabled": speed_limit_up_enabled,
                "speed-limit-up": speed_limit_up,
                "speed-limit-down-enabled": speed_limit_down_enabled,
            }
        )

        self._request(RpcMethod.GroupSet, arguments, timeout=timeout)

    def get_group(self, name: str, *, timeout: _Timeout | None = None) -> Group | None:
        self._rpc_version_warning(17)
        result: dict[str, Any] = self._request(RpcMethod.GroupGet, {"group": name}, timeout=timeout)

        if result["group"]:
            return Group(fields=result["group"][0])

        return None

    def get_groups(self, name: list[str] | None = None, *, timeout: _Timeout | None = None) -> dict[str, Group]:
        payload = {}
        if name is not None:
            payload = {"group": name}

        result: dict[str, Any] = self._request(RpcMethod.GroupGet, payload, timeout=timeout)

        return {x["name"]: Group(fields=x) for x in result["group"]}

    def __enter__(self) -> Self:
        return self

    def __exit__(
        self,
        exc_type: type[BaseException] | None,
        exc_val: BaseException | None,
        exc_tb: types.TracebackType | None,
    ) -> None:
        self._http_session.close()


T = TypeVar("T")


def _single_str_as_list(v: Iterable[str] | None) -> list[str] | None:
    if v is None:
        return v
    if isinstance(v, str):
        return [v]
    return list(v)


def list_or_none(v: Iterable[T] | None) -> list[T] | None:
    if v is None:
        return None
    return list(v)


def remove_unset_value(data: dict[str, Any]) -> dict[str, Any]:
    return {key: value for key, value in data.items() if value is not None}