playground/transmission.py

#!/usr/bin/env -S uv run
# /// script
# dependencies = [
#     "httpx",
# ]
# ///
import argparse
import base64
import logging
from os import getenv
import os
import sys
import subprocess
import time
import httpx
from pathlib import Path


def move_to_trash(file_path: Path):
    if not file_path.is_file():
        raise FileNotFoundError

    if sys.platform == "darwin":
        args = [
            "osascript",
            "-e",
            f'tell app "Finder" to move POSIX file "{file_path}" to trash',
        ]
        return subprocess.run(
            args,
            check=True,
            capture_output=True,
        )
    if sys.platform == "win32":
        args = [
            "powershell",
            "-NoProfile",
            "-Command",
            f'Add-Type -AssemblyName Microsoft.VisualBasic; [Microsoft.VisualBasic.FileIO.FileSystem]::DeleteFile("{file_path}", "OnlyErrorDialogs", "SendToRecycleBin")',
        ]
        return subprocess.run(
            args,
            check=True,
        )

    raise NotImplementedError("Unsupported platform")


def add_torrent(client: httpx.Client, torrent_path: Path) -> str | None:
    encoded_torrent = base64.b64encode(torrent_path.read_bytes()).decode()

    payload = {
        "method": "torrent-add",
        "arguments": {
            "download-dir": "/downloads/",
            "metainfo": encoded_torrent,
            "paused": False,
        },
    }

    res = client.post("/transmission/rpc", json=payload)
    if res.is_error:
        raise Exception(f"HTTP {res.status_code}: {res.text}")
    result = res.json()

    if result["result"] not in ["duplicate-torrent", "success"]:
        return None
    added = result["arguments"].get("torrent-added") or result["arguments"].get("torrent-duplicate")
    return added["hashString"]


def print_download_paths(client: httpx.Client, hashes: list[str]):
    payload = {
        "method": "torrent-get",
        "arguments": {
            "ids": hashes,
            "fields": ["hashString", "name", "downloadDir"],
        },
    }
    res = client.post("/transmission/rpc", json=payload)
    if res.is_error:
        raise Exception(f"HTTP {res.status_code}: {res.text}")
    for t in res.json()["arguments"]["torrents"]:
        print(str(Path(t["downloadDir"]) / t["name"]))


def poll_until_done(client: httpx.Client, hashes: list[str], interval: int):
    pending = set(hashes)
    while pending:
        payload = {
            "method": "torrent-get",
            "arguments": {
                "ids": list(pending),
                "fields": ["hashString", "name", "percentDone", "isFinished", "status"],
            },
        }
        res = client.post("/transmission/rpc", json=payload)
        if res.is_error:
            raise Exception(f"HTTP {res.status_code}: {res.text}")
        torrents = res.json()["arguments"]["torrents"]
        for t in torrents:
            logging.info(f"{t['name']}: {t['percentDone'] * 100:.1f}%")
            if t["isFinished"] or t["status"] == 6:
                logging.info(f"Done: {t['name']}")
                pending.discard(t["hashString"])
        if pending:
            time.sleep(interval)


def parse_args():
    parser = argparse.ArgumentParser(description="Add torrent to Transmission via RPC")
    parser.add_argument("torrent_paths", type=Path, nargs="+", help="Path to the torrent file")
    parser.add_argument("--host", required=True, default=getenv("TRANSMISSION_HOST", "http://localhost:9091"), help="Transmission RPC host URL")
    parser.add_argument("--auth", default=getenv("TRANSMISSION_AUTH"), help="Transmission RPC username:password")
    parser.add_argument("--clean", action="store_true", help="Delete torrent file after successful upload")
    parser.add_argument("--wait", action="store_true", help="Block until all added torrents finish downloading")
    parser.add_argument("--interval", type=int, default=5, metavar="SECS", help="Polling interval in seconds when using --wait (default: 5)")
    parser.add_argument("--print-paths", action="store_true", help="Print download paths after adding (one per line)")

    return parser.parse_args()


def main():
    args = parse_args()
    logging.basicConfig(level=logging.INFO)

    client = httpx.Client(
        base_url=args.host,
        auth=httpx.BasicAuth(*args.auth.split(":")) if args.auth else None,
        timeout=20,
    )
    csrf_token = client.post("/transmission/rpc", json={"method": "session-get"}).headers["x-transmission-session-id"]
    client.headers["X-Transmission-Session-Id"] = csrf_token

    hashes = []
    for torrent_path in args.torrent_paths:
        if not torrent_path.is_file():
            logging.error(f"Torrent file not found: {torrent_path}")
            raise SystemError(1)
        hash_string = add_torrent(client=client, torrent_path=torrent_path)
        if hash_string:
            hashes.append(hash_string)
        if args.clean:
            move_to_trash(torrent_path)

    if args.wait and hashes:
        poll_until_done(client=client, hashes=hashes, interval=args.interval)

    if args.print_paths and hashes:
        print_download_paths(client=client, hashes=hashes)


if __name__ == "__main__":
    main()