sane-scripts: port sane-bt-rm to python
also fix missing lib in sane-bt-add
This commit is contained in:
parent
b3a605c76b
commit
c977665214
|
@ -5,6 +5,14 @@
|
|||
}:
|
||||
|
||||
let
|
||||
# for a python script that needs the lib/ directory
|
||||
# TODO: would be better to package the lib directory as its own python library
|
||||
pythonWithLib = args: static-nix-shell.mkPython3Bin (args // {
|
||||
postInstall = args.postInstall or "" + ''
|
||||
mkdir -p $out/bin/lib
|
||||
cp -R lib/* $out/bin/lib/
|
||||
'';
|
||||
});
|
||||
nix-shell-scripts = {
|
||||
# anything added to this attrset gets symlink-joined into `sane-scripts`
|
||||
# and is made available through `sane-scripts.passthru`
|
||||
|
@ -18,12 +26,12 @@ let
|
|||
src = ./src;
|
||||
pkgs = [ "duplicity" ];
|
||||
};
|
||||
bt-add = static-nix-shell.mkPython3Bin {
|
||||
bt-add = pythonWithLib {
|
||||
pname = "sane-bt-add";
|
||||
src = ./src;
|
||||
pkgs = [ "transmission" ];
|
||||
};
|
||||
bt-rm = static-nix-shell.mkBash {
|
||||
bt-rm = pythonWithLib {
|
||||
pname = "sane-bt-rm";
|
||||
src = ./src;
|
||||
pkgs = [ "transmission" ];
|
||||
|
|
107
pkgs/additional/sane-scripts/src/lib/sane_torrent.py
Normal file
107
pkgs/additional/sane-scripts/src/lib/sane_torrent.py
Normal file
|
@ -0,0 +1,107 @@
|
|||
from __future__ import annotations
|
||||
|
||||
from argparse import ArgumentParser, Namespace
|
||||
from dataclasses import dataclass
|
||||
|
||||
import os.path
|
||||
import subprocess
|
||||
|
||||
@dataclass
|
||||
class MediaMeta:
|
||||
title: str | None
|
||||
prefix: str | None
|
||||
author: str | None
|
||||
type_: "film" | "show" | "book" | "audiobook" | "vn" # TODO: use enumeration
|
||||
|
||||
@classmethod
|
||||
def add_arguments(self, parser: ArgumentParser):
|
||||
parser.add_argument("--prefix", help="additional path component before anything implied by the other options (but after the base media dir")
|
||||
parser.add_argument("--film", action="store_true")
|
||||
parser.add_argument("--show", help="ShowTitle")
|
||||
parser.add_argument("--book", help="BookTitle")
|
||||
parser.add_argument("--audiobook", help="AudiobookTitle")
|
||||
parser.add_argument("--vn", help="VisualNovelTitle (for comics/manga)")
|
||||
parser.add_argument("--author", help="FirstnameLastname")
|
||||
|
||||
@classmethod
|
||||
def from_arguments(self, args: Namespace) -> Self:
|
||||
title = None
|
||||
type_ = None
|
||||
if args.film:
|
||||
type_ = "film"
|
||||
if args.show != None:
|
||||
type_ = "show"
|
||||
title = args.show
|
||||
if args.book != None:
|
||||
type_ = "book"
|
||||
title = args.book
|
||||
if args.audiobook != None:
|
||||
type_ = "audiobook"
|
||||
title = args.audiobook
|
||||
if args.vn != None:
|
||||
type_ = "vn"
|
||||
title = args.vn
|
||||
assert type_ is not None, "no torrent type specified!"
|
||||
return MediaMeta(
|
||||
title=title,
|
||||
prefix=args.prefix,
|
||||
author=args.author,
|
||||
type_=type_,
|
||||
)
|
||||
|
||||
@property
|
||||
def type_path(self) -> str:
|
||||
return dict(
|
||||
film="Videos/Film/",
|
||||
show="Videos/Shows/",
|
||||
book="Books/Books/",
|
||||
audiobook="Books/Audiobooks/",
|
||||
vn="Books/Visual/",
|
||||
)[self.type_]
|
||||
|
||||
def fs_path(self, base: str="/var/lib/uninsane/media/"):
|
||||
return os.path.join(base, self.prefix or "", self.type_path, self.author or "", self.title or "")
|
||||
|
||||
|
||||
def dry_check_call(args: list[str]):
|
||||
print("not invoking because dry run: " + ' '.join(args))
|
||||
|
||||
class TransmissionApi:
|
||||
ENDPOINT="https://bt.uninsane.org/transmission/rpc"
|
||||
PASSFILE="/run/secrets/transmission_passwd"
|
||||
|
||||
def __init__(self, check_call = subprocess.check_call):
|
||||
self.check_call = check_call
|
||||
|
||||
@staticmethod
|
||||
def add_arguments(parser: ArgumentParser):
|
||||
parser.add_argument("--dry-run", action="store_true", help="only show what would be done; don't invoke transmission")
|
||||
|
||||
@staticmethod
|
||||
def from_arguments(args: Namespace) -> Self:
|
||||
return TransmissionApi(check_call = dry_check_call if args.dry_run else subprocess.check_call)
|
||||
|
||||
@property
|
||||
def auth(self) -> str:
|
||||
return open(self.PASSFILE, "r").read().strip()
|
||||
|
||||
def add_torrent(self, meta: MediaMeta, torrent: str):
|
||||
print(f"saving to {meta.fs_path()}")
|
||||
self.call_transmission([
|
||||
"--download-dir", meta.fs_path(),
|
||||
"--add", torrent,
|
||||
])
|
||||
|
||||
def rm_torrent(self, torrent: str | int):
|
||||
print(f"deleting {torrent}")
|
||||
self.call_transmission([
|
||||
"--torrent", torrent,
|
||||
"--remove-and-delete"
|
||||
])
|
||||
|
||||
def call_transmission(self, args: list[str]):
|
||||
self.check_call([
|
||||
"transmission-remote",
|
||||
self.ENDPOINT,
|
||||
"--auth", f"colin:{self.auth}",
|
||||
] + args)
|
|
@ -3,52 +3,23 @@
|
|||
# vim: set filetype=python :
|
||||
|
||||
import argparse
|
||||
import subprocess
|
||||
import sys
|
||||
|
||||
sys.path.insert(0, ".") # to import `lib`
|
||||
|
||||
from lib.sane_torrent import MediaMeta
|
||||
from lib.sane_torrent import MediaMeta, TransmissionApi
|
||||
|
||||
def dry_check_call(args: list[str]):
|
||||
print("not invoking because dry run: " + ' '.join(args))
|
||||
|
||||
class Executor:
|
||||
ENDPOINT="https://bt.uninsane.org/transmission/rpc"
|
||||
PASSFILE="/run/secrets/transmission_passwd"
|
||||
|
||||
def __init__(self, check_call = subprocess.check_call):
|
||||
self.check_call = check_call
|
||||
|
||||
@property
|
||||
def auth(self) -> str:
|
||||
return open(self.PASSFILE, "r").read()
|
||||
|
||||
def add_torrent(self, meta: MediaMeta, torrent: str):
|
||||
print(f"saving to {meta.fs_path()}")
|
||||
self.call_transmission([
|
||||
"--download-dir", meta.fs_path(),
|
||||
"--add", torrent,
|
||||
])
|
||||
|
||||
def call_transmission(self, args: list[str]):
|
||||
self.check_call([
|
||||
"transmission-remote",
|
||||
self.ENDPOINT,
|
||||
"--auth", f"colin:{self.auth}",
|
||||
] + args)
|
||||
|
||||
def main():
|
||||
parser = argparse.ArgumentParser()
|
||||
TransmissionApi.add_arguments(parser)
|
||||
MediaMeta.add_arguments(parser)
|
||||
parser.add_argument("--dry-run", action="store_true", help="only show what would be done; don't invoke transmission")
|
||||
parser.add_argument("torrent", help="magnet: URI or file path to torrent")
|
||||
|
||||
args = parser.parse_args()
|
||||
meta = MediaMeta.from_arguments(args)
|
||||
dry_run = args.dry_run
|
||||
executor = TransmissionApi.from_arguments(args)
|
||||
torrent = args.torrent
|
||||
executor = Executor(check_call = dry_check_call if dry_run else subprocess.check_call)
|
||||
|
||||
executor.add_torrent(meta, torrent)
|
||||
|
||||
|
|
|
@ -1,11 +1,27 @@
|
|||
#!/usr/bin/env nix-shell
|
||||
#!nix-shell -i bash -p transmission
|
||||
#!nix-shell -i python3 -p "python3.withPackages (ps: [ ])" -p transmission
|
||||
# vim: set filetype=python :
|
||||
|
||||
# removes a torrent and trashes its data
|
||||
# usage: sane-bt-rm <torrent>
|
||||
# where <torrent> is a magnet URL, or an identifier from sane-bt-show (e.g. 132)
|
||||
|
||||
endpoint=https://bt.uninsane.org/transmission/rpc
|
||||
PASS=$(sudo cat /run/secrets/transmission_passwd)
|
||||
import argparse
|
||||
import sys
|
||||
|
||||
transmission-remote "$endpoint" --auth "colin:$PASS" --torrent "$1" --remove-and-delete
|
||||
sys.path.insert(0, ".") # to import `lib`
|
||||
|
||||
from lib.sane_torrent import TransmissionApi
|
||||
|
||||
|
||||
def main():
|
||||
parser = argparse.ArgumentParser()
|
||||
TransmissionApi.add_arguments(parser)
|
||||
parser.add_argument("torrent", help="numerical ID of the torrent in the transmission list (see sane-bt-show)")
|
||||
|
||||
args = parser.parse_args()
|
||||
executor = TransmissionApi.from_arguments(args)
|
||||
torrent = args.torrent
|
||||
|
||||
executor.rm_torrent(torrent)
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
|
|
Loading…
Reference in New Issue
Block a user