transmission: Add transmission CLI to start downloading torrents
This commit is contained in:
Executable
+64
@@ -0,0 +1,64 @@
|
||||
#!/usr/bin/env -S uv run
|
||||
# /// script
|
||||
# dependencies = [
|
||||
# "httpx",
|
||||
# ]
|
||||
# ///
|
||||
import argparse
|
||||
import base64
|
||||
import logging
|
||||
from os import getenv
|
||||
import httpx
|
||||
from pathlib import Path
|
||||
|
||||
|
||||
def add_torrent(client: httpx.Client, torrent_path: Path):
|
||||
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()
|
||||
|
||||
return result["result"] in ["duplicate-torrent", "success"]
|
||||
|
||||
|
||||
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")
|
||||
|
||||
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=10,
|
||||
)
|
||||
csrf_token = client.post("/transmission/rpc", json={"method": "session-get"}).headers["x-transmission-session-id"]
|
||||
client.headers["X-Transmission-Session-Id"] = csrf_token
|
||||
|
||||
for torrent_path in args.torrent_paths:
|
||||
if not torrent_path.is_file():
|
||||
logging.error(f"Torrent file not found: {args.torrent_path}")
|
||||
raise SystemError(1)
|
||||
add_torrent(client=client, torrent_path=torrent_path)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
Reference in New Issue
Block a user