git.mar77i.info Git - hublib/blob - hub/hubapp.py

   1 from pathlib import Path
   2
   3 from falcon.constants import MEDIA_HTML, MEDIA_JS, MEDIA_TEXT
   4 from falcon.status_codes import HTTP_OK
   5 from jinja2 import Template
   6
   7 from .websocket import WebSocketHub
   8
   9 MEDIA_CSS = "text/css"
  10
  11 class StaticFile:
  12     media_types_per_suffix = {
  13         ".html": MEDIA_HTML,
  14         ".js": MEDIA_JS,
  15         ".css": MEDIA_CSS,
  16         ".txt": MEDIA_TEXT,
  17     }
  18     content: str | None
  19
  20     def __init__(self, path):
  21         self.path = path
  22         self.media_type = self.get_media_type(path)
  23         self.mtime = None
  24         self.content = None
  25
  26     @classmethod
  27     def get_media_type(cls, path):
  28         return cls.media_types_per_suffix[path.suffix]
  29
  30     def get(self):
  31         mtime = self.path.stat().st_mtime
  32         if mtime != self.mtime:
  33             with open(self.path) as fh:
  34                 self.content = fh.read()
  35             self.mtime = mtime
  36         return self.content
  37
  38     def __repr__(self):
  39         return f"<{type(self).__name__} {self.path}>"
  40
  41
  42 class StaticTemplateFile(StaticFile):
  43     TEMPLATE_SUFFIX = ".j2"
  44     content: Template | None
  45
  46     def __init__(self, path, hubapp):
  47         super().__init__(path)
  48         self.hubapp = hubapp
  49         self.context = {"hubapp": self.hubapp}
  50
  51     def get(self) -> str:
  52         mtime = self.path.stat().st_mtime
  53         if mtime != self.mtime:
  54             env = self.hubapp.app.env
  55             self.content = env.get_template(
  56                 str(self.path.relative_to(env.loader.searchpath[0]))
  57             )
  58             self.mtime = mtime
  59         return self.content.render(self.context)
  60
  61     @classmethod
  62     def get_media_type(cls, path):
  63         assert path.suffix == cls.TEMPLATE_SUFFIX
  64         return cls.media_types_per_suffix[path.suffixes[-2]]
  65
  66
  67 class BaseHubApp:
  68     SCAN_FILES_RECURSIVELY = True
  69
  70     def __init__(self, app, base_dir, name=None):
  71         self.app = app
  72         self.base_dir = base_dir
  73         self.name = name if name is not None else base_dir.name
  74         self.app.hubapps[self.name] = self
  75         self.files_per_uris = {
  76             self.uri_from(file.path): file for file in self.scan_files(base_dir)
  77         }
  78         for uri in self.files_per_uris:
  79             app.add_route(uri, self)
  80
  81     @staticmethod
  82     def is_master_uri(uri_tail):
  83         slash = "/"
  84         start = uri_tail.rfind(slash)
  85         pos = uri_tail.find(".", start if start >= 0 else 0)
  86         return (
  87             uri_tail[start + len(slash):pos] == "master"
  88             or "master" in uri_tail[:start].split(slash)
  89         )
  90
  91     def _uri_tail(self, path, suffix):
  92         if isinstance(path, Path):
  93             if path.is_absolute():
  94                 path = path.relative_to(self.base_dir)
  95             uri_tail = str(path)
  96         else:
  97             uri_tail = str(path)
  98         if uri_tail.endswith(suffix):
  99             uri_tail = uri_tail[:-len(suffix)]
 100
 101         if self.is_master_uri(uri_tail):
 102             return self.app.scramble(uri_tail)
 103         elif uri_tail == "index.html":
 104             return ""
 105         return uri_tail
 106
 107     def uri_from(self, path) -> str | Path:
 108         uri_tail = self._uri_tail(path, StaticTemplateFile.TEMPLATE_SUFFIX)
 109         name = self.name
 110         if name == "root":
 111             name = ""
 112         return f"/{name}{'/' if name and uri_tail else ''}{uri_tail}"
 113
 114     def scan_files(self, base_dir):
 115         stack = [base_dir.iterdir()]
 116         while len(stack):
 117             try:
 118                 path = next(stack[-1])
 119             except StopIteration:
 120                 stack.pop()
 121                 continue
 122             if path.is_dir():
 123                 if self.SCAN_FILES_RECURSIVELY:
 124                     stack.append(path.iterdir())
 125             elif not self.is_ignored_filename(path):
 126                 if path.suffix == StaticTemplateFile.TEMPLATE_SUFFIX:
 127                     static_file = StaticTemplateFile(path, self)
 128                 else:
 129                     static_file = StaticFile(path)
 130                 yield static_file
 131
 132     @staticmethod
 133     def is_ignored_filename(path):
 134         return path.name.startswith(".")
 135
 136     async def on_get(self, req, resp):
 137         resource = self.files_per_uris[req.path]
 138         resp.content_type = resource.media_type
 139         resp.data = resource.get().encode()
 140         resp.status = HTTP_OK
 141
 142
 143 class HubApp(BaseHubApp):
 144     def __init__(self, app, base_dir):
 145         super().__init__(app, base_dir)
 146         self.wsh = WebSocketHub(self)
 147         self.ws_uris = {self.uri_from(f"ws_{value}"): value for value in ("client", "master")}
 148         for uri, suffix in self.ws_uris.items():
 149             app.add_route(uri, self.wsh, suffix=suffix)
 150
 151
 152 class RootApp(BaseHubApp):
 153     SCAN_FILES_RECURSIVELY = False
 154
 155     def __init__(self, app, base_dir):
 156         super().__init__(app, base_dir, "root")
 157
 158     @staticmethod
 159     def is_master_uri(uri_tail):
 160         return True