Write very basic ActivityPub FUSE

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])