""" The sign-in screen, and the main screen: two panes and the transfers. F5 copies the selection to the other pane's folder — a download or an upload, with the rules `get` and `put` follow: an upload is decided whole before it starts (names the node refuses stop it, names already there are asked about once for the batch), nothing on either side is replaced without being asked, and a name from a group is made writable here before it is written. Delete removes the selection from a group with the rules `rm` follows: files only, every one of them this account's to delete (uploaded by it, or this key the node's operator) or none is, then asked once for the batch. """ import asyncio import contextlib import threading from pathlib import Path from rich.text import Text from textual.app import ComposeResult from textual.binding import Binding from textual.containers import Horizontal, Vertical from textual.screen import Screen from textual.widgets import Button, Footer, Input, Static from ..client.account import Account from ..client.names import UnsafeName, local_path from ..client.paths import handle from ..client.pool import Pool from ..client.remove import delete_file, may_delete from ..client.store import FILE_KEY_NOTICE, KEY_FILE from ..client.transfers import PART_SUFFIX, TransferError, create_dir, download, upload from ..client.tree import Tree from ..client.upload_plan import plan_upload, refused_names, taken_names from ..errors import CliError from ..output import safe from .panes import LocalPane, RemotePane from .widgets import Choice, TransferStrip PER_GROUP = 2 class SignInScreen(Screen): DEFAULT_CSS = """ SignInScreen { align: center middle; } SignInScreen > Vertical { width: 64; height: auto; border: thick $accent; padding: 1 2; background: $surface; } SignInScreen Input { margin-bottom: 1; } SignInScreen #error { color: $error; height: auto; } """ def compose(self) -> ComposeResult: ctx = self.app.ctx with Vertical(): yield Static(Text("Sign in to MeshBay")) yield Input(value=ctx.hub_url, placeholder="hub", id="hub") yield Input(value=ctx.username or ctx.stored_username or "", placeholder="username", id="user") yield Input(placeholder="passphrase", password=True, id="pass") yield Button("Sign in", id="go", variant="primary") yield Static("", id="error", markup=False) def on_mount(self) -> None: empty = next((i for i in self.query(Input) if not i.value), None) (empty or self.query_one("#pass", Input)).focus() def on_input_submitted(self, event: Input.Submitted) -> None: self._submit() def on_button_pressed(self, event: Button.Pressed) -> None: self._submit() def _submit(self) -> None: hub = self.query_one("#hub", Input).value.strip().rstrip("/") user = self.query_one("#user", Input).value.strip() passphrase = self.query_one("#pass", Input).value if not (hub and user and passphrase): self.query_one("#error", Static).update("Hub, username and passphrase.") return self.query_one("#error", Static).update("Signing in…") self.run_worker(self._sign_in(hub, user, passphrase), exclusive=True) async def _sign_in(self, hub_url: str, user: str, passphrase: str) -> None: app = self.app try: await app.use_hub(hub_url) account = Account(app.hub, user, lambda: passphrase) await account.hub.check_version() await account.sign_in(remember=True) kind = await account.remember(app.ctx.store) except CliError as e: self.query_one("#error", Static).update(e.message) return if kind == KEY_FILE: app.notify(FILE_KEY_NOTICE, timeout=10, markup=False) app.start(account) class MainScreen(Screen): BINDINGS = [ Binding("f5", "copy", "Copy"), Binding("delete", "delete", "Delete"), Binding("q", "app.quit", "Quit"), Binding("tab", "app.focus_next", "Switch", show=False), ] DEFAULT_CSS = """ MainScreen > Horizontal { height: 1fr; } """ def __init__(self, account: Account): super().__init__() self.account = account self.pool = Pool(account) self._trees: dict[str, Tree] = {} self._gates: dict[str, asyncio.Semaphore] = {} self._watching: set[str] = set() def compose(self) -> ComposeResult: with Horizontal(): yield LocalPane(Path.cwd(), id="local") yield RemotePane(self, id="remote") yield TransferStrip(id="transfers") yield Footer() def on_mount(self) -> None: self.query_one(LocalPane).table.focus() @property def local(self) -> LocalPane: return self.query_one(LocalPane) @property def remote(self) -> RemotePane: return self.query_one(RemotePane) @property def strip(self) -> TransferStrip: return self.query_one(TransferStrip) # ── Groups ─────────────────────────────────────────────────────────────── async def tree(self, group: dict) -> Tree: """The group's index, asked once and then kept current by the node's pushes.""" if group["id"] not in self._trees: session = await self.pool.session(group) self._trees[group["id"]] = Tree.from_index(await session.index()) if group["id"] not in self._watching: self._watching.add(group["id"]) session.ch.on_push("index_delta", lambda _msg, g=group: self._changed(g)) return self._trees[group["id"]] def forget_tree(self, group: dict) -> None: self._trees.pop(group["id"], None) def _changed(self, group: dict) -> None: """Something changed in the group: the index is asked for again, and the pane redrawn if it is showing that group.""" self.forget_tree(group) remote = self.remote if remote.group is not None and remote.group["id"] == group["id"]: remote.load() # ── F5 ─────────────────────────────────────────────────────────────────── def action_copy(self) -> None: focused = self.app.focused if focused is not None and self.remote in focused.ancestors_with_self: self.run_worker(self._download(), group="copy") else: self.run_worker(self._upload(), group="copy") def _gate(self, group: dict) -> asyncio.Semaphore: return self._gates.setdefault(group["id"], asyncio.Semaphore(PER_GROUP)) async def _ask(self, message: str, options: list[tuple[str, str]]) -> str: return await self.app.push_screen(Choice(message, options), wait_for_dismiss=True) async def _download(self) -> None: remote, local = self.remote, self.local rows = remote.selection() if remote.group is None or not rows: self.app.notify("Open a group and choose what to download.", timeout=4) return group, dest = remote.group, local.path try: tree = await self.tree(group) except CliError as e: self.app.notify(e.message, severity="error", markup=False) return plan: list[tuple[dict, Path]] = [] dirs: list[Path] = [] problems: list[str] = [] for row in rows: path = row.data["path"] top = path.rsplit("/", 1)[-1] if not tree.root_available(path): problems.append(f"{top}: its folder on the node is unavailable") continue if not row.is_dir: try: plan.append((row.data["entry"], local_path(dest, [top])[0])) except UnsafeName as e: problems.append(str(e)) continue for folder, items in tree.walk(path): rel = [top] + [p for p in folder[len(path):].strip("/").split("/") if p] try: dirs.append(local_path(dest, rel)[0]) plan += [(i.entry, local_path(dest, rel + [i.name])[0]) for i in items if not i.is_dir] except UnsafeName as e: problems.append(str(e)) for p in problems: self.app.notify(p, severity="warning", markup=False) existing = [t for _, t in plan if t.exists() and not t.with_name(t.name + PART_SUFFIX).exists()] overwrite = False if existing: names = ", ".join(safe(t.name) for t in existing[:5]) + ( f" and {len(existing) - 5} more" if len(existing) > 5 else "") answer = await self._ask( f"{len(existing)} file(s) already in {dest}: {names}.", [("skip", "Skip them"), ("overwrite", "Overwrite"), ("cancel", "Cancel")]) if answer == "cancel": return overwrite = answer == "overwrite" if not overwrite: plan = [(e, t) for e, t in plan if t not in existing] for d in dirs: await asyncio.to_thread(d.mkdir, parents=True, exist_ok=True) remote.marks.clear() remote.redraw() session = await self.pool.session(group) await asyncio.gather(*(self._one_download(group, session, e, t, overwrite) for e, t in plan)) local.action_refresh() async def _one_download(self, group, session, entry: dict, target: Path, overwrite: bool) -> None: job = self.strip.add("↓", entry["name"], int(entry["size"])) async with self._gate(group): self.strip.update_job(job, state="running") try: await download( session, entry, target, overwrite=overwrite, progress=lambda n: self.strip.update_job(job, done=job.done + n), on_queued=lambda n: self.strip.update_job(job, state="queued", note=f"({n} ahead)")) self.strip.update_job(job, state="done", note="" if target.name == entry["name"] else f"saved as {target.name}") except TransferError as e: self.strip.update_job(job, state="failed", note=e.message) async def _upload(self) -> None: remote, local = self.remote, self.local rows = local.selection() if not rows: return if remote.group is None or not remote.path: self.app.notify("Open a folder of a group on the right to upload into.", timeout=4) return group, folder = remote.group, remote.path try: tree = await self.tree(group) except CliError as e: self.app.notify(e.message, severity="error", markup=False) return root = tree.roots.get(folder.split("/", 1)[0], {}) if not root.get("writable") or not root.get("available", True): self.app.notify(f"{safe(handle(group))}/{safe(folder)} does not accept " "uploads (read-only or unavailable).", severity="error", markup=False) return try: dirs, files, problems = plan_upload([str(local.path / r.name) for r in rows], recursive=True) except CliError as e: self.app.notify(e.message, severity="error", markup=False) return problems += refused_names([f.rel[-1] for f in files], dirs) if problems: await self._ask("Nothing was sent:\n" + "\n".join(safe(p) for p in problems), [("ok", "OK")]) return tops = sorted({d[0] for d in dirs} | {f.rel[0] for f in files}) taken = taken_names(tree, folder, tops) if taken: answer = await self._ask( f"Already in {safe(folder)}: {', '.join(safe(t) for t in sorted(taken))}. " "The node never replaces a file.", [("skip", "Skip them"), ("cancel", "Cancel")]) if answer != "skip": return dirs = [d for d in dirs if d[0] not in taken] files = [f for f in files if f.rel[0] not in taken] local.marks.clear() local.redraw() session = await self.pool.session(group) remote_dir = {(): 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: self.app.notify(e.message, severity="error", markup=False) return await asyncio.gather(*(self._one_upload(group, session, f, remote_dir[tuple(f.rel[:-1])]) for f in files)) self._changed(group) async def _one_upload(self, group, session, item, folder: str) -> None: name = item.rel[-1] job = self.strip.add("↑", name, item.size) async with self._gate(group): self.strip.update_job(job, state="running") lock = threading.Lock() try: with open(item.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, item.size, name, folder, progress=lambda n: self.strip.update_job(job, done=job.done + n), on_queued=lambda n: self.strip.update_job(job, state="queued", note=f"({n} ahead)")) self.strip.update_job(job, state="done", note="" if stored == name else f"stored as {stored}") except OSError as e: self.strip.update_job(job, state="failed", note=e.strerror or str(e)) except TransferError as e: self.strip.update_job(job, state="failed", note=e.message) # ── Delete ─────────────────────────────────────────────────────────────── def action_delete(self) -> None: focused = self.app.focused if focused is None or self.remote not in focused.ancestors_with_self: self.app.notify("Delete removes files from a group: select them on the " "right.", timeout=4) return self.run_worker(self._delete(), group="delete") async def _delete(self) -> None: remote = self.remote rows = remote.selection() if remote.group is None or not rows: self.app.notify("Open a group and choose the files to delete.", timeout=4) return group = remote.group try: session = await self.pool.session(group) except CliError as e: self.app.notify(e.message, severity="error", markup=False) return folders = [r.name for r in rows if r.is_dir] not_yours = [r.name for r in rows if not r.is_dir and not may_delete(session, r.data["entry"], self.account.user_id)] if folders or not_yours: why = (["Folders are not deleted here: " + ", ".join(map(safe, folders))] if folders else []) if not_yours: why.append("You did not upload these, and this machine's key is not the " "node's operator: " + ", ".join(map(safe, not_yours))) await self._ask("Nothing was deleted.\n" + "\n".join(why), [("ok", "OK")]) return names = ", ".join(safe(r.name) for r in rows[:5]) + ( f" and {len(rows) - 5} more" if len(rows) > 5 else "") answer = await self._ask( f"Delete {len(rows)} file(s) from {safe(handle(group))} for every member? " f"{names}", [("delete", "Delete"), ("cancel", "Cancel")]) if answer != "delete": return remote.marks.clear() deleted = 0 for row in rows: try: await delete_file(session, row.data["entry"]["id"]) deleted += 1 except CliError as e: self.app.notify(f"{safe(row.name)}: {e.message}", severity="error", markup=False, timeout=8) if deleted: self.app.notify(f"Deleted {deleted} file(s).", timeout=4) self._changed(group) async def on_unmount(self) -> None: with contextlib.suppress(Exception): await self.pool.close() class DeviceCode: """`account.on_device_code` for the interactive mode: the code in a notification that stays until the approval lands; the node is retried meanwhile.""" def __init__(self, app): self.app = app def __call__(self, group: dict, pending, *, first: bool) -> bool: if first: self.app.notify( f"This machine is not yet a device of yours on the node hosting " f"{safe(handle(group))}. Approve it from the MeshBay application or a " f"browser: the group's Settings → \"Your devices on this node\" → " f"enter {pending.link_code}; or on a machine already linked: " f"meshbay-cli devices approve {safe(handle(group))} {pending.link_code}. " "Waiting…", title="Approve this machine", timeout=600, markup=False) return True