Add TransmissionBt client implementation

src/clients/transmission.py

@@ -0,0 +1,134 @@
+import base64
+import json
+from enum import Enum
+from http import HTTPStatus
+
+import requests
+
+from requests.auth import HTTPBasicAuth
+from requests.structures import CaseInsensitiveDict
+
+from ..filesystem import sane_join
+from ..parser import get_bencoded_data, calculate_infohash
+from ..errors import TorrentClientError, TorrentClientAuthenticationError, TorrentExistsInClientError
+from .torrent_client import TorrentClient
+
+class StatusEnum(Enum):
+  STOPPED = 0
+  QUEUED_VERIFY = 1
+  VERIFYING = 2
+  QUEUE_DOWNLOAD = 3
+  DOWNLOADING = 4
+  QUEUED_SEED = 5
+  SEEDING = 6
+
+class TransmissionBt(TorrentClient):
+  X_TRANSMISSION_SESSION_ID = 'X-Transmission-Session-Id'
+
+  def __init__(self, rpc_url):
+    super().__init__()
+    transmission_url_parts = self._extract_credentials_from_url(rpc_url, "transmission/rpc")
+    self._base_url = transmission_url_parts[0]
+    self._basic_auth = HTTPBasicAuth(transmission_url_parts[1], transmission_url_parts[2])
+    self._transmission_session_id = None
+
+  def setup(self):
+    self.__authenticate()
+    return self
+
+  def get_torrent_info(self, infohash):
+    response = self.__wrap_request("torrent-get", arguments={
+      "fields": ["labels", "downloadDir", "percentDone", "status", "doneDate", "name"],
+      "ids": [infohash]
+    })
+
+    if response:
+      try:
+        parsed_response = json.loads(response)
+      except json.JSONDecodeError as json_parse_error:
+        raise TorrentClientError(f"Client returned malformed json response") from json_parse_error
+
+      if not parsed_response.get("arguments", {}).get("torrents", []):
+        raise TorrentClientError(f"Torrent not found in client ({infohash})")
+
+      torrent = parsed_response['arguments']['torrents'][0]
+      torrent_completed = (torrent["percentDone"] == 1.0 or torrent["doneDate"] > 0) and torrent["status"] in [StatusEnum.SEEDING.value, StatusEnum.QUEUED_SEED.value]
+
+      return {
+        "complete": torrent_completed,
+        "label": torrent["labels"],
+        "save_path": torrent["downloadDir"],
+        "content_path": sane_join(torrent["downloadDir"], torrent["name"]),
+      }
+    else:
+      raise TorrentClientError("Client returned unexpected response")
+
+  def inject_torrent(self, source_torrent_infohash, new_torrent_filepath, save_path_override=None):
+    source_torrent_info = self.get_torrent_info(source_torrent_infohash)
+
+    if not source_torrent_info["complete"]:
+      raise TorrentClientError("Cannot inject a torrent that is not complete")
+
+    new_torrent_infohash = calculate_infohash(get_bencoded_data(new_torrent_filepath)).lower()
+    new_torrent_already_exists = self.__does_torrent_exist_in_client(new_torrent_infohash)
+    if new_torrent_already_exists:
+      raise TorrentExistsInClientError(f"New torrent already exists in client ({new_torrent_infohash})")
+
+    self.__wrap_request("torrent-add", arguments={
+      "download-dir": save_path_override if save_path_override else source_torrent_info["save_path"],
+      "metainfo": base64.b64encode(open(new_torrent_filepath, "rb").read()).decode("utf-8"),
+      "labels": source_torrent_info['label']
+    })
+
+    return new_torrent_infohash
+
+  def __authenticate(self):
+    try:
+      # This method specifically does not use the __wrap_request method
+      # because we want to avoid an infinite loop of re-authenticating
+      response = requests.post(self._base_url, auth=self._basic_auth)
+      if response.status_code == HTTPStatus.CONFLICT:
+        self._transmission_session_id = response.headers[self.X_TRANSMISSION_SESSION_ID]
+      else:
+        response.raise_for_status()
+    except requests.RequestException as e:
+      raise TorrentClientAuthenticationError(f"TransmissionBt login failed: {e}")
+
+    if not self._transmission_session_id:
+      raise TorrentClientAuthenticationError("TransmissionBt login failed: Invalid username or password")
+
+  def __wrap_request(self, method, arguments, files=None):
+    try:
+      return self.__request(method, arguments, files)
+    except TorrentClientAuthenticationError:
+      self.__authenticate()
+      return self.__request(method, arguments, files)
+
+  def __request(self, method, arguments=None, files=None):
+    try:
+      response = requests.post(
+        self._base_url,
+        auth=self._basic_auth,
+        headers=CaseInsensitiveDict({self.X_TRANSMISSION_SESSION_ID: self._transmission_session_id}),
+        json={
+          "method": method,
+          "arguments": arguments
+        },
+        files=files,
+      )
+
+      response.raise_for_status()
+
+      return response.text
+    except requests.RequestException as e:
+      if e.response.status_code == 401:
+        print(e.response.text)
+        raise TorrentClientAuthenticationError("Failed to authenticate with TransmissionBt")
+
+      raise TorrentClientError(f"TransmissionBt request to '{self._base_url}' for method '{method}' failed: {e}")
+
+  def __does_torrent_exist_in_client(self, infohash):
+    try:
+      return bool(self.get_torrent_info(infohash))
+    except TorrentClientError:
+      return False

