| from fsspec import AbstractFileSystem |
| from fsspec.utils import tokenize |
|
|
|
|
| class AbstractArchiveFileSystem(AbstractFileSystem): |
| """ |
| A generic superclass for implementing Archive-based filesystems. |
| |
| Currently, it is shared amongst |
| :class:`~fsspec.implementations.zip.ZipFileSystem`, |
| :class:`~fsspec.implementations.libarchive.LibArchiveFileSystem` and |
| :class:`~fsspec.implementations.tar.TarFileSystem`. |
| """ |
|
|
| def __str__(self): |
| return f"<Archive-like object {type(self).__name__} at {id(self)}>" |
|
|
| __repr__ = __str__ |
|
|
| def ukey(self, path): |
| return tokenize(path, self.fo, self.protocol) |
|
|
| def _all_dirnames(self, paths): |
| """Returns *all* directory names for each path in paths, including intermediate |
| ones. |
| |
| Parameters |
| ---------- |
| paths: Iterable of path strings |
| """ |
| if len(paths) == 0: |
| return set() |
|
|
| dirnames = {self._parent(path) for path in paths} - {self.root_marker} |
| return dirnames | self._all_dirnames(dirnames) |
|
|
| def info(self, path, **kwargs): |
| self._get_dirs() |
| path = self._strip_protocol(path) |
| if path in {"", "/"} and self.dir_cache: |
| return {"name": "", "type": "directory", "size": 0} |
| if path in self.dir_cache: |
| return self.dir_cache[path] |
| elif path + "/" in self.dir_cache: |
| return self.dir_cache[path + "/"] |
| else: |
| raise FileNotFoundError(path) |
|
|
| def ls(self, path, detail=True, **kwargs): |
| self._get_dirs() |
| paths = {} |
| for p, f in self.dir_cache.items(): |
| p = p.rstrip("/") |
| if "/" in p: |
| root = p.rsplit("/", 1)[0] |
| else: |
| root = "" |
| if root == path.rstrip("/"): |
| paths[p] = f |
| elif all( |
| (a == b) |
| for a, b in zip(path.split("/"), [""] + p.strip("/").split("/")) |
| ): |
| |
| ppath = p.rstrip("/").split("/", 1)[0] |
| if ppath not in paths: |
| out = {"name": ppath, "size": 0, "type": "directory"} |
| paths[ppath] = out |
| if detail: |
| out = sorted(paths.values(), key=lambda _: _["name"]) |
| return out |
| else: |
| return sorted(paths) |
|
|
