build.py

#!/usr/bin/env python3

import asyncio
import sys
from contextlib import redirect_stdout
from pathlib import Path

import uvicorn
from starlette.applications import Starlette
from starlette.middleware.cors import CORSMiddleware
from starlette.staticfiles import *
from starlette.types import Scope

BASE_DIR = Path(__file__).parent
OUTFILE = BASE_DIR / Path("content.js")
PORT = 5000
URL = f"https://0.0.0.0:{PORT}"

try:
    DEV_MODE = sys.argv[1] == "dev"
except IndexError:
    DEV_MODE = False


def main():
    with open(OUTFILE, "w") as file, redirect_stdout(file):
        print(PRELUDE)

        js_paths = list(BASE_DIR.glob("src/*.js"))
        INDEX_JS = BASE_DIR / "src" / "index.js"
        if INDEX_JS in js_paths:
            js_paths.remove(INDEX_JS)
            js_paths.append(INDEX_JS)

        if DEV_MODE:
            print(DEV_MODE_CODE)

        for path in js_paths:
            rel_path = path.relative_to(BASE_DIR)
            if DEV_MODE:
                print(f'makeRequest("{rel_path}", injectOnLoad);')
                print(f'scripts.set("{rel_path}", null);')
            else:
                code = escapejs(path.read_text())
                print(f'scripts.set("{rel_path}", `{code}`);')

        if not DEV_MODE:
            print(f"injectAvailable();")

        print("})();")

    if DEV_MODE:
        app = Starlette()
        app.mount("/", WaitingStaticFiles(directory=BASE_DIR), name="static")
        app.add_middleware(
            CORSMiddleware,
            allow_origins=["*"],
            expose_headers=["*"],
            allow_headers=["*"],
        )

        uvicorn.run(
            app,
            host="0.0.0.0",
            port=PORT,
            ssl_keyfile="./0.0.0.0-key.pem",
            ssl_certfile="./0.0.0.0.pem",
        )


PRELUDE = """\
//
// DO NOT EDIT
// Generated by 'build.py'
//
(function() {
let scripts = new Map();

function injectAvailable() {
    for (let [path, code] of Array.from(scripts)) {
        if (!code) return;
        let elem = document.createElement("script");
        elem.innerHTML = code;    
        document.body.appendChild(elem);
        console.log('[yt-magic] Injected:', elem);
        scripts.delete(path);
    }
}
"""

DEV_MODE_CODE = (
        """\
        let BASE_URL = '%s';
        
        function makeRequest(path, onload, etag = null) {
            let url = `${BASE_URL}/${path}`;
            let req = new XMLHttpRequest();
            req.onload = onload;
            req.open('GET', url);
            req.setRequestHeader('if-modified-since', null);
            req.setRequestHeader('if-none-match', etag);
            req.send();
            console.log('[yt-magic] Sent download request to:', url, ', etag:', etag);
        }      
        
        function injectOnLoad(e) {
            let req = e.target;
            let url = req.responseURL;
            let path = url.slice(BASE_URL.length + 1);    
            console.log('[yt-magic] Downloaded:', path, ', from:', url);      
            scripts.set(path, req.responseText);
            injectAvailable();
            makeRequest(path, () => location.reload(), req.getResponseHeader('etag'));
        }
        """
        % URL
)


def escapejs(value):
    """Hex encode characters for use in JavaScript strings."""
    return str(value).translate(js_escapes)


js_escapes = {
    ord("\\"): "\\u005C",
    ord("'"): "\\u0027",
    ord('"'): "\\u0022",
    ord(">"): "\\u003E",
    ord("<"): "\\u003C",
    ord("&"): "\\u0026",
    ord("="): "\\u003D",
    ord("-"): "\\u002D",
    ord(";"): "\\u003B",
    ord("`"): "\\u0060",
    ord("\u2028"): "\\u2028",
    ord("\u2029"): "\\u2029",
    ord("$"): "\\u0024",
}
# Escape every ASCII character with a value less than 32.
js_escapes.update((ord("%c" % z), "\\u%04X" % z) for z in range(32))


class WaitingFileResponse(FileResponse):
    not_modified: bool
    lookup_path: callable

    async def __call__(self, *args, **kwargs) -> None:
        if self.not_modified:
            etag = self.headers["etag"]

            while True:
                await asyncio.sleep(0.1)

                del self.headers["content-length"]
                del self.headers["last-modified"]
                del self.headers["etag"]

                new_stat = (await self.lookup_path(self.path))[1]
                super().set_stat_headers(new_stat)

                if self.headers["etag"] == etag:
                    continue
                break

        return await super().__call__(*args, **kwargs)


class WaitingStaticFiles(StaticFiles):
    def file_response(
            self,
            full_path: str,
            stat_result: os.stat_result,
            scope: Scope,
            status_code: int = 200,
    ) -> Response:
        method = scope["method"]
        request_headers = Headers(scope=scope)

        response = WaitingFileResponse(
            full_path, status_code=status_code, stat_result=stat_result, method=method
        )
        response.not_modified = super().is_not_modified(
            response.headers, request_headers
        )
        response.lookup_path = super().lookup_path
        return response


if __name__ == "__main__":
    main()