transbeam/cli/transbeam-cli

70 lines
2.8 KiB
Python
Executable file

#!/usr/bin/env python3
import argparse
import asyncio
import getpass
import json
import math
import pathlib
import sys
from tqdm import tqdm
from websockets import connect
def file_loader(files):
with tqdm(desc="Total", total=sum(size for (path, size) in files), unit='B', unit_scale=True, leave=True, position=1) as total_progress:
for (path, size) in files:
with tqdm(desc=path.name, total=size, unit='B', unit_scale=True, leave=True, position=0) as file_progress:
with path.open("rb") as f:
while f.tell() < size:
data = f.read(min(16384, size - f.tell()))
if data == "":
tqdm.write("file ended early!")
exit(1)
total_progress.update(len(data))
file_progress.update(len(data))
yield data
async def send(paths, host, password, lifetime, collection_name=None):
paths = [path for path in paths if path.is_file()]
fileMetadata = [
{
"name": path.name,
"size": path.stat().st_size,
"modtime": math.floor(path.stat().st_mtime * 1000),
} for path in paths
]
manifest = {
"files": fileMetadata,
"lifetime": lifetime,
"password": password,
}
if collection_name is not None:
manifest["collection_name"] = collection_name
async with connect("wss://{}/upload".format(host)) as ws:
await ws.send(json.dumps(manifest))
resp = json.loads(await ws.recv())
if resp["type"] != "ready":
print("unexpected response: {}".format(resp))
exit(1)
print("Download: https://{}/download?code={}".format(host, resp["code"]))
loader = file_loader([(paths[i], fileMetadata[i]["size"]) for i in range(len(paths))])
for data in loader:
await ws.send(data)
parser = argparse.ArgumentParser(description="Upload files to transbeam")
parser.add_argument("-l", "--lifetime", type=int, default=7, help="Lifetime in days for files (default 7)")
parser.add_argument("-H", "--host", type=str, default="transbeam.link", help="transbeam host (default transbeam.link)")
parser.add_argument("-n", "--collection-name", type=str, help="Name for a collection of multiple files")
parser.add_argument("files", type=pathlib.Path, nargs="+", help="Files to upload")
async def main():
args = parser.parse_args()
if len(args.files) == 1 and args.collection_name is not None:
print("--collection-name is only applicable when multiple files are being uploaded")
exit(1)
password = getpass.getpass()
await send(args.files, args.host, password, args.lifetime, args.collection_name)
asyncio.run(main())