codekingpro/portable-devtools

1from contextlib import contextmanager2from ctypes import (3    CFUNCTYPE,4    POINTER,5    c_int,6    c_longlong,7    c_void_p,8    cast,9    create_string_buffer,10)11 12import libarchive13import libarchive.ffi as ffi14 15from fsspec import open_files16from fsspec.archive import AbstractArchiveFileSystem17from fsspec.implementations.memory import MemoryFile18from fsspec.utils import DEFAULT_BLOCK_SIZE19 20# Libarchive requires seekable files or memory only for certain archive21# types. However, since we read the directory first to cache the contents22# and also allow random access to any file, the file-like object needs23# to be seekable no matter what.24 25# Seek call-backs (not provided in the libarchive python wrapper)26SEEK_CALLBACK = CFUNCTYPE(c_longlong, c_int, c_void_p, c_longlong, c_int)27read_set_seek_callback = ffi.ffi(28    "read_set_seek_callback", [ffi.c_archive_p, SEEK_CALLBACK], c_int, ffi.check_int29)30new_api = hasattr(ffi, "NO_OPEN_CB")31 32 33@contextmanager34def custom_reader(file, format_name="all", filter_name="all", block_size=ffi.page_size):35    """Read an archive from a seekable file-like object.36 37    The `file` object must support the standard `readinto` and 'seek' methods.38    """39    buf = create_string_buffer(block_size)40    buf_p = cast(buf, c_void_p)41 42    def read_func(archive_p, context, ptrptr):43        # readinto the buffer, returns number of bytes read44        length = file.readinto(buf)45        # write the address of the buffer into the pointer46        ptrptr = cast(ptrptr, POINTER(c_void_p))47        ptrptr[0] = buf_p48        # tell libarchive how much data was written into the buffer49        return length50 51    def seek_func(archive_p, context, offset, whence):52        file.seek(offset, whence)53        # tell libarchvie the current position54        return file.tell()55 56    read_cb = ffi.READ_CALLBACK(read_func)57    seek_cb = SEEK_CALLBACK(seek_func)58 59    if new_api:60        open_cb = ffi.NO_OPEN_CB61        close_cb = ffi.NO_CLOSE_CB62    else:63        open_cb = libarchive.read.OPEN_CALLBACK(ffi.VOID_CB)64        close_cb = libarchive.read.CLOSE_CALLBACK(ffi.VOID_CB)65 66    with libarchive.read.new_archive_read(format_name, filter_name) as archive_p:67        read_set_seek_callback(archive_p, seek_cb)68        ffi.read_open(archive_p, None, open_cb, read_cb, close_cb)69        yield libarchive.read.ArchiveRead(archive_p)70 71 72class LibArchiveFileSystem(AbstractArchiveFileSystem):73    """Compressed archives as a file-system (read-only)74 75    Supports the following formats:76    tar, pax , cpio, ISO9660, zip, mtree, shar, ar, raw, xar, lha/lzh, rar77    Microsoft CAB, 7-Zip, WARC78 79    See the libarchive documentation for further restrictions.80    https://www.libarchive.org/81 82    Keeps file object open while instance lives. It only works in seekable83    file-like objects. In case the filesystem does not support this kind of84    file object, it is recommended to cache locally.85 86    This class is pickleable, but not necessarily thread-safe (depends on the87    platform). See libarchive documentation for details.88    """89 90    root_marker = ""91    protocol = "libarchive"92    cachable = False93 94    def __init__(95        self,96        fo="",97        mode="r",98        target_protocol=None,99        target_options=None,100        block_size=DEFAULT_BLOCK_SIZE,101        **kwargs,102    ):103        """104        Parameters105        ----------106        fo: str or file-like107            Contains ZIP, and must exist. If a str, will fetch file using108            :meth:`~fsspec.open_files`, which must return one file exactly.109        mode: str110            Currently, only 'r' accepted111        target_protocol: str (optional)112            If ``fo`` is a string, this value can be used to override the113            FS protocol inferred from a URL114        target_options: dict (optional)115            Kwargs passed when instantiating the target FS, if ``fo`` is116            a string.117        """118        super().__init__(self, **kwargs)119        if mode != "r":120            raise ValueError("Only read from archive files accepted")121        if isinstance(fo, str):122            files = open_files(fo, protocol=target_protocol, **(target_options or {}))123            if len(files) != 1:124                raise ValueError(125                    f'Path "{fo}" did not resolve to exactly one file: "{files}"'126                )127            fo = files[0]128        self.of = fo129        self.fo = fo.__enter__()  # the whole instance is a context130        self.block_size = block_size131        self.dir_cache = None132 133    @contextmanager134    def _open_archive(self):135        self.fo.seek(0)136        with custom_reader(self.fo, block_size=self.block_size) as arc:137            yield arc138 139    @classmethod140    def _strip_protocol(cls, path):141        # file paths are always relative to the archive root142        return super()._strip_protocol(path).lstrip("/")143 144    def _get_dirs(self):145        fields = {146            "name": "pathname",147            "size": "size",148            "created": "ctime",149            "mode": "mode",150            "uid": "uid",151            "gid": "gid",152            "mtime": "mtime",153        }154 155        if self.dir_cache is not None:156            return157 158        self.dir_cache = {}159        list_names = []160        with self._open_archive() as arc:161            for entry in arc:162                if not entry.isdir and not entry.isfile:163                    # Skip symbolic links, fifo entries, etc.164                    continue165                self.dir_cache.update(166                    {167                        dirname: {"name": dirname, "size": 0, "type": "directory"}168                        for dirname in self._all_dirnames(set(entry.name))169                    }170                )171                f = {key: getattr(entry, fields[key]) for key in fields}172                f["type"] = "directory" if entry.isdir else "file"173                list_names.append(entry.name)174 175                self.dir_cache[f["name"]] = f176        # libarchive does not seem to return an entry for the directories (at least177        # not in all formats), so get the directories names from the files names178        self.dir_cache.update(179            {180                dirname: {"name": dirname, "size": 0, "type": "directory"}181                for dirname in self._all_dirnames(list_names)182            }183        )184 185    def _open(186        self,187        path,188        mode="rb",189        block_size=None,190        autocommit=True,191        cache_options=None,192        **kwargs,193    ):194        path = self._strip_protocol(path)195        if mode != "rb":196            raise NotImplementedError197 198        data = b""199        with self._open_archive() as arc:200            for entry in arc:201                if entry.pathname != path:202                    continue203 204                if entry.size == 0:205                    # empty file, so there are no blocks206                    break207 208                for block in entry.get_blocks(entry.size):209                    data = block210                    break211                else:212                    raise ValueError213        return MemoryFile(fs=self, path=path, data=data)214