1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
|
"""
`put`: upload files, or folders with -R, into a folder of a group.
**Decided whole before anything is sent**, as a drop on the Files tab is
(docs/MESHBAY_DESIGN.md §6.4): a name already in the destination — compared
without case — or one the node does not accept refuses the run and lists every
offender, so a folder never arrives with holes in it. `-n` skips the names
already there instead. There is no overwrite: the node never replaces a file,
and if one appears while an upload runs, the upload takes a free name and the
CLI says which.
"""
import asyncio
import contextlib
import os
import sys
import tempfile
import threading
from pathlib import Path
from ..client.groups import my_groups
from ..client.paths import handle, parse, resolve_group
from ..client.pool import Pool
from ..client.transfers import TransferError, create_dir, upload
from ..client.tree import Tree
from ..client.upload_plan import UploadFile, plan_upload, refused_names, taken_names
from ..errors import NOT_FOUND, PARTIAL, REFUSED, USAGE, CliError
from ..output import Progress, err, human_size, safe, sentence
from .context import Context
HELP = "upload files, or folders with -R, into a folder of a group"
PER_GROUP = 2
def add_parser(sub, parent) -> None:
p = sub.add_parser("put", parents=[parent], help=HELP, description=sentence(HELP)
+ ". With - as the source, stdin is sent and DEST names the file.")
p.add_argument("sources", nargs="+", metavar="SOURCE",
help="local files or folders, or - for stdin")
p.add_argument("dest", metavar="DEST",
help="the destination folder: group[@owner]/root/folder; "
"with - as the source, the file to write")
p.add_argument("-R", dest="recursive", action="store_true", help="folders, recursively")
p.add_argument("-n", dest="skip", action="store_true",
help="skip names already in the destination instead of refusing")
async def run(ctx: Context) -> int:
a = ctx.args
sources, dest_text = a.sources, a.dest
remote = parse(dest_text)
if remote is None or not remote.parts:
raise CliError(USAGE, "The destination is a folder of a group: "
"group[@owner]/root/folder")
from_stdin = sources == ["-"]
if "-" in sources and not from_stdin:
raise CliError(USAGE, "- (stdin) is the only source when it is used.")
await ctx.hub.check_version()
account = ctx.account()
pool = Pool(account)
spool = None
try:
group = resolve_group(await my_groups(account), remote.group)
session = await pool.session(group)
tree = Tree.from_index(await session.index())
parts = remote.parts[:-1] if from_stdin else remote.parts
folder = "/".join(parts)
_check_folder(tree, group, folder)
if from_stdin:
spool = await asyncio.to_thread(_spool_stdin)
name = remote.parts[-1]
plan_dirs: list[list[str]] = []
files = [UploadFile(Path(spool.name), [name], os.path.getsize(spool.name))]
problems = refused_names([name], [])
problems += [f"{folder}/{t} is already there"
for t in sorted(taken_names(tree, folder, [name]))]
else:
plan_dirs, files, problems = plan_upload(sources, a.recursive)
tops = sorted({tuple(d[:1]) for d in plan_dirs} | {tuple(f.rel[:1]) for f in files})
taken = taken_names(tree, folder, [t[0] for t in tops])
if taken and a.skip:
plan_dirs = [d for d in plan_dirs if d[0] not in taken]
files = [f for f in files if f.rel[0] not in taken]
elif taken:
problems += [f"{folder}/{t} is already there (-n to skip it)"
for t in sorted(taken)]
problems += refused_names([f.rel[-1] for f in files], plan_dirs)
if problems:
raise CliError(REFUSED, "Nothing was sent:\n "
+ "\n ".join(safe(p) for p in problems))
return await _send(ctx, session, group, folder, plan_dirs, files,
display_names=[remote.parts[-1]] if from_stdin else None)
finally:
if spool is not None:
with contextlib.suppress(OSError):
os.unlink(spool.name)
await pool.close()
await account.close()
def _check_folder(tree: Tree, group: dict, folder: str) -> None:
shown = f"{handle(group)}/{folder}"
if not tree.is_dir(folder):
raise CliError(NOT_FOUND, f"No folder {safe(shown)}.")
root = tree.roots.get(folder.split("/", 1)[0], {})
if not root.get("available", True):
raise CliError(REFUSED, f"{safe(shown)} is unavailable on the node "
"(its drive is disconnected).")
if not root.get("writable"):
raise CliError(REFUSED, f"{safe(shown)} is read-only: its operator does not "
"accept uploads there.")
def _spool_stdin():
"""stdin to a file first: every chunk names the total, so the size comes first."""
f = tempfile.NamedTemporaryFile(prefix="meshbay-cli-", delete=False)
with f:
while block := sys.stdin.buffer.read(1024 * 1024):
f.write(block)
return f
async def _send(ctx: Context, session, group: dict, folder: str, dirs: list[list[str]],
files: list[UploadFile], display_names: list[str] | None) -> int:
a = ctx.args
progress = Progress("↑", quiet=a.quiet, as_json=a.json)
progress.plan(len(files), sum(f.size for f in files))
# Folders first, parents before children, one at a time.
remote_dir: dict[tuple, str] = {(): folder}
try:
for d in dirs:
remote_dir[tuple(d)] = await create_dir(session, remote_dir[tuple(d[:-1])], d[-1])
except TransferError as e:
progress.clear()
created = len(remote_dir) - 1
raise CliError(PARTIAL if created else e.code,
f"Stopped: {e.message}"
+ (f" ({created} folder{'s' if created != 1 else ''} created)"
if created else "")) from e
failed = 0
stop: TransferError | None = None
gate = asyncio.Semaphore(PER_GROUP)
async def one(job: UploadFile) -> None:
nonlocal failed, stop
async with gate:
if stop is not None:
return
name = job.rel[-1]
shown = f"{handle(group)}/{remote_dir[tuple(job.rel[:-1])]}/{name}"
progress.start(safe(name))
lock = threading.Lock()
try:
with open(job.local, "rb") as fh:
def read(offset: int, length: int) -> bytes:
with lock:
fh.seek(offset)
return fh.read(length)
stored = await upload(
session, read, job.size, name, remote_dir[tuple(job.rel[:-1])],
progress=progress.advance,
on_queued=lambda n: progress.queued(safe(name), n))
except OSError as e:
failed += 1
progress.note(f"meshbay-cli: {job.local}: {e.strerror}")
return
except TransferError as e:
failed += 1
progress.note(f"meshbay-cli: {safe(shown)}: {e.message}")
if e.fatal:
stop = e
return
if stored != name:
progress.note(f"meshbay-cli: {safe(name)} was stored as {safe(stored)} "
"(the name was taken meanwhile)")
progress.finish(safe(name), path=shown.rsplit("/", 1)[0] + "/" + stored,
size=job.size)
await asyncio.gather(*(one(f) for f in files))
progress.clear()
if stop is not None:
skipped = len(files) - progress.files_done - failed
err(f"meshbay-cli: stopped: {stop.message}"
+ (f" ({skipped} file{'s' if skipped != 1 else ''} not sent)" if skipped else ""))
failed += skipped
if not a.quiet and not a.json:
done = progress.files_done
err(f"Uploaded {done} file{'s' if done != 1 else ''} "
f"({human_size(progress.bytes_done)})" + (f"; {failed} failed." if failed else "."))
return PARTIAL if failed else 0
|