""" A group's index as the Files tab draws it: roots, then folders, then files. Index entries carry `path` = the root's name and the folders, and `name`; the index also lists every directory (`dirs`, roots included), because an empty folder is not an entry and would otherwise not exist; and each root's `available` and `writable` (docs/MESHBAY_DESIGN.md §6.2, §9.5). """ from dataclasses import dataclass, field from fnmatch import fnmatchcase from ..errors import NOT_FOUND, CliError GLOB_CHARS = frozenset("*?[") def is_pattern(segment: str) -> bool: return any(c in GLOB_CHARS for c in segment) @dataclass class Item: name: str path: str # inside the group: root/folders/name is_dir: bool entry: dict | None = None # the index entry, for a file root: dict | None = None # the root's state, for a root available: bool = True @property def size(self) -> int: return int(self.entry.get("size", 0)) if self.entry else 0 @property def added_at(self) -> int: return int(self.entry.get("added_at", 0)) if self.entry else 0 @dataclass class Tree: roots: dict[str, dict] dirs: set[str] files: dict[str, list[dict]] = field(default_factory=dict) # parent path → entries @classmethod def from_index(cls, index: dict) -> "Tree": roots = {r["name"]: r for r in index.get("roots", [])} dirs = set(index.get("dirs", [])) files: dict[str, list[dict]] = {} for e in index.get("entries", []): files.setdefault(e["path"], []).append(e) # A file's folders exist even if the node's directory walk missed one. parts = e["path"].split("/") for i in range(1, len(parts) + 1): dirs.add("/".join(parts[:i])) dirs |= set(roots) return cls(roots, dirs, files) def apply_delta(self, delta: dict) -> bool: """ Bring the tree up to an `index_delta`, in place. True when the delta changed what the tree holds — a file added or removed, a root's state — rather than only the fields of files already there. A delta carries no `dirs`, so an empty folder made elsewhere is not in it: a structural change is where the caller asks for the whole index again, once. The common delta is enrichment filling in a file's fields, about one a second while a library is being enriched, and asking for the whole index on each of those took the download's own event loop: 67 full indexes in the 80 s of one download. """ updates = {e["id"]: e for e in delta.get("updates", [])} gone = set(delta.get("deletions", [])) for parent, entries in self.files.items(): if gone or any(e["id"] in updates for e in entries): self.files[parent] = [updates.get(e["id"], e) for e in entries if e["id"] not in gone] known = {e["id"] for entries in self.files.values() for e in entries} added = [e for e in delta.get("additions", []) if e["id"] not in known] for e in added: self.files.setdefault(e["path"], []).append(e) parts = e["path"].split("/") for i in range(1, len(parts) + 1): self.dirs.add("/".join(parts[:i])) roots = {r["name"]: r for r in delta.get("roots", [])} moved_roots = bool(roots) and roots != self.roots if moved_roots: self.roots = roots self.dirs |= set(roots) return bool(gone or added or moved_roots) def root_available(self, path: str) -> bool: root = self.roots.get(path.split("/", 1)[0]) return bool(root.get("available", True)) if root else True def is_dir(self, path: str) -> bool: return path in self.dirs def file(self, path: str) -> dict | None: parent, _, name = path.rpartition("/") for e in self.files.get(parent, []): if e["name"] == name: return e return None def children(self, path: str, *, include_unavailable: bool = False) -> list[Item]: """What a folder holds — at the top, the roots. Unavailable roots' content is left out unless asked for, as the Files tab does.""" if not path: return [Item(name, name, True, root=r, available=bool(r.get("available", True))) for name, r in self.roots.items()] if not self.is_dir(path): raise CliError(NOT_FOUND, f"No folder {path!r} in this group.") available = self.root_available(path) if not available and not include_unavailable: return [] prefix = path + "/" items = [Item(d[len(prefix):], d, True, available=available) for d in self.dirs if d.startswith(prefix) and "/" not in d[len(prefix):]] items += [Item(e["name"], f"{path}/{e['name']}", False, entry=e, available=available) for e in self.files.get(path, [])] return items def walk(self, path: str, *, include_unavailable: bool = False): """(folder, children) for `path` and every folder below it, depth first.""" kids = self.children(path, include_unavailable=include_unavailable) yield path, kids for item in sorted((k for k in kids if k.is_dir), key=lambda k: k.name.lower()): yield from self.walk(item.path, include_unavailable=include_unavailable) def expand(self, parts: list[str]) -> list[str]: """ The paths `parts` names, wildcards (`*`, `?`, `[...]`, one segment each) matched against the index rather than by a shell, so a pattern means the same in bash and in PowerShell. Matching is case-sensitive, as paths are. """ paths = [""] for seg in parts: found = [] for base in paths: if base and not self.is_dir(base): continue if is_pattern(seg): found += [i.path for i in self.children(base) if fnmatchcase(i.name, seg)] else: p = f"{base}/{seg}" if base else seg if self.is_dir(p) or self.file(p) is not None: found.append(p) paths = found return sorted(set(paths))