storage/vaultfs.py

"""
Read interface over a Vault, plus an optional WinFsp mount.

`VaultFS` is a dependency-free read model: list files, stat, and read byte ranges
that are reconstructed (and RS-healed) on demand from the Vault. This is what the
CLI `export` uses and what the mount serves.

`mount()` exposes the Vault as a real drive letter via WinFsp (Windows FUSE). It
is guarded: if `winfspy` / WinFsp is not installed, it raises with instructions
instead of failing obscurely. The dependency-free path (VaultFS + CLI export)
works everywhere; the live drive lights up once WinFsp is present.
"""
from __future__ import annotations
from collections import OrderedDict
from .vault import Vault


class VaultFS:
    def __init__(self, vault: Vault, cache_files: int = 8):
        self.vault = vault
        self._cache: OrderedDict[str, bytes] = OrderedDict()
        self._cache_n = cache_files
        self._index = {rel: sz for rel, sz in vault.list_files()}

    def listdir(self):
        return sorted(self._index.items())            # [(relpath, size)]

    def size(self, relpath: str) -> int:
        return self._index[relpath]

    def _data(self, relpath: str) -> bytes:
        if relpath in self._cache:
            self._cache.move_to_end(relpath)
            return self._cache[relpath]
        data = self.vault.read_file(relpath)           # reconstruct + RS-heal
        self._cache[relpath] = data
        if len(self._cache) > self._cache_n:
            self._cache.popitem(last=False)
        return data

    def read(self, relpath: str, offset: int = 0, length: int | None = None) -> bytes:
        data = self._data(relpath)
        if length is None:
            return data[offset:]
        return data[offset:offset + length]


def winfsp_available() -> bool:
    try:
        import winfspy  # noqa: F401
        return True
    except Exception:
        return False


def mount(vault: Vault, mountpoint: str, label: str = "NeuralVault"):
    """Mount the vault as a read-only drive via WinFsp. Requires WinFsp + winfspy.
    Experimental: the dependency-free `export` path is the tested one."""
    if not winfsp_available():
        raise RuntimeError(
            "WinFsp / winfspy not found. Install WinFsp (https://winfsp.dev) and "
            "`pip install winfspy`, or use the dependency-free CLI: "
            "`python cli.py export <vault> <folder>`."
        )
    # Imported lazily so the module loads without winfspy present.
    from winfspy import (FileSystem, BaseFileSystemOperations, NTStatusObjectNameNotFound,
                         FILE_ATTRIBUTE)
    import time

    fs_view = VaultFS(vault)
    now = int(time.time() * 1e7) + 116444736000000000  # FILETIME epoch

    class Ops(BaseFileSystemOperations):
        def _info(self, is_dir, size=0):
            return {
                "file_attributes": FILE_ATTRIBUTE.FILE_ATTRIBUTE_DIRECTORY if is_dir
                else FILE_ATTRIBUTE.FILE_ATTRIBUTE_NORMAL,
                "allocation_size": size, "file_size": size,
                "creation_time": now, "last_access_time": now,
                "last_write_time": now, "change_time": now, "index_number": 0,
            }

        def _norm(self, p):
            return p.replace("\\", "/").lstrip("/")

        def get_volume_info(self):
            return {"total_size": 0, "free_size": 0, "volume_label": label}

        def get_security_by_name(self, file_name):
            rel = self._norm(file_name)
            is_dir = rel == "" or any(f.startswith(rel + "/") for f in fs_view._index)
            if not is_dir and rel not in fs_view._index:
                raise NTStatusObjectNameNotFound()
            return (self._info(is_dir, 0 if is_dir else fs_view.size(rel))["file_attributes"], None, 0)

        def open(self, file_name, create_options, granted_access):
            rel = self._norm(file_name)
            is_dir = rel == "" or any(f.startswith(rel + "/") for f in fs_view._index)
            if not is_dir and rel not in fs_view._index:
                raise NTStatusObjectNameNotFound()
            return {"rel": rel, "is_dir": is_dir}

        def get_file_info(self, fh):
            return self._info(fh["is_dir"], 0 if fh["is_dir"] else fs_view.size(fh["rel"]))

        def read(self, fh, offset, length):
            return fs_view.read(fh["rel"], offset, length)

        def read_directory(self, fh, marker):
            prefix = "" if fh["rel"] == "" else fh["rel"] + "/"
            names = set()
            for f in fs_view._index:
                if f.startswith(prefix):
                    rest = f[len(prefix):]
                    names.add(rest.split("/")[0])
            out = []
            for nm in sorted(names):
                full = prefix + nm
                is_dir = full not in fs_view._index
                out.append({"file_name": nm,
                            "file_info": self._info(is_dir, 0 if is_dir else fs_view.size(full))})
            return out

        def close(self, fh):
            pass

    fs = FileSystem(mountpoint, Ops(), volume_label=label)
    fs.start()
    return fs