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

   1 import socket
   2 import sys
   3 from argparse import ArgumentParser
   4 from pathlib import Path
   5 from subprocess import run
   6 from traceback import print_exception
   7 from urllib.parse import urlunsplit
   8
   9 from falcon.asgi import App as FalconApp
  10 from falcon.constants import MEDIA_HTML
  11 from jinja2 import Environment, FileSystemLoader, select_autoescape
  12 from uvicorn import Config, Server
  13
  14 from .hubapp import RootApp, HubApp
  15
  16
  17 class App(FalconApp):
  18     def __init__(self, secret, **kwargs):
  19         kwargs.setdefault("media_type", MEDIA_HTML)
  20         super().__init__(**kwargs)
  21         self.base_dir = Path(__file__).parents[1] / "webroot"
  22         self.env = Environment(
  23             loader=FileSystemLoader(self.base_dir),
  24             autoescape=select_autoescape(),
  25             extensions=["hub.utils.StaticTag"],
  26         )
  27         self.hubapps = {"root": RootApp(self, self.base_dir, "/derp", secret)}
  28         for base_dir in self.base_dir.iterdir():
  29             if not base_dir.is_dir() or RootApp.is_ignored_filename(base_dir):
  30                 continue
  31             self.hubapps[base_dir.name] = HubApp(
  32                 self, base_dir, base_uri=f"/derp/{base_dir.name}"
  33             )
  34         self.add_error_handler(Exception, self.print_exception)
  35
  36     async def print_exception(self, req, resp, ex, params):
  37         print_exception(*sys.exc_info())
  38
  39
  40 class HubServer(Server):
  41     def __init__(self, *args, browser, **kwargs):
  42         self.browser = browser
  43         super().__init__(*args, **kwargs)
  44
  45     async def startup(self, sockets: list[socket.socket] | None = None) -> None:
  46         await super().startup(sockets)
  47         root_app = self.config.loaded_app.app.hubapps["root"]
  48         print("Secret:", root_app.secret)
  49         for uri, file in root_app.files_per_uris.items():
  50             if file.name == "index.html":
  51                 break
  52         else:
  53             raise ValueError("Root page not found!")
  54         host, port, ssl = self.config.host, self.config.port, bool(self.config.ssl)
  55         if port and port != (80, 443)[ssl]:
  56             host = f"{host}:{port}"
  57         url = urlunsplit((f"http{'s'[:ssl]}", host, uri, "", ""))
  58         print("URL:", url)
  59         if self.browser:
  60             run([self.browser, url])
  61
  62
  63 async def main():
  64     ap = ArgumentParser()
  65     ap.add_argument("--secret")
  66     ap.add_argument("--browser", default="xdg-open")
  67     args = ap.parse_args()
  68     app = App(args.secret)
  69     await app.hubapps["root"].setup()
  70     config = Config(app, port=5000, log_level="info")
  71     config.setup_event_loop()
  72     hs = HubServer(config, browser=args.browser)
  73     await hs.serve()