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
|
"""
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))
|