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
   6 from .websocket import WebSocketHub
   7 from .utils import scramble
   8
   9 MEDIA_CSS = "text/css"
  10
  11
  12 class StaticFile:
  13     media_types_per_suffix = {
  14         ".html": MEDIA_HTML,
  15         ".js": MEDIA_JS,
  16         ".css": MEDIA_CSS,
  17         ".txt": MEDIA_TEXT,
  18     }
  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
  45     def __init__(self, path, hubapp):
  46         super().__init__(path)
  47         self.hubapp = hubapp
  48         self.context = {"hubapp": self.hubapp}
  49         self.template = hubapp.app.env.get_template(
  50             str(path.relative_to(hubapp.app.env.loader.searchpath[0]))
  51         )
  52
  53     def get(self):
  54         return self.template.render(self.context)
  55
  56     @classmethod
  57     def get_media_type(cls, path):
  58         assert path.suffix == cls.TEMPLATE_SUFFIX
  59         return cls.media_types_per_suffix[path.suffixes[-2]]
  60
  61
  62 class BaseHubApp:
  63     SCAN_FILES_RECURSIVELY = True
  64
  65     def __init__(self, app, base_dir, name=None):
  66         self.app = app
  67         self.base_dir = base_dir
  68         self.name = name if name is not None else base_dir.name
  69         self.app.hubapps[self.name] = self
  70         self.files_per_uris = {
  71             self.uri_from(file.path): file for file in self.scan_files(base_dir)
  72         }
  73         for uri in self.files_per_uris:
  74             app.add_route(uri, self)
  75
  76     @staticmethod
  77     def is_master_uri(uri_tail):
  78         slash = "/"
  79         start = uri_tail.rfind(slash)
  80         pos = uri_tail.find(".", start if start >= 0 else 0)
  81         return (
  82             uri_tail[start + len(slash):pos] == "master"
  83             or "master" in uri_tail[:start].split(slash)
  84         )
  85
  86     def uri_from(self, path):
  87         if isinstance(path, Path) and path.is_absolute():
  88             uri_tail = str(path.relative_to(self.base_dir))
  89         else:
  90             uri_tail = str(path)
  91         suffix = StaticTemplateFile.TEMPLATE_SUFFIX
  92         if uri_tail.endswith(suffix):
  93             uri_tail = uri_tail[:-len(suffix)]
  94
  95         if self.is_master_uri(uri_tail):
  96             uri_tail = scramble(self.app.secret, uri_tail)
  97         elif uri_tail == "index.html":
  98             uri_tail = ""
  99         name = self.name
 100         if name and uri_tail:
 101             name = f"{name}/"
 102         return f"/{name}{uri_tail}"
 103
 104     def scan_files(self, base_dir):
 105         stack = [base_dir.iterdir()]
 106         while len(stack):
 107             try:
 108                 path = next(stack[-1])
 109             except StopIteration:
 110                 stack.pop()
 111                 continue
 112             if path.is_dir():
 113                 if self.SCAN_FILES_RECURSIVELY:
 114                     stack.append(path.iterdir())
 115             elif not self.is_ignored_filename(path):
 116                 if path.suffix == StaticTemplateFile.TEMPLATE_SUFFIX:
 117                     static_file = StaticTemplateFile(path, self)
 118                 else:
 119                     static_file = StaticFile(path)
 120                 yield static_file
 121
 122     @staticmethod
 123     def is_ignored_filename(path):
 124         return path.name.startswith(".")
 125
 126     async def on_get(self, req, resp):
 127         resource = self.files_per_uris[req.path]
 128         resp.content_type = resource.media_type
 129         resp.data = resource.get().encode()
 130         resp.status = HTTP_OK
 131
 132
 133 class HubApp(BaseHubApp):
 134     def __init__(self, app, base_dir):
 135         super().__init__(app, base_dir)
 136         self.wsh = WebSocketHub(self)
 137         self.ws_uris = {self.uri_from(f"ws_{value}"): value for value in ("client", "master")}
 138         for uri, suffix in self.ws_uris.items():
 139             app.add_route(uri, self.wsh, suffix=suffix)
 140
 141
 142 class RootApp(BaseHubApp):
 143     SCAN_FILES_RECURSIVELY = False
 144
 145     def __init__(self, app, base_dir):
 146         super().__init__(app, base_dir, "")
 147
 148     @staticmethod
 149     def is_master_uri(uri_tail):
 150         return True