Spaces:
Running
Running
| """Minimal HTTP file server with Range support, for local verification only. | |
| Python's stdlib SimpleHTTPRequestHandler does not honor `Range:` requests, but | |
| cdx_toolkit fetches WARC records by byte range, so we need one. Serves files under | |
| a root directory; supports single-range `bytes=start-end` requests (206) which is | |
| all cdx_toolkit issues. | |
| Usage: python scripts/range_server.py <root_dir> <port> | |
| """ | |
| from __future__ import annotations | |
| import os | |
| import sys | |
| from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer | |
| def make_handler(root: str): | |
| root = os.path.abspath(root) | |
| class RangeHandler(BaseHTTPRequestHandler): | |
| def log_message(self, *args): # quiet | |
| pass | |
| def _resolve(self): | |
| path = self.path.split("?", 1)[0].lstrip("/") | |
| full = os.path.abspath(os.path.join(root, path)) | |
| if not full.startswith(root) or not os.path.isfile(full): | |
| return None | |
| return full | |
| def do_HEAD(self): | |
| full = self._resolve() | |
| if full is None: | |
| self.send_error(404) | |
| return | |
| size = os.path.getsize(full) | |
| self.send_response(200) | |
| self.send_header("Content-Length", str(size)) | |
| self.send_header("Accept-Ranges", "bytes") | |
| self.end_headers() | |
| def do_GET(self): | |
| full = self._resolve() | |
| if full is None: | |
| self.send_error(404) | |
| return | |
| size = os.path.getsize(full) | |
| rng = self.headers.get("Range") | |
| if rng and rng.startswith("bytes="): | |
| start_s, _, end_s = rng[len("bytes="):].partition("-") | |
| start = int(start_s) if start_s else 0 | |
| end = int(end_s) if end_s else size - 1 | |
| end = min(end, size - 1) | |
| length = end - start + 1 | |
| with open(full, "rb") as f: | |
| f.seek(start) | |
| data = f.read(length) | |
| self.send_response(206) | |
| self.send_header("Content-Range", f"bytes {start}-{end}/{size}") | |
| self.send_header("Content-Length", str(len(data))) | |
| self.send_header("Accept-Ranges", "bytes") | |
| self.end_headers() | |
| self.wfile.write(data) | |
| else: | |
| with open(full, "rb") as f: | |
| data = f.read() | |
| self.send_response(200) | |
| self.send_header("Content-Length", str(len(data))) | |
| self.send_header("Accept-Ranges", "bytes") | |
| self.end_headers() | |
| self.wfile.write(data) | |
| return RangeHandler | |
| if __name__ == "__main__": | |
| root = sys.argv[1] if len(sys.argv) > 1 else "tests/fixtures/cc" | |
| port = int(sys.argv[2]) if len(sys.argv) > 2 else 8080 | |
| server = ThreadingHTTPServer(("0.0.0.0", port), make_handler(root)) | |
| print(f"serving {os.path.abspath(root)} on :{port} (Range-capable)", flush=True) | |
| server.serve_forever() | |