Custom Import Script for importing to Radarr via qBittorrent's setLocation API instead of a simple copy

radarr-qbit-move.py

#!/usr/bin/env python3

# about:
# Instead of copying from a qBittorrent download into your Radarr movie folder,
# this just tells qBittorrent to move the torrent to where Radarr would copy it
# to. This prevents duplicate files on setups that do not support hardlinks.
# I.e. almost every good windows setup that has 2 or more drives. E.g. chads who
# use StableBit DrivePool.

# note:
# I initially set this up to use Radarr's API to grab the host/user/pass/etc, but
# that fell through when I realized the API doesn't return the password, so it's
# auto-usefulness dried up immediately, so I just went with env for the details.

# support:
# - Radarr: qBittorrent, SABnzbd
# - Sonarr: Not yet

import os
from pathlib import Path
import sys
from typing import Literal
import requests

# --- qBittorrent Web UI Settings ---
url = os.environ.get("QBITTORRENT_URL", "http://localhost:")
username = os.environ.get("QBITTORRENT_USERNAME", "")
password = os.environ.get("QBITTORRENT_PASSWORD", "")
debug = False  # Set to True to log env vars for debugging purposes

# DO NOT EDIT ANYTHING FURTHER
client_type = os.environ.get("RADARR_DOWNLOAD_CLIENT_TYPE")  # e.g. qBittorrent or SABnzbd
torrent_hash = os.environ.get("RADARR_DOWNLOAD_ID")          # e.g. 1234567890abcdef1234567890abcdef12345678
movie_folder = os.environ.get("RADARR_MOVIE_PATH")           # e.g. D:\movies\Coraline (2009)
source_path = os.environ.get("RADARR_SOURCEPATH")            # e.g. D:\downloads\complete\Coraline (2009).mkv
destination_path = os.environ.get("RADARR_DESTINATIONPATH")  # e.g. D:\movies\Coraline (2009)\Coraline (2009).mkv

class qBittorrent:
    def __init__(self):
        self.session = requests.Session()
        self.url: str = None
        self.authenticated: bool = False

    def __enter__(self):
        return self

    def __exit__(self, *_):
        self.logout()

    def login(self, url: str, username: str, password: str) -> None:
        """
        Logs into the qBittorrent Web UI.
        
        Status codes:
        - 403: User's IP is banned for too many failed login attempts
        - 200: All other scenarios
        """
        if self.authenticated:
            raise EnvironmentError(f"Already authenticated with: {self.url}")

        res = self.session.post(
            f"{url}/api/v2/auth/login",
            data={
                "username": username,
                "password": password
            },
            headers={
                "Referer": url.rstrip("/")
            }
        )

        if "SID" in res.cookies:
            self.url = url
            self.authenticated = True
        elif res.status_code == 403:
            raise EnvironmentError("User's IP is banned for too many failed login attempts.")
        else:
            raise Exception(f"Login failed: {res.text}")

    def logout(self) -> None:
        """
        Logs out of the qBittorrent Web UI.
        
        Status codes:
        - 200: All scenarios
        """
        if not self.authenticated:
            return

        res = self.session.post(
            f"{self.url}/api/v2/auth/logout"
        )
        if not res.ok:
            raise Exception(f"Could not logout: ", res.text)

        self.url = None
        self.authenticated = False

    def set_location(self, hashes: list[str] | Literal["all"], location: os.PathLike) -> int:
        """
        Sets the location for the specified torrents.
        If the location doesn't exist, the torrent's location is unchanged.

        Status codes:
        - 400: Save path is empty
        - 403: User does not have write access to directory
        - 409: Unable to create save path directory
        - 200: All other scenarios
        """
        if not self.authenticated:
            raise EnvironmentError("Not authenticated. Please login first.")

        res = self.session.post(
            f"{self.url}/api/v2/torrents/setLocation",
            data={
                "hashes": "|".join(hashes),
                "location": location
            }
        )

        return res.status_code

    def set_category(self, hashes: list[str] | Literal["all"], category: str) -> int:
        """
        Sets the category for the specified torrents.

        Status codes:
        - 409: Category name does not exist
        - 200: All other scenarios
        """
        if not self.authenticated:
            raise EnvironmentError("Not authenticated. Please login first.")

        res = self.session.post(
            f"{self.url}/api/v2/torrents/setCategory",
            data={
                "hashes": "|".join(hashes),
                "category": category
            }
        )

        return res.status_code

def main():
    if debug:
        radarr_envs = {k: v for k, v in os.environ.items() if k.startswith("RADARR_")}
        env_log_file = Path(__file__).with_name("radarr_envs.txt")
        with open(env_log_file, "a", encoding="utf-8") as f:
            for k, v in radarr_envs.items():
                f.write(f"{k}={v}\n")
            f.write("---------------------\n")

    if not client_type:
        print("Missing environment variable: RADARR_DOWNLOAD_CLIENT_TYPE")
        sys.exit(2)

    if client_type == "qBittorrent":
        if not torrent_hash or not movie_folder:
            print("Missing environment variables: RADARR_DOWNLOAD_ID and/or RADARR_MOVIE_PATH")
            sys.exit(3)
        with qBittorrent() as qb:
            qb.login(url, username, password)
            res = qb.set_location([torrent_hash], movie_folder)
            if res == 200:
                print(f"qBittorrent Successfully moved torrent {torrent_hash} to {movie_folder}")
            else:
                print(f"qBittorrent Failed to move torrent: {res}")
                sys.exit(4)
    elif client_type == "SABnzbd":
        if not source_path or not destination_path:
            print("Missing environment variables: RADARR_SOURCEPATH and/or RADARR_DESTINATIONPATH")
            sys.exit(5)
        source = Path(source_path)
        destination = Path(destination_path)
        try:
            destination.parent.mkdir(parents=True, exist_ok=True)
            res = source.replace(destination)
            if res != destination:
                print(f"Failed to move file: {source_path} to {destination_path}")
                sys.exit(6)
            else:
                print(f"Moved file from {source_path} to {destination_path}")
        except Exception as e:
            print(f"Failed to move file: {e}")
            sys.exit(7)
    else:
        # we cannot just sys.exit(0) or *arr will just delete the download and not error/warn
        print(f"Unsupported download client: {client_type}. Only SABnzbd & qBittorrent is supported.")
        sys.exit(8)

if __name__ == "__main__":
    main()