Preview server

#!/usr/bin/env python3
"""Build and serve the local notebook site with no Python dependencies.

The generated page loads pinned MathJax at runtime for TeX typesetting and
retains readable TeX when that browser dependency is unavailable.
"""

import argparse
import hashlib
import json
import os
import re
import subprocess
import sys
from functools import partial
from http.server import SimpleHTTPRequestHandler, ThreadingHTTPServer
from pathlib import Path
from typing import Optional, Sequence
from urllib.parse import urlsplit


ROOT = Path(__file__).resolve().parents[1]
DEFAULT_HOST = "127.0.0.1"
DEFAULT_PORT = 8765
PREVIEW_PATH = "/previews/index.html"
BUILD_PROVENANCE = ROOT / "previews" / "build_provenance.json"

UPSTREAM_STEPS = (
    ("build_chronological_index.py",),
    ("build_curriculum_notebooks.py",),
    ("execute_notebooks.py", "--write"),
)
RENDER_STEP = ("render_field_notebook_preview.py",)


class NotebookSiteHandler(SimpleHTTPRequestHandler):
    """Serve repository files with a notebook landing page and no caching."""

    def end_headers(self) -> None:
        self.send_header(
            "Cache-Control",
            "no-store, no-cache, must-revalidate, max-age=0",
        )
        self.send_header("Pragma", "no-cache")
        self.send_header("Expires", "0")
        super().end_headers()

    def send_head(self):  # type: ignore[no-untyped-def]
        requested = urlsplit(self.path).path
        exact_routes = {
            "/": PREVIEW_PATH,
            "/book": "/output/pdf/the-evidence-path-readers-edition.pdf",
            "/python": "/previews/source/index.html",
            "/systems": "/previews/systems/index.html",
            "/research": "/previews/research/index.html",
            "/notebooks": "/previews/notebooks/index.html",
            "/research/field_notebook": "/previews/read/prologue.html",
            "/research/field_notebook/": "/previews/read/prologue.html",
        }
        target = exact_routes.get(requested)
        if target is None:
            patterns = (
                (r"^/read/([a-z0-9-]+)$", "/previews/read/{}.html"),
                (r"^/systems/([a-z0-9-]+)$", "/previews/systems/{}.html"),
                (r"^/research/([a-z0-9-]+)$", "/previews/research/{}.html"),
                (r"^/notebooks/view/([a-z0-9-]+)$", "/previews/notebooks/{}.html"),
                (r"^/python/([a-z0-9-]+)$", "/previews/source/{}.html"),
            )
            for pattern, destination in patterns:
                match = re.fullmatch(pattern, requested)
                if match:
                    target = destination.format(match.group(1))
                    break
        if target is not None:
            self.path = target
        return super().send_head()


class NotebookSiteServer(ThreadingHTTPServer):
    """Threaded development server that does not hold shutdown on requests."""

    allow_reuse_address = True
    daemon_threads = True


def _environment(root: Path) -> dict:
    environment = os.environ.copy()
    source_path = str(root / "src")
    existing_pythonpath = environment.get("PYTHONPATH")
    environment["PYTHONPATH"] = (
        os.pathsep.join((source_path, existing_pythonpath))
        if existing_pythonpath
        else source_path
    )
    environment["PYTHONDONTWRITEBYTECODE"] = "1"
    return environment


def _digest(path: Path) -> str:
    return hashlib.sha256(path.read_bytes()).hexdigest()


