update_deluge_announce_url.py

#!/usr/bin/env python3
import asyncio
import json
import random
import requests
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://localhost:8112/json"
deluge_password = "deluge"
old_announce = "https://www.google.com/announce/r435435435"
new_announce = "https://www.google.com/announce/gfd5465gfd"

### 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()
        self.disconnecting = False

    async 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
                await self.call("auth.login", [deluge_password], 0)

                if self.deluge_cookie:
                    return await 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


async def main():
    deluge_handler = DelugeHandler()

    try:
        # auth.login
        auth_response = await deluge_handler.call("auth.login", [deluge_password], 0)

        # checks the status of webui being connected, and connects to the daemon
        webui_connected = (await deluge_handler.call("web.connected", [], 0)).get(
            "result"
        )
        if not webui_connected:
            web_ui_daemons = await deluge_handler.call("web.get_hosts", [], 0)
            webui_connected = await deluge_handler.call(
                "web.connect", [web_ui_daemons.get("result")[0][0]], 0
            )
            if not webui_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)
        print(
            f"[{CGREEN}json-rpc{CEND}/{CYELLOW}auth.login{CEND}]",
            auth_response,
        )
        if not auth_response.get("result"):
            exit(1)
        torrent_list = (
            (
                await deluge_handler.call(
                    "web.update_ui",
                    [
                        ["name", "state", "progress", "total_remaining", "tracker"],
                        {"tracker_host": "Error"},
                    ],
                )
            )
            .get("result", {})
            .get("torrents", {})
        )
        filtered_trackers = []

        for key, inner_dict in torrent_list.items():
            # check if 'tracker' key value length is empty string
            # this signifies a tracker was never contacted
            check_ssl = (
                (
                    await deluge_handler.call(
                        "core.get_torrent_status",
                        [key, ["trackers"]],
                    )
                )
                .get("result", {})
                .get("trackers", {})[0]
                .get("url", {})
            )
            if check_ssl == old_announce:
                filtered_trackers.append(key)
                await deluge_handler.call(
                    "core.set_torrent_trackers",
                    [
                        key,
                        [
                            {"tier": 0, "url": new_announce},
                        ],
                    ],
                )

    except Exception as e:
        print(f"\n\n[{CRED}error{CEND}]: {CBOLD}{e}{CEND}\n\n")
    print(
        f"\n\n[{CGREEN}completed{CEND}]: {CBOLD}updated {CBLUE}{len(filtered_trackers)}{CEND}{CBOLD} torrent trackers\n\n"
    )
    await deluge_handler.call("auth.delete_session", [], 0)

    deluge_handler.session.close()
    exit(0)


if __name__ == "__main__":
    asyncio.run(main())