deluge_label_export.py

#!/usr/bin/env python3
import asyncio
import json
import os
from pathlib import Path
import random
import re
import shutil
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:8113/json"
deluge_password = "deluge"
export_label_only = True
export_label = "exporting_label"

# LEAVE THE R BEFORE THE QUOTATIONS
export_directory = r"C:\export"
state_directory = r"X:\deluge-develop\state"


### 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)
        filter_query = {}
        if export_label_only:
            filter_query = {"label": export_label}
        torrent_list = (
            (
                await deluge_handler.call(
                    "web.update_ui",
                    [
                        ["name", "tracker"],
                        filter_query,
                    ],
                )
            )
            .get("result", {})
            .get("torrents", {})
        )

        all_torrents_files = [
            entry.name
            for entry in os.scandir(state_directory)
            if entry.is_file() and ".torrent" in entry.name
        ]
        for key, value in torrent_list.items():
            if f"{key}.torrent" in all_torrents_files:
                infohash = torrent_list.get(key.replace(".torrent", ""))
                # labeling regexs
                movieregex = (
                    r"^(?P<title>.+?)[_.\s()\[\]]?(?P<year>\d{4})[)\]]?(?![pi])"
                )
                name = value.get("name", infohash)
                seasonregex = r"^(?P<title>.+?)[_.\s-]+(?P<season>S(eason\s)?\d+)(?:[_.\s-]*?(?P<seasonmax>S?\d+))?(?=[_.\s](?!E\d+))"
                epregex = r"^(?P<title>.+?)[_.\s-]+(?:(?P<season>S\d+)?[_.\s]?(?P<episode>E\d+(?:[\s-]?E?\d+)?(?![ip]))(?!\d+[ip])|(?P<date>(?P<year>\d{4})[_.\s-](?P<month>\d{2})[_.\s-](?P<day>\d{2})))"
                media_type = "unknown"
                season_match = re.match(seasonregex, name, re.IGNORECASE)
                if season_match:
                    media_type = "pack"
                else:
                    ep_match = re.match(epregex, name, re.IGNORECASE):
                    if ep_match:
                        media_type = "episode"
                    else:
                        movie_match = re.match(movieregex, name, re.IGNORECASE)
                        if movie_match:
                            media_type = "movie"
                host = urlparse(value.get("tracker", "error")).hostname
                new_torrent_name = f"[{media_type}][{host}]{name}"
                original_path = Path(state_directory) / f"{key}.torrent"
                dest = Path(export_directory) / f"{new_torrent_name}.torrent"
                print(f"Copying {str(original_path)} -> {str(dest)}")
                shutil.copy(original_path, dest)

    except Exception as e:
        print(f"\n\n[{CRED}error{CEND}]: {CBOLD}{e}{CEND}\n\n")
    print(
        f"\n\n[{CGREEN}exported{CEND}]: {CBLUE}{len(torrent_list)}{CEND}{CBOLD} torrents files\n\n"
    )
    await deluge_handler.call("auth.delete_session", [], 0)
    deluge_handler.session.close()
    # exit(0)


if __name__ == "__main__":

    asyncio.run(main())