def run_build_step(
    step: Sequence[str],
    root: Path = ROOT,
    environment: Optional[dict] = None,
) -> dict:
    """Run one upstream step and return a stable, repository-relative record."""

    environment = environment or _environment(root)
    command = [sys.executable, str(root / "scripts" / step[0]), *step[1:]]
    result = subprocess.run(
        command,
        cwd=root,
        env=environment,
        check=False,
        capture_output=True,
        text=True,
    )
    if result.stdout:
        print(result.stdout, end="", flush=True)
    if result.stderr:
        print(result.stderr, end="", file=sys.stderr, flush=True)
    if result.returncode:
        raise subprocess.CalledProcessError(
            result.returncode,
            command,
            output=result.stdout,
            stderr=result.stderr,
        )
    return {
        "id": Path(step[0]).stem,
        "script": f"scripts/{step[0]}",
        "arguments": list(step[1:]),
        "returncode": result.returncode,
        "stdout": result.stdout,
        "stderr": result.stderr,
    }


def write_build_provenance(records: Sequence[dict], root: Path = ROOT) -> Path:
    """Write deterministic build evidence before the self-rendering step."""

    scripts = tuple(root / record["script"] for record in records)
    artifacts = (
        root / "research" / "chronological_index.md",
        *sorted((root / "notebooks").glob("*.ipynb")),
    )
    payload = {
        "schema_version": 1,
        "steps": list(records),
        "source_sha256": {
            path.relative_to(root).as_posix(): _digest(path) for path in scripts
        },
        "artifact_sha256": {
            path.relative_to(root).as_posix(): _digest(path) for path in artifacts
        },
    }
    target = root / "previews" / "build_provenance.json"
    target.parent.mkdir(parents=True, exist_ok=True)
    target.write_text(
        json.dumps(payload, ensure_ascii=False, indent=2, sort_keys=True) + "\n",
        encoding="utf-8",
    )
    print(target.relative_to(root), flush=True)
    return target


def rebuild_site(root: Path = ROOT) -> None:
    """Regenerate, execute, record, and render artifacts in stable order."""

    environment = _environment(root)
    records = [run_build_step(step, root, environment) for step in UPSTREAM_STEPS]
    write_build_provenance(records, root)

    command = [sys.executable, str(root / "scripts" / RENDER_STEP[0])]
    subprocess.run(command, cwd=root, env=environment, check=True)


def create_server(host: str, port: int, root: Path = ROOT) -> NotebookSiteServer:
    """Create a server rooted at the repository without changing cwd."""

    handler = partial(NotebookSiteHandler, directory=str(root))
    return NotebookSiteServer((host, port), handler)


def site_url(host: str, port: int) -> str:
    """Return the root reader URL, including IPv6 brackets when needed."""

    display_host = f"[{host}]" if ":" in host and not host.startswith("[") else host
    return f"http://{display_host}:{port}/"


def serve(host: str, port: int, root: Path = ROOT) -> int:
    """Serve until interrupted, always closing the listening socket."""

    server = create_server(host, port, root)
    actual_port = int(server.server_address[1])
    print(site_url(host, actual_port), flush=True)
    try:
        server.serve_forever()
    except KeyboardInterrupt:
        print("\nNotebook site stopped.", flush=True)
    finally:
        server.server_close()
    return 0


def port_number(value: str) -> int:
    port = int(value)
    if not 0 <= port <= 65535:
        raise argparse.ArgumentTypeError("port must be between 0 and 65535")
    return port


def parse_args(argv: Optional[Sequence[str]] = None) -> argparse.Namespace:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--host", default=DEFAULT_HOST, help="bind host")
    parser.add_argument(
        "--port",
        type=port_number,
        default=DEFAULT_PORT,
        help="bind port",
    )
    build_mode = parser.add_mutually_exclusive_group()
    build_mode.add_argument(
        "--no-build",
        action="store_true",
        help="serve existing artifacts without rebuilding them",
    )
    build_mode.add_argument(
        "--build-only",
        action="store_true",
        help="rebuild the complete site and exit without opening a server",
    )
    return parser.parse_args(argv)


def main(argv: Optional[Sequence[str]] = None) -> int:
    arguments = parse_args(argv)
    if not arguments.no_build:
        rebuild_site()
    if arguments.build_only:
        return 0
    return serve(arguments.host, arguments.port)


if __name__ == "__main__":
    raise SystemExit(main())