fuse-python使用fuse来挂载fs

winfsp

安装winfsp,https://winfsp.dev/

挂载window目录

python安装fusepy

#!/usr/bin/env python3
import os
import stat
from fuse import FUSE, FuseOSError, Operations

class Passthrough(Operations):
    def __init__(self, root):
        self.root = root

    # 辅助函数：将挂载点路径转换为根目录路径
    def _full_path(self, partial):
        if partial.startswith("/"):
            partial = partial[1:]
        path = os.path.join(self.root, partial)
        return path

    # 获取文件或目录的属性
    def getattr(self, path, fh=None):
        full_path = self._full_path(path)
        st = os.lstat(full_path)
        return dict((key, getattr(st, key)) for key in ('st_atime', 'st_ctime',
                     'st_gid', 'st_mode', 'st_mtime', 'st_nlink', 'st_size', 'st_uid'))

    # 读取目录内容
    def readdir(self, path, fh):
        full_path = self._full_path(path)
        dirents = ['.', '..']
        if os.path.isdir(full_path):
            dirents.extend(os.listdir(full_path))
        for r in dirents:
            yield r

    # 打开文件
    def open(self, path, flags):
        full_path = self._full_path(path)
        return os.open(full_path, flags)

    # 读取文件内容
    def read(self, path, length, offset, fh):
        os.lseek(fh, offset, os.SEEK_SET)
        return os.read(fh, length)

def main(mountpoint, root):
    try:
        FUSE(Passthrough(root), mountpoint, nothreads=True, foreground=True)
    except Exception as e:
        print(f"Error: {e}")

if __name__ == '__main__':
    import sys
    if len(sys.argv) != 3:
        print('usage: %s <root> <mountpoint>' % sys.argv[0])
        sys.exit(1)
    main(sys.argv[2],sys.argv[1])

pyinstaller --onefile mount.py

打包成mount命令

mount d:\ z:

挂载http文件服务

import os
import requests
from fuse import FUSE, FuseOSError, Operations


class HTTPFuse(Operations):
    def __init__(self, base_url):
        self.base_url = base_url

    def getattr(self, path, fh=None):
        if path == '/':
            # 根目录属性
            return {
                'st_mode': (stat.S_IFDIR | 0o755),
                'st_nlink': 2
            }
        url = self.base_url + path.lstrip('/')
        try:
            response = requests.head(url)
            if response.status_code == 200:
                size = int(response.headers.get('Content-Length', 0))
                return {
                    'st_mode': (stat.S_IFREG | 0o444),
                    'st_nlink': 1,
                    'st_size': size
                }
            else:
                raise FuseOSError(2)  # 2 表示文件或目录不存在
        except requests.RequestException:
            raise FuseOSError(2)

    def readdir(self, path, fh):
        # 这里简单假设没有子目录，仅返回当前文件列表
        if path == '/':
            # 可以通过解析页面或者其他方式获取文件列表，这里简化处理
            yield '.'
            yield '..'
            # 这里假设可以通过某种方式获取文件列表，此处简化示例
            # 实际中可能需要解析页面或者调用 API 获取
            # 示例代码只是占位，需要根据实际情况修改
            # 假设我们知道有两个文件
            yield 'file1.txt'
            yield 'file2.txt'

    def read(self, path, length, offset, fh):
        url = self.base_url + path.lstrip('/')
        try:
            response = requests.get(url, headers={'Range': f'bytes={offset}-{offset + length - 1}'})
            if response.status_code == 206:  # 206 表示部分内容
                return response.content
            elif response.status_code == 200:  # 可能不支持 Range 请求
                return response.content[offset:offset + length]
            else:
                raise FuseOSError(2)
        except requests.RequestException:
            raise FuseOSError(2)


def main(mountpoint, base_url):
    FUSE(HTTPFuse(base_url), mountpoint, nothreads=True, foreground=True)


if __name__ == '__main__':
    import sys
    if len(sys.argv) != 3:
        print('usage: %s  <base_url> <mountpoint>' % sys.argv[0])
        sys.exit(1)
    main(sys.argv[2],sys.argv[1])