src/config.py

@@ -23,6 +23,7 @@ def build_config_dict(cls, config_filepath: str, env_vars: dict):
         "port": env_vars.get("PORT"),
         "inject_torrents": True if env_vars.get("INJECT_TORRENTS", "").lower().strip() == "true" else False,
         "deluge_rpc_url": env_vars.get("DELUGE_RPC_URL"),
+        "transmission_rpc_url": env_vars.get("TRANSMISSION_RPC_URL"),
         "qbittorrent_url": env_vars.get("QBITTORRENT_URL"),
         "injection_link_directory": env_vars.get("INJECTION_LINK_DIRECTORY"),
       }.items()
@@ -50,6 +51,10 @@ def server_port(self) -> str:
   def deluge_rpc_url(self) -> ParseResult | None:
     return self._config.get("deluge_rpc_url")
 
+  @property
+  def transmission_rpc_url(self) -> ParseResult | None:
+    return self._config.get("transmission_rpc_url")
+
   @property
   def qbittorrent_url(self) -> ParseResult | None:
     return self._config.get("qbittorrent_url")

src/config_validator.py

@@ -7,7 +7,7 @@
 
 class ConfigValidator:
   REQUIRED_KEYS = ["red_key", "ops_key"]
-  TORRENT_CLIENT_KEYS = ["deluge_rpc_url", "qbittorrent_url"]
+  TORRENT_CLIENT_KEYS = ["deluge_rpc_url", "transmission_rpc_url", "qbittorrent_url"]
 
   def __init__(self, config_dict):
     self.config_dict = config_dict
@@ -16,6 +16,7 @@ def __init__(self, config_dict):
       "ops_key": self.__is_valid_ops_key,
       "port": self.__is_valid_port,
       "deluge_rpc_url": self.__is_valid_deluge_url,
+      "transmission_rpc_url": self.__is_valid_transmission_rpc_url,
       "qbittorrent_url": self.__is_valid_qbit_url,
       "inject_torrents": self.__is_boolean,
       "injection_link_directory": assert_path_exists,
@@ -97,6 +98,17 @@ def __is_valid_deluge_url(url):
       return parsed_url.geturl()  # return the parsed URL
     raise ValueError(f'Invalid "deluge_rpc_url" provided: {url}')
 
+  @staticmethod
+  def __is_valid_transmission_rpc_url(url):
+    parsed_url = urlparse(url)
+    if parsed_url.scheme and parsed_url.netloc:
+      if not parsed_url.password:
+        raise Exception(
+          "You need to define a password in the TransmissionBt RPC URL. (e.g. http://:<PASSWORD>@localhost:51413/transmission/rpc)"
+        )
+      return parsed_url.geturl()  # return the parsed URL
+    raise ValueError(f'Invalid "transmission_rpc_url" provided: {url}')
+
   @staticmethod
   def __is_boolean(value):
     coerced = value.lower().strip()

src/injection.py

@@ -3,6 +3,7 @@
 
 from .clients.deluge import Deluge
 from .clients.qbittorrent import Qbittorrent
+from .clients.transmission import TransmissionBt
 from .config import Config
 from .errors import TorrentInjectionError
 from .parser import calculate_infohash, get_bencoded_data
@@ -39,7 +40,7 @@ def __validate_config(config: Config):
     if not config.injection_link_directory:
       raise TorrentInjectionError("No injection link directory specified in the config file.")
 
-    if (not config.deluge_rpc_url) and (not config.qbittorrent_url):
+    if (not config.deluge_rpc_url) and (not config.transmission_rpc_url) and (not config.qbittorrent_url):
       raise TorrentInjectionError("No torrent client configuration specified in the config file.")
 
     return config
@@ -48,6 +49,8 @@ def __validate_config(config: Config):
   def __determine_torrent_client(config: Config):
     if config.deluge_rpc_url:
       return Deluge(config.deluge_rpc_url)
+    elif config.transmission_rpc_url:
+      return TransmissionBt(config.transmission_rpc_url)
     elif config.qbittorrent_url:
       return Qbittorrent(config.qbittorrent_url)