100 lines
2.9 KiB
Python
100 lines
2.9 KiB
Python
import signal
|
|
import os
|
|
import sys
|
|
from types import FrameType
|
|
from http.server import ThreadingHTTPServer, SimpleHTTPRequestHandler
|
|
import threading
|
|
from typing import override
|
|
from watchdog.observers import Observer
|
|
from zona.builder import ZonaBuilder
|
|
from watchdog.events import FileSystemEventHandler, FileSystemEvent
|
|
from pathlib import Path
|
|
from zona.log import get_logger
|
|
|
|
logger = get_logger()
|
|
|
|
|
|
class QuietHandler(SimpleHTTPRequestHandler):
|
|
@override
|
|
def log_message(self, format, *args): # type: ignore
|
|
pass
|
|
|
|
|
|
class ZonaServer(ThreadingHTTPServer):
|
|
@override
|
|
def handle_error(self, request, client_address): # type: ignore
|
|
_, exc_value = sys.exc_info()[:2]
|
|
if not isinstance(
|
|
exc_value, (BrokenPipeError, ConnectionResetError)
|
|
):
|
|
super().handle_error(request, client_address)
|
|
|
|
|
|
class ZonaReloadHandler(FileSystemEventHandler):
|
|
def __init__(self, builder: ZonaBuilder, output: Path):
|
|
self.builder: ZonaBuilder = builder
|
|
self.output: Path = output.resolve()
|
|
|
|
def _should_ignore(self, event: FileSystemEvent) -> bool:
|
|
path = Path(str(event.src_path)).resolve()
|
|
return (
|
|
self.output in path.parents
|
|
or path == self.output
|
|
or event.is_directory
|
|
)
|
|
|
|
@override
|
|
def on_modified(self, event: FileSystemEvent):
|
|
if not self._should_ignore(event):
|
|
logger.info(f"Modified: {event.src_path}, rebuilding...")
|
|
self.builder.build()
|
|
|
|
@override
|
|
def on_created(self, event: FileSystemEvent):
|
|
if not self._should_ignore(event):
|
|
logger.info(f"Modified: {event.src_path}, rebuilding...")
|
|
self.builder.build()
|
|
|
|
|
|
def run_http_server(dir: Path, host: str = "localhost", port: int = 8000):
|
|
os.chdir(dir)
|
|
handler = QuietHandler
|
|
httpd = ZonaServer(
|
|
server_address=(host, port), RequestHandlerClass=handler
|
|
)
|
|
logger.info(f"Serving {dir} at http://{host}:{port}")
|
|
logger.info(f"Exit with <c-c>")
|
|
httpd.serve_forever()
|
|
|
|
|
|
def serve(
|
|
root: Path | None = None,
|
|
output: Path | None = None,
|
|
host: str = "localhost",
|
|
port: int = 8000,
|
|
):
|
|
builder = ZonaBuilder(root, output)
|
|
builder.build()
|
|
if output is None:
|
|
output = builder.layout.output
|
|
if root is None:
|
|
root = builder.layout.root
|
|
|
|
server_thread = threading.Thread(
|
|
target=run_http_server, args=(output, host, port), daemon=True
|
|
)
|
|
server_thread.start()
|
|
|
|
event_handler = ZonaReloadHandler(builder, output)
|
|
observer = Observer()
|
|
observer.schedule(event_handler, path=str(root), recursive=True)
|
|
observer.start()
|
|
|
|
def shutdown_handler(_a: int, _b: FrameType | None):
|
|
logger.info("Shutting down...")
|
|
observer.stop()
|
|
|
|
signal.signal(signal.SIGINT, shutdown_handler)
|
|
signal.signal(signal.SIGTERM, shutdown_handler)
|
|
|
|
observer.join()
|