server/mapserv/main.py

   1 #!/usr/bin/env python3
   2
   3 import asyncio
   4 import http.server
   5 import io
   6 import json
   7 import os
   8 import shutil
   9 import socketserver
  10 from html.parser import HTMLParser
  11 from http import HTTPStatus
  12 from http.client import HTTPMessage
  13 from threading import Thread
  14 from typing import Dict, Awaitable, NoReturn, cast, Any, Optional, List, Tuple, AsyncGenerator
  15 from urllib.parse import unquote
  16 from urllib.request import urlopen
  17
  18 UTF_8 = "utf-8"
  19 IO_BLOCK = 16 * 1024
  20
  21
  22 class Config:
  23     __slots__ = \
  24         "port", \
  25         "upstream", \
  26         "dlcache"
  27
  28     def __init__(self) -> None:
  29         self.port = int(os.getenv("PORT", "8000"))
  30         # upstream file url, should end in slash
  31         self.upstream = os.getenv("UPSTREAM", "http://beta.xonotic.org/autobuild-bsp/")
  32         self.dlcache = os.getenv("DLCACHE", os.path.join(os.getcwd(), "dlcache"))
  33
  34
  35 class App:
  36     __slots__ = \
  37         "config", \
  38         "_files"
  39
  40     def __init__(self, config: Config) -> None:
  41         self.config = config
  42         os.makedirs(config.dlcache, exist_ok=True)
  43         self._files: Dict[str, Awaitable[bool]] = {}
  44
  45     async def file_get(self, name: str) -> None:
  46         url = self.config.upstream + name
  47         future = self._files.get(url)
  48         if not future:
  49             print("cache miss")
  50             future = asyncio.get_event_loop().create_future()
  51             self._files[url] = future
  52             out = os.path.join(self.config.dlcache, name)
  53             with open(out, "wb") as f:
  54                 async for buf in fetch(url):
  55                     f.write(buf)
  56             future.set_result(True)
  57         else:
  58             print("using existing")
  59             await future
  60
  61     async def file_wait(self, url: str) -> None:
  62         await self._files[url]
  63
  64     async def ls(self) -> List[str]:
  65         class Parser(HTMLParser):
  66             __slots__ = "paks"
  67
  68             def __init__(self) -> None:
  69                 super().__init__()
  70                 self.paks: List[str] = []
  71
  72             def handle_starttag(self, tag: str, attrs: List[Tuple[str, str]]) -> None:
  73                 if tag == "a":
  74                     file: Optional[str] = next((unquote(v) for (k, v) in attrs if k == "href"), None)
  75                     if file and file.endswith(".pk3"):
  76                         self.paks.append(file)
  77
  78         parser = Parser()
  79         arr = bytearray()
  80         async for buf in fetch(self.config.upstream):
  81             arr.extend(buf)
  82         buf = arr
  83         parser.feed(buf.decode(UTF_8))
  84         return parser.paks
  85
  86
  87 def main() -> None:
  88     config = Config()
  89     app = App(config)
  90
  91     def router() -> Router:
  92         return RouterCombinator(
  93             fetch=FileFetch(),
  94             ls=FileList()
  95         )
  96
  97     class FileFetch(Router):
  98         async def __call__(self, path: str, req: Request) -> Response:
  99             await app.file_get(path)
 100             return {
 101                 "ready": True,
 102             }
 103
 104     class FileList(Router):
 105         async def __call__(self, path: str, req: Request) -> Response:
 106             return {
 107                 "list": await app.ls(),
 108             }
 109
 110     loop = asyncio.get_event_loop()
 111     start_server(config, loop, router())
 112     try:
 113         loop.run_forever()
 114     except KeyboardInterrupt:
 115         pass
 116     finally:
 117         loop.close()
 118
 119
 120 class Request:
 121     pass
 122
 123
 124 Response = Optional[dict]
 125
 126
 127 class Router:
 128     async def __call__(self, path: str, req: Request) -> Response:
 129         pass
 130
 131
 132 class RouterCombinator(Router):
 133     __slots__ = "_routers"
 134
 135     def __init__(self, **kwargs: Any) -> None:
 136         self._routers: Dict[str, Router] = kwargs
 137
 138     async def __call__(self, path: str, req: Request) -> Response:
 139         while True:
 140             args: List[str] = path.split("/", 1)
 141             name = args[0]
 142             rest = args[1] if len(args) > 1 else None
 143             if name:
 144                 break
 145             if not rest:
 146                 return None
 147             path = rest
 148         route = self._routers.get(name, None)
 149         if not route:
 150             return None
 151         return await route(rest or "", req)
 152
 153
 154 async def fetch(url: str) -> AsyncGenerator[bytes, None]:
 155     res = cast(Any, urlopen(url))
 156     msg: HTTPMessage = res.info()
 157     print("msg", msg)
 158     length = msg.get("Content-length")
 159     file_size = int(str(length)) if length else None
 160     print(f"downloading {file_size or '???'} bytes...")
 161     progress = 0
 162     while True:
 163         buf: bytes = res.read(IO_BLOCK)
 164         if not buf:
 165             break
 166         progress += len(buf)
 167         print(f"downloaded {progress}/{file_size or '???'} bytes")
 168         yield buf
 169         await asyncio.sleep(0)
 170
 171
 172 def start_server(config: Config, loop: asyncio.AbstractEventLoop, router: Router) -> None:
 173     async def on_message(req: http.server.BaseHTTPRequestHandler) -> None:
 174         ret = await router(req.path, Request())
 175         if not ret:
 176             ret = {}
 177         req.send_response(HTTPStatus.OK)
 178         req.send_header("Content-Type", "application/json")
 179         s = json.dumps(ret, indent=2).encode(UTF_8)
 180         req.send_header("Content-Length", str(len(s)))
 181         req.end_headers()
 182         f = io.BytesIO()
 183         f.write(s)
 184         f.seek(0)
 185         try:
 186             shutil.copyfileobj(f, req.wfile)
 187         finally:
 188             f.close()
 189
 190     def serve(loop: asyncio.AbstractEventLoop) -> NoReturn:
 191         class ThreadingHTTPServer(socketserver.ThreadingMixIn, http.server.HTTPServer):
 192             pass
 193
 194         class RequestHandler(http.server.BaseHTTPRequestHandler):
 195             def do_GET(self) -> None:
 196                 asyncio.run_coroutine_threadsafe(on_message(self), loop).result()
 197
 198         with ThreadingHTTPServer(("", config.port), RequestHandler) as httpd:
 199             print("serving at port", config.port)
 200             httpd.serve_forever()
 201
 202         assert False, "Unreachable"
 203
 204     server = Thread(target=serve, args=(loop,))
 205     server.daemon = True
 206     server.start()
 207
 208
 209 if __name__ == "__main__":
 210     main()