reconnect.py

#!/usr/bin/env python3
import json
import random
import requests
import time
from enum import Enum
from urllib.parse import urlparse

### CONFIGURATION VARIABLES ###

# this webui will need to be the JSON-RPC endpoint
# this ends with '/json'
deluge_webui = "http://deluge:8112/json"
deluge_password = "deluged"

### STOP EDITING HERE ###
### STOP EDITING HERE ###
### STOP EDITING HERE ###
### STOP EDITING HERE ###


# error codes we could potentially receive
class DelugeErrorCode(Enum):
    NO_AUTH = 1
    BAD_METHOD = 2
    CALL_ERR = 3
    RPC_FAIL = 4
    BAD_JSON = 5


# color codes for terminal
use_colors_codes = False

CRED = "\033[91m" if (use_colors_codes) else ""
CGREEN = "\33[32m" if (use_colors_codes) else ""
CYELLOW = "\33[33m" if (use_colors_codes) else ""
CBLUE = "\33[4;34m" if (use_colors_codes) else ""
CBOLD = "\33[1m" if (use_colors_codes) else ""
CEND = "\033[0m" if (use_colors_codes) else ""


class DelugeHandler:
    def __init__(self):
        self.deluge_cookie = None
        self.session = requests.Session()

    def call(self, method, params, retries=1):
        url = urlparse(deluge_webui).geturl()
        headers = {"Content-Type": "application/json"}
        id = random.randint(0, 0x7FFFFFFF)

        # set our cookie if we have it
        if self.deluge_cookie:
            headers["Cookie"] = self.deluge_cookie

        if method == "auth.login":
            print(
                f"[{CGREEN}init{CEND}/{CYELLOW}script{CEND}] -> {CYELLOW}Connecting to Deluge:{CEND} {CBLUE}{url}{CEND}"
            )

        # send our request to the JSON-RPC endpoint
        try:
            response = self.session.post(
                url,
                data=json.dumps({"method": method, "params": params, "id": id}),
                headers=headers,
            )
            response.raise_for_status()
        except requests.exceptions.RequestException as network_error:
            raise ConnectionError(
                f"[{CRED}json-rpc{CEND}/{CRED}error{CEND}]: Failed to connect to Deluge at {CBLUE}{url}{CEND}"
            ) from network_error

        # make sure the json response is valid
        try:
            json_response = response.json()
        except json.JSONDecodeError as json_parse_error:
            raise ValueError(
                f"[{CRED}json-rpc{CEND}/{CRED}error{CEND}]: Deluge method {method} response was {CYELLOW}non-JSON{CEND}: {json_parse_error}"
            )

        # check for authorization failures, and retry once
        if json_response.get("error", [None]) != None:
            if (
                json_response.get("error", [None]).get("code")
                == DelugeErrorCode.NO_AUTH
                and retries > 0
            ):
                self.deluge_cookie = None
                self.call("auth.login", [deluge_password], 0)

                if self.deluge_cookie:
                    return self.call(method, params)
                else:
                    raise ConnectionError(
                        f"[{CRED}json-rpc{CEND}/{CRED}error{CEND}]: Connection lost with Deluge. Reauthentication {CYELLOW}failed{CEND}."
                    )

        self.handle_cookies(response.headers)
        return json_response

    def handle_cookies(self, headers):
        deluge_cookie = headers.get("Set-Cookie")
        if deluge_cookie:
            deluge_cookie = deluge_cookie.split(";")[0]
        else:
            deluge_cookie = None


def main():
    deluge_handler = DelugeHandler()

    try:
        # auth.login
        auth_response = deluge_handler.call("auth.login", [deluge_password], 0)
        print(
            f"[{CGREEN}json-rpc{CEND}/{CYELLOW}auth.login{CEND}]",
            auth_response,
            "\n",
        )
        if auth_response.get("result") != True:
            exit(1)
        webui_connected = deluge_handler.call("web.connected", [], 0)
        print(f"[json-rpc/web.connected] {webui_connected}")

        time.sleep(2)
        # get hosts list
        web_ui_daemons = deluge_handler.call("web.get_hosts", [], 0).get("result")
        # check which host is connected
        for daemon in web_ui_daemons:
            webui_connected_host = daemon[0]
            webui_connected = deluge_handler.call(
                "web.get_host_status", [webui_connected_host], 0
            ).get("result")
            if webui_connected[1] == "Connected":
                # reconnect the web daemon to the previously connected host
                web_disconnect = deluge_handler.call("web.disconnect", [], 0)
                webui_connected = deluge_handler.call(
                    "web.get_host_status", [webui_connected_host], 0
                ).get("result")
                print(f"[json-rpc/web.disconnect] {web_disconnect}")
                break

        # checks the status of webui being connected, and connects to the daemon
        webui_connected = webui_connected[1]
        if webui_connected == "Online":
            deluge_handler.call("web.connect", [webui_connected_host], 0)
            time.sleep(1)
            webui_connected = deluge_handler.call(
                "web.get_host_status", [webui_connected_host], 0
            ).get("result")
            if webui_connected[1] != "Connected":
                print(
                    f"\n\n[{CRED}error{CEND}]: {CYELLOW}Your WebUI is not automatically connectable to the Deluge daemon.{CEND}\n"
                    f"{CYELLOW}\t Open the WebUI's connection manager to resolve this.{CEND}\n\n"
                )
                exit(1)
            else:
                print(f"[json-rpc/web.connect] Successfully reconnected to daemon.\n")

    except Exception as e:
        print(f"\n\n[{CRED}error{CEND}]: {CBOLD}{e}{CEND}\n\n")
    deluge_handler.call("auth.delete_session", [], 0)
    deluge_handler.session.close()
    exit(0)


if __name__ == "__main__":
    main()