51 lines
1.5 KiB
Python
51 lines
1.5 KiB
Python
|
import logging
|
||
|
|
from pathlib import Path
|
||
|
|
import time
|
||
|
|
import tempfile
|
||
|
|
from typing import Optional
|
||
|
|
|
||
|
|
import libtorrent
|
||
|
|
|
||
|
|
|
||
|
|
def magnet2torrent(magnet: str, output: Path):
|
||
|
|
log = logging.getLogger()
|
||
|
|
|
||
|
|
if not output.parent.is_dir():
|
||
|
|
raise Exception("Invalid output file: directory does not exist")
|
||
|
|
|
||
|
|
with tempfile.TemporaryDirectory() as tmp_dir:
|
||
|
|
session = libtorrent.session()
|
||
|
|
#params = {
|
||
|
|
# "save_path": tmp_dir,
|
||
|
|
# "storage_mode": libtorrent.storage_mode_t(2),
|
||
|
|
# "paused": False,
|
||
|
|
# "auto_managed": True,
|
||
|
|
# "duplicate_is_error": True,
|
||
|
|
#}
|
||
|
|
#handle = libtorrent.add_magnet_uri(session, magnet, params)
|
||
|
|
|
||
|
|
params = libtorrent.parse_magnet_uri(magnet)
|
||
|
|
handle = session.add_torrent(params)
|
||
|
|
log.info("Downloading metadata")
|
||
|
|
while not handle.has_metadata():
|
||
|
|
try:
|
||
|
|
time.sleep(1)
|
||
|
|
except KeyboardInterrupt:
|
||
|
|
log.info("Cleaning up")
|
||
|
|
session.pause()
|
||
|
|
raise
|
||
|
|
session.pause()
|
||
|
|
log.info("Done")
|
||
|
|
|
||
|
|
info = handle.get_torrent_info()
|
||
|
|
torrent_file = libtorrent.create_torrent(info)
|
||
|
|
torrent_path = Path(info.name() + ".torrent").absolute()
|
||
|
|
|
||
|
|
if output.is_dir():
|
||
|
|
output = output / torrent_path.name
|
||
|
|
|
||
|
|
log.info("Writing to %s", output)
|
||
|
|
content = libtorrent.bencode(torrent_file.generate())
|
||
|
|
output.write_bytes(content)
|
||
|
|
session.remove_torrent(handle)
|