diff options
| -rw-r--r-- | hello-fusepy.py | 112 |
1 files changed, 85 insertions, 27 deletions
diff --git a/hello-fusepy.py b/hello-fusepy.py index 41c30a5..658d88e 100644 --- a/hello-fusepy.py +++ b/hello-fusepy.py @@ -1,50 +1,108 @@ import sys import errno import stat -from fuse import FUSE, Operations, FuseOSError +import logging +import requests +from fuse import ( + FUSE, + Operations, + FuseOSError, + LoggingMixIn, + fuse_exit, + fuse_get_context, +) -class HelloWorld(Operations): +api_url = "https://furry.engineer/users/uvok/outbox?min_id=0&page=true" + + +class Status(object): + def __init__(self, id: str, content: str): + self.id = id + self.content = content + + +class HelloWorld(Operations, LoggingMixIn): def __init__(self): - self.files = {'/hello': 'hello world'} - self.attr = { - '/': { - 'st_mode': (stat.S_IFDIR | 0o755), # Directory - 'st_nlink': 2 - }, - '/hello': { - 'st_mode': (stat.S_IFREG | 0o444), # Regular file, read-only - 'st_nlink': 1, - 'st_size': len(self.files['/hello']), - } - } + self.statuses: list[Status] = [] + self.fd = 0 + # '/hello': { + # 'st_mode': (stat.S_IFREG | 0o444), # Regular file, read-only + # 'st_nlink': 1, + # 'st_size': len(self.files['/hello']), + # } def getattr(self, path, fh=None): - if path in self.attr: - return self.attr[path] + (uid, gid, _) = fuse_get_context() + if path == "/": + return { + "st_mode": (stat.S_IFDIR | 0o700), # Directory + "st_nlink": 2, + "st_uid": uid, + "st_gid": gid, + } + found = next((s for s in self.statuses if "/" + s.id == path), None) + if found: + return { + "st_mode": (stat.S_IFREG | 0o400), + "st_size": len(found.content.encode("utf8")), + "st_uid": uid, + "st_gid": gid, + } raise FuseOSError(errno.ENOENT) + def load_statuses(self): + res = requests.get(api_url) + res.raise_for_status() + stats = res.json() + logging.debug(f"Status: ${stats['id']}") + self.statuses = [ + Status(s["object"]["id"].split("/")[-1], s["object"]["content"]) + for s in stats["orderedItems"] + ] + pass + + def list_dir(self) -> list[str]: + return [s.id for s in self.statuses] + def readdir(self, path, fh): - if path == '/': - return ['.', '..'] + [file[1:] for file in self.files.keys()] - else: + dir_entries = [] + if path != "/": raise FuseOSError(errno.ENOENT) + dir_entries = [".", ".."] + if not self.statuses: + self.load_statuses() + dir_entries += self.list_dir() + # + [file[1:] for file in self.files.keys()] + # else: + return dir_entries def open(self, path, flags): - if path not in self.files: - raise FuseOSError(errno.ENOENT) - return 0 + # raise FuseOSError(errno.ENOENT) + # if path not in self.files: + self.fd += 1 + return self.fd def read(self, path, size, offset, fh): - if path in self.files: - return self.files[path][offset:offset + size] + found = next(s for s in self.statuses if "/" + s.id == path) + if found: + return found.content.encode("utf8") raise FuseOSError(errno.ENOENT) + # if path in self.files: + # return self.files[path][offset:offset + size] + def main(mountpoint): - FUSE(HelloWorld(), mountpoint, nothreads=True, foreground=True) + try: + myfuse = FUSE(HelloWorld(), mountpoint, nothreads=True, foreground=True) + except: + fuse_exit() + raise + -if __name__ == '__main__': +if __name__ == "__main__": + logging.basicConfig(level=logging.DEBUG) if len(sys.argv) != 2: - print(f'Usage: {sys.argv[0]} <mountpoint>') + print(f"Usage: {sys.argv[0]} <mountpoint>") sys.exit(1) main(sys.argv[1]) |