git.mar77i.info Git - hublib/blob - hub/static.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 WebSocketApp
   8
   9 MEDIA_CSS = "text/css"
  10
  11
  12 class StaticFile:
  13     """
  14     Basic static file wrapper.
  15     """
  16     media_types_per_suffix = {
  17         ".html": MEDIA_HTML,
  18         ".js": MEDIA_JS,
  19         ".css": MEDIA_CSS,
  20         ".txt": MEDIA_TEXT,
  21     }
  22     content: str | None
  23
  24     def __init__(self, path):
  25         self.path = path
  26         self.name = path.name
  27         self.media_type = self.get_media_type(path)
  28         self.mtime = None
  29         self.content = None
  30
  31     @classmethod
  32     def get_media_type(cls, path):
  33         return cls.media_types_per_suffix[path.suffix]
  34
  35     def get(self):
  36         mtime = self.path.stat().st_mtime
  37         if mtime != self.mtime:
  38             with open(self.path) as fh:
  39                 self.content = fh.read()
  40             self.mtime = mtime
  41         return self.content
  42
  43     def __repr__(self):
  44         return f"<{type(self).__name__} {self.path}>"
  45
  46
  47 class TreeFileApp:
  48     """
  49     Map a directory tree base_dir to a base_uri and serve it statically.
  50     Map index.html files to their relative roots:
  51     index.html to "/" (root uri) to index.html and
  52     a/index.html to "/a" (directory uri)
  53     """
  54     @staticmethod
  55     def is_ignored_filename(path):
  56         return path.name.startswith(".")
  57
  58     @classmethod
  59     def scan_files(cls, base_dir):
  60         stack = [base_dir.iterdir()]
  61         while len(stack):
  62             try:
  63                 path = next(stack[-1])
  64             except StopIteration:
  65                 stack.pop()
  66                 continue
  67             if path.is_dir():
  68                 stack.append(path.iterdir())
  69             elif not cls.is_ignored_filename(path):
  70                 yield path
  71
  72     def get_file(self, path: Path) -> StaticFile:
  73         return StaticFile(path)
  74
  75     def uri_tail(self, path: Path) -> str:
  76         """
  77         Return the "local" path, relative to self.base_dir, if applicable
  78         """
  79         assert isinstance(path, Path)
  80         if path.is_absolute():
  81             path = path.relative_to(self.base_dir)
  82         return str(path)
  83
  84     def uri(self, path: Path) -> str:
  85         uri_tail = self.uri_tail(path)
  86         if uri_tail == "index.html":
  87             return self.base_uri or "/"
  88         index_suffix = "/index.html"
  89         if uri_tail.endswith(index_suffix):
  90             uri_tail = uri_tail[:-len(index_suffix)]
  91         return f"{self.base_uri}/{uri_tail.lstrip('/')}"
  92
  93     @staticmethod
  94     def create_name(root_base_uri, base_uri):
  95         if base_uri.startswith(root_base_uri):
  96             base_uri = base_uri[len(root_base_uri):]
  97         return base_uri.replace("/", ".").strip(".")
  98
  99     def __init__(self, root, base_dir, base_uri="/", name=None):
 100         self.root = root
 101         self.base_dir = base_dir
 102         self.base_uri = base_uri.rstrip("/")
 103         self.name = name or self.create_name(self.root.base_uri, self.base_uri)
 104         assert self.name, self.name
 105         self.files_per_uris = {}
 106         for path in self.scan_files(base_dir):
 107             static_file = self.get_file(path)
 108             uri = self.uri(static_file.path)
 109             self.files_per_uris[uri] = static_file
 110             root.add_route(uri, self)
 111
 112     async def on_get(self, req, resp):
 113         resource = self.files_per_uris[req.path]
 114         resp.content_type = resource.media_type
 115         resp.data = resource.get().encode()
 116         resp.status = HTTP_OK
 117
 118
 119 class StaticTemplateFile(StaticFile):
 120     TEMPLATE_SUFFIX = ".j2"
 121     content: Template | None
 122
 123     def __init__(self, path, hubapp):
 124         super().__init__(path)
 125         self.name = path.stem
 126         self.hubapp = hubapp
 127         self.context = {"static_file": self}
 128
 129     def get(self) -> str:
 130         mtime = self.path.stat().st_mtime
 131         if mtime != self.mtime:
 132             env = self.hubapp.root.env
 133             self.content = env.get_template(
 134                 str(self.path.relative_to(env.loader.searchpath[0]))
 135             )
 136             self.mtime = mtime
 137         return self.content.render(self.context)
 138
 139     @classmethod
 140     def get_media_type(cls, path):
 141         assert path.suffix == cls.TEMPLATE_SUFFIX
 142         return cls.media_types_per_suffix[path.suffixes[-2]]
 143
 144
 145 class TemplateTreeFileApp(TreeFileApp):
 146     def get_file(self, path: Path) -> StaticFile:
 147         if path.suffix == StaticTemplateFile.TEMPLATE_SUFFIX:
 148             return StaticTemplateFile(path, self)
 149         return super().get_file(path)
 150
 151     def uri_tail(self, path: Path) -> str:
 152         uri_tail = super().uri_tail(path)
 153         if uri_tail.endswith(StaticTemplateFile.TEMPLATE_SUFFIX):
 154             uri_tail = uri_tail[:-len(StaticTemplateFile.TEMPLATE_SUFFIX)]
 155         return uri_tail
 156
 157
 158 class HubApp(TemplateTreeFileApp):
 159     def __init__(self, app, base_dir, base_uri="/"):
 160         super().__init__(app, base_dir, base_uri)
 161         self.ws_app = WebSocketApp(self)
 162         for suffix in ("client", "master"):
 163             uri = self.uri(Path(f"ws_{suffix}"))
 164             self.files_per_uris[uri] = self.ws_app
 165             app.add_route(uri, self.ws_app, suffix=suffix)
 166
 167     @staticmethod
 168     def is_master_uri(path: Path) -> bool:
 169         assert isinstance(path, Path)
 170         basename = path.name
 171         pos = basename.find(".")
 172         if pos != -1:
 173             basename = basename[:pos]
 174         return basename in ("master", "ws_master")
 175
 176     def uri(self, path: Path) -> str:
 177         uri = super().uri(path)
 178         if not self.is_master_uri(path):
 179             return uri
 180         pos = uri.rstrip("/").rfind("/")
 181         assert pos >= 0
 182         scrambled_uri = self.root.scramble(uri)
 183         return f"{uri[:pos]}/{scrambled_uri}"