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
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
|
"""
`get`: download files and folders from groups.
The whole run is planned before the first byte: paths and patterns resolved
against each group's index, local names made writable here, existing files
found. A path that names nothing stops the run there (exit 3); a file that
cannot be written is reported and the others go on (exit 4).
"""
import asyncio
import os
from dataclasses import dataclass
from pathlib import Path
from meshbay_common.paths import fold
from ..client.groups import my_groups
from ..client.names import UnsafeName, local_path
from ..client.paths import handle, parse, resolve_group
from ..client.pool import Pool
from ..client.session import GroupSession
from ..client.transfers import PART_SUFFIX, TransferError, download
from ..client.tree import Tree
from ..errors import NODE, NOT_FOUND, PARTIAL, USAGE, CliError
from ..output import Progress, err, human_size, safe, sentence
from .context import Context
HELP = "download files, or folders with -R"
PER_GROUP = 2 # the node's default lease cap per member and group (§5.5)
def add_parser(sub, parent) -> None:
p = sub.add_parser("get", parents=[parent], help=HELP, description=sentence(HELP)
+ ". Wildcards (* ? [...]) are matched against the group, "
"so quote them.")
p.add_argument("paths", nargs="+", metavar="PATH", help="group[@owner]/root/...")
p.add_argument("-R", dest="recursive", action="store_true", help="folders, recursively")
p.add_argument("-o", dest="dest", metavar="DEST",
help="a folder (created when it ends with /), a file name for a "
"single file, or - for stdout; default: the current folder")
exist = p.add_mutually_exclusive_group()
exist.add_argument("-f", dest="force", action="store_true",
help="overwrite existing files")
exist.add_argument("-n", dest="skip", action="store_true",
help="skip existing files silently")
@dataclass
class Job:
group: dict
session: GroupSession
entry: dict
remote: str # as shown: handle/root/folders/name
target: Path | None # None: stdout
async def run(ctx: Context) -> int:
await ctx.hub.check_version()
account = ctx.account()
pool = Pool(account)
try:
return await _run(ctx, account, pool)
finally:
await pool.close()
await account.close()
async def _run(ctx: Context, account, pool: Pool) -> int:
a = ctx.args
groups = await my_groups(account)
trees: dict[str, Tree] = {}
async def tree_of(group: dict) -> tuple[GroupSession, Tree]:
session = await pool.session(group)
if group["id"] not in trees:
trees[group["id"]] = Tree.from_index(await session.index())
return session, trees[group["id"]]
# ── What the paths name ──────────────────────────────────────────────────
named: list[tuple[dict, GroupSession, Tree, str]] = [] # group, session, tree, path
missing: list[str] = []
unreachable: list[str] = []
for text in a.paths:
remote = parse(text)
if remote is None:
raise CliError(USAGE, "Name a group, and what to fetch in it.")
group = resolve_group(groups, remote.group)
try:
session, tree = await tree_of(group)
except CliError as e:
# One group whose node is away does not cost the others their files.
if e.code != NODE:
raise
unreachable.append(f"{text}: {e.message}")
continue
found = tree.expand(remote.parts)
if not found and (remote.inner and (tree.is_dir(remote.inner)
or tree.file(remote.inner))):
found = [remote.inner] # a literal name with brackets in it
if not found:
missing.append(text)
named += [(group, session, tree, p) for p in found]
if missing:
raise CliError(NOT_FOUND, "Nothing in your groups matches: "
+ ", ".join(safe(m) for m in missing))
to_stdout = a.dest == "-"
files = [(g, s, t, p) for g, s, t, p in named if p and not t.is_dir(p)]
folders = [(g, s, t, p) for g, s, t, p in named if not p or t.is_dir(p)]
if unreachable and not named:
raise CliError(NODE, "\n".join(safe(u) for u in unreachable))
if folders and not a.recursive:
raise CliError(USAGE, "These are folders; use -R: "
+ ", ".join(safe(_shown(g, p)) for g, _, _, p in folders))
if to_stdout and (folders or len(files) != 1):
raise CliError(USAGE, "-o - writes exactly one file to stdout.")
# ── Where it goes ────────────────────────────────────────────────────────
single_file = not folders and len(files) == 1 and not to_stdout
dest, file_name = _destination(a.dest, single_file)
problems: list[str] = list(unreachable)
jobs: list[Job] = []
dirs: list[Path] = []
planned: set[str] = set()
renamed: list[tuple[str, Path]] = []
def plan_file(group, session, tree, path, parts):
entry = tree.file(path)
shown = _shown(group, path)
if not tree.root_available(path):
problems.append(f"{shown}: its folder on the node is unavailable")
return
if to_stdout:
jobs.append(Job(group, session, entry, shown, None))
return
try:
target, changed = local_path(dest, parts)
except UnsafeName as e:
problems.append(f"{shown}: {e}")
return
key = fold(os.path.normcase(str(target)))
if key in planned:
problems.append(f"{shown}: another file of this run has the same local name "
f"{target}")
return
planned.add(key)
if changed:
renamed.append((shown, target))
if target.exists() and not target.with_name(target.name + PART_SUFFIX).exists():
if a.skip:
return
if not a.force:
problems.append(f"{target} already exists (-f to overwrite, -n to skip)")
return
jobs.append(Job(group, session, entry, shown, target))
for group, session, tree, path in files:
name = path.rsplit("/", 1)[-1]
plan_file(group, session, tree, path, [file_name or name])
for group, session, tree, path in folders:
top = path.rsplit("/", 1)[-1] if path else group["name"]
for folder, items in tree.walk(path):
rel = [top] + (folder[len(path):].strip("/").split("/")
if folder != path else [])
rel = [r for r in rel if r]
try:
dirs.append(local_path(dest, rel)[0])
except UnsafeName as e:
problems.append(f"{_shown(group, folder)}: {e}")
continue
for item in items:
if not item.is_dir:
plan_file(group, session, tree, item.path, rel + [item.name])
# ── Fetch ────────────────────────────────────────────────────────────────
progress = Progress("↓", quiet=a.quiet or to_stdout and not a.json, as_json=a.json)
for shown, target in renamed:
progress.note(f"meshbay-cli: {safe(shown)} is saved as {target.name} "
"(its name cannot be written here as it is)")
for problem in problems:
progress.note(f"meshbay-cli: {safe(problem)}")
for d in dirs:
d.mkdir(parents=True, exist_ok=True)
progress.plan(len(jobs), sum(int(j.entry["size"]) for j in jobs))
failed = len(problems)
gates: dict[str, asyncio.Semaphore] = {}
async def one(job: Job) -> None:
nonlocal failed
gate = gates.setdefault(job.group["id"], asyncio.Semaphore(PER_GROUP))
async with gate:
progress.start(safe(job.entry["name"]))
try:
await download(job.session, job.entry, job.target, overwrite=a.force,
progress=progress.advance,
on_queued=lambda n: progress.queued(
safe(job.entry["name"]), n))
progress.finish(safe(job.entry["name"]), path=job.remote,
to=str(job.target or "-"))
except TransferError as e:
failed += 1
progress.note(f"meshbay-cli: {safe(job.remote)}: {e.message}")
await asyncio.gather(*(one(j) for j in jobs))
progress.clear()
if not a.quiet and not to_stdout and not a.json:
done = len(jobs) - (failed - len(problems))
err(f"Downloaded {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
def _shown(group: dict, path: str) -> str:
return f"{handle(group)}/{path}".rstrip("/")
def _destination(dest: str | None, single_file: bool) -> tuple[Path, str | None]:
"""The folder files land in, and a file name when -o names one."""
if dest is None or dest == "-":
return Path.cwd(), None
path = Path(dest).expanduser()
if dest.endswith(("/", os.sep)):
path.mkdir(parents=True, exist_ok=True)
return path, None
if path.is_dir():
return path, None
if single_file:
if not path.parent.is_dir():
raise CliError(USAGE, f"{path.parent} is not a folder.")
return path.parent, path.name
raise CliError(USAGE, f"{dest} is not a folder (end it with / to create it).")
|