from aiohttp import web import sys, os, asyncio, json from asyncio import create_subprocess_exec from urllib.parse import urlparse, unquote as urlunquote, quote as urlquote import argparse """ Todo: use proper http error responses for errors, see aiohttp/http_exceptions.py """ async def is_makeable (path, cwd=".", makefile="makefile"): rpath = os.path.relpath(path, cwd) p = await create_subprocess_exec("make", "--question", "-f", makefile, rpath, cwd = cwd) # retcode 0=file is up to date, 1=file needs remaking, 2=file is not makeable ret = await p.wait() == 1 # print (f"is_makeable {path}: {ret}", file=sys.stderr) return ret async def make (path, cwd=".", makefile="makefile"): rpath = os.path.relpath(path, cwd) p = await create_subprocess_exec("make", "-f", makefile, rpath, cwd = cwd) ret = await p.wait() == 0 # print (f"make {path}: {ret}", file=sys.stderr) return ret textchars = bytearray({7,8,9,10,12,13,27} | set(range(0x20, 0x100)) - {0x7f}) is_binary_string = lambda bytes: bool(bytes.translate(None, textchars)) def is_binary_file (p): """ returns none on ioerror """ try: return not os.path.isdir(p) and is_binary_string(open(p, 'rb').read(1024)) except IOError: return None def editable (path): _, ext = os.path.splitext(path) ext = ext.lower()[1:] print (f"editable? ext {ext}", file=sys.stderr) return ((ext not in ("html", "htm")) and not is_binary_file(path)) # From aiohttp/web_urldispatcher.py def directory_as_html(filepath, directory, prefix, editor_url=None): # returns directory's index as html # sanity check assert os.path.isdir(filepath) # relative_path_to_dir = filepath.relative_to(self._directory) relative_path_to_dir = os.path.relpath(filepath, directory) index_of = "Index of /{}".format(relative_path_to_dir) h1 = "

{}

".format(index_of) index_list = [] dir_index = os.listdir(filepath) for _file in sorted(dir_index): # show file url as relative to static path # rel_path = _file.relative_to(self._directory).as_posix() # rel_path = os.path.relpath(_file, directory) # file_url = prefix + '/' + rel_path fp = os.path.join(filepath, _file) file_url = '/' + urlquote(fp) target = None if editable(fp) and editor_url: file_url = f"{editor_url}#{file_url}" target = "editor" # if file is a directory, add '/' to the end of the name if os.path.isdir(_file): file_name = "{}/".format(_file) else: file_name = _file if target: link_src = '
  • {name}
  • '.format( url=file_url, target="editor", name=file_name) else: link_src = '
  • {name}
  • '.format( url=file_url, name=file_name) index_list.append(link_src) ul = "".format('\n'.join(index_list)) body = "\n{}\n{}\n".format(h1, ul) head_str = "\n{}\n".format(index_of) html = "\n{}\n{}\n".format(head_str, body) return html async def route_get (request): # return FileResponse(filepath, chunk_size=self._chunk_size) path = urlunquote(urlparse(request.rel_url.raw_path.lstrip("/")).path) print ("route_get", path, file=sys.stderr) if path == '': path = '.' if path and await is_makeable(path, makefile=request.app['makefile']): await make(path, makefile=request.app['makefile']) if os.path.exists(path): if os.path.isdir(path): return web.Response(text=directory_as_html(path, '', '', editor_url=request.app['editor']), content_type="text/html") else: return web.FileResponse(path, chunk_size=256*1024) else: return web.Response(text=f"404 on path {path}") async def route_post (request): """ write posted text value to file """ path = urlunquote(urlparse(request.rel_url.raw_path.lstrip("/")).path) print ("route_post", path, file=sys.stderr) if os.path.exists(path) and os.path.isfile(path): data = await request.post() if 'text' in data: text = data['text'] # doing file io inline here is not strictly speaking very async ;) with open(path, "w") as f: f.write(text) return web.Response(text=json.dumps({"response": "ok"}), content_type="application/json") else: return web.Response(text=f"no text") else: return web.Response(text=f"post not allowed on {path}") def main (): ap = argparse.ArgumentParser("make & serve") ap.add_argument("--makefile", default="makefile") ap.add_argument("--editor", default=None) args = ap.parse_args() if sys.platform == 'win32': # Folowing: https://docs.python.org/3/library/asyncio-subprocess.html # On Windows, the default event loop is SelectorEventLoop which does not support subprocesses. # ProactorEventLoop should be used instead. loop = asyncio.ProactorEventLoop() asyncio.set_event_loop(loop) app = web.Application() app['makefile'] = args.makefile app['editor'] = args.editor app.add_routes([web.get('/{make:.*}', route_get)]) app.add_routes([web.post('/{make:.*}', route_post)]) web.run_app(app) if __name__ == "__main__": main()