aboutsummaryrefslogtreecommitdiffstats
path: root/packages/meshbay-cli/src/meshbay_cli/client/tree.py
diff options
context:
space:
mode:
Diffstat (limited to 'packages/meshbay-cli/src/meshbay_cli/client/tree.py')
-rw-r--r--packages/meshbay-cli/src/meshbay_cli/client/tree.py119
1 files changed, 119 insertions, 0 deletions
diff --git a/packages/meshbay-cli/src/meshbay_cli/client/tree.py b/packages/meshbay-cli/src/meshbay_cli/client/tree.py
new file mode 100644
index 0000000..6e537ef
--- /dev/null
+++ b/packages/meshbay-cli/src/meshbay_cli/client/tree.py
@@ -0,0 +1,119 @@
+"""
+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 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))