Spaces:
Running
Running
File size: 3,046 Bytes
15eb4b9 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 | """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()
|