82 lines
3.0 KiB
Python
Executable File
82 lines
3.0 KiB
Python
Executable File
#!/usr/bin/env nix-shell
|
|
#!nix-shell -i python3 -p "python3.withPackages (ps: [ ps.requests ps.sane-lib.bt ])" -p sane-scripts.lib.bt.propagatedBuildInputs
|
|
# vim: set filetype=python :
|
|
|
|
import argparse
|
|
import logging
|
|
import requests
|
|
import tempfile
|
|
|
|
from requests.exceptions import InvalidSchema
|
|
|
|
from sane_bt import MediaMeta, TransmissionApi
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
# TODO: handle 'InvalidSchema's when kat.ph redirects directly to the magnet:
|
|
#
|
|
# class MagnetAdapter:
|
|
# def send(self, req, **kwargs):
|
|
# resp = requests.Response()
|
|
# resp.history = [req.url]
|
|
# return resp
|
|
# s = requests.session()
|
|
# s.mount('magnet:', MagnetAdapter())
|
|
# resp = s.request('GET', uri)
|
|
# resp.history[-1] # contains the magnet: URI, if everything's proper
|
|
# except i think history gets overwritten, so may need to assign a different field
|
|
|
|
def resolve_torrent(uri: str) -> str:
|
|
"""
|
|
given a URI, coerce it into something native to transmission
|
|
i.e. magnet: or .torrent.
|
|
"""
|
|
if uri.startswith("http://") or uri.startswith("https://"):
|
|
logger.info(f"downloading remote torrent data: {uri}")
|
|
try:
|
|
response = requests.get(uri)
|
|
except InvalidSchema as e:
|
|
# sometimes the URI resolves to a magnet: link (popular with KAT)
|
|
# that's *FINE*, unfortunately `requests` library gives us no way to access that link except by parsing the error message.
|
|
# alternatively i could install some special requests adapter that handles the magnet: schema.
|
|
logger.debug(f"URL resolution error: {e}")
|
|
if len(e.args) == 1 and "magnet:" in e.args[0]:
|
|
uri = e.args[0].replace("No connection adapters were found for ", "")
|
|
uri = uri.strip("'")
|
|
if not uri.startswith("magnet:"): raise
|
|
logger.info(f"resolved torrent to {uri}")
|
|
else:
|
|
raise
|
|
else:
|
|
# N.B.: without `delete=False`, file gets deleted when the function returns.
|
|
# could work around this, but it's nice to keep the file around anyway for troubleshooting
|
|
f = tempfile.NamedTemporaryFile(suffix=".torrent", delete=False)
|
|
f.write(response.content)
|
|
uri = f.name
|
|
|
|
if not (uri.startswith("magnet:") or uri.endswith(".torrent")):
|
|
logger.warn(f"unknown type for torrent URI: {uri}")
|
|
|
|
return uri
|
|
|
|
|
|
def main():
|
|
logging.basicConfig()
|
|
logging.getLogger().setLevel(logging.INFO)
|
|
|
|
parser = argparse.ArgumentParser(description="instruct servo to download some torrent")
|
|
TransmissionApi.add_arguments(parser)
|
|
MediaMeta.add_arguments(parser)
|
|
parser.add_argument("torrent", help="file path or URI to torrent (magnet, https)")
|
|
|
|
args = parser.parse_args()
|
|
meta = MediaMeta.from_arguments(args)
|
|
bt_api = TransmissionApi.from_arguments(args)
|
|
torrent = args.torrent.strip()
|
|
|
|
torrent = resolve_torrent(torrent)
|
|
bt_api.add_or_move_torrent(meta, torrent)
|
|
|
|
if __name__ == "__main__":
|
|
main()
|