aboutsummaryrefslogtreecommitdiffstats
path: root/packages/meshbay-cli/src/meshbay_cli/client/tree.py
blob: 6e537ef6e4dbd2ac711e3fb2c6decfa253869843 (plain) (blame)
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
"""
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))