Spaces:
Running
Running
| #!/usr/bin/env python3 | |
| import argparse | |
| import os | |
| import re | |
| from http.server import SimpleHTTPRequestHandler, ThreadingHTTPServer | |
| class RangeRequestHandler(SimpleHTTPRequestHandler): | |
| protocol_version = "HTTP/1.1" | |
| def send_head(self): | |
| path = self.translate_path(self.path) | |
| if os.path.isdir(path) or "Range" not in self.headers: | |
| response = super().send_head() | |
| if response and not os.path.isdir(path): | |
| self._range = None | |
| return response | |
| try: | |
| file_handle = open(path, "rb") | |
| except OSError: | |
| self.send_error(404, "File not found") | |
| return None | |
| file_size = os.fstat(file_handle.fileno()).st_size | |
| match = re.fullmatch(r"bytes=(\d*)-(\d*)", self.headers["Range"].strip()) | |
| if not match: | |
| file_handle.close() | |
| self.send_error(400, "Invalid Range header") | |
| return None | |
| start_text, end_text = match.groups() | |
| if not start_text: | |
| suffix_length = int(end_text) | |
| start = max(file_size - suffix_length, 0) | |
| end = file_size - 1 | |
| else: | |
| start = int(start_text) | |
| end = int(end_text) if end_text else file_size - 1 | |
| if start >= file_size or start > end: | |
| file_handle.close() | |
| self.send_response(416) | |
| self.send_header("Content-Range", f"bytes */{file_size}") | |
| self.send_header("Content-Length", "0") | |
| self.end_headers() | |
| return None | |
| end = min(end, file_size - 1) | |
| content_length = end - start + 1 | |
| self._range = (start, end) | |
| self.send_response(206) | |
| self.send_header("Content-Type", self.guess_type(path)) | |
| self.send_header("Accept-Ranges", "bytes") | |
| self.send_header("Content-Range", f"bytes {start}-{end}/{file_size}") | |
| self.send_header("Content-Length", str(content_length)) | |
| self.send_header("Last-Modified", self.date_time_string(os.path.getmtime(path))) | |
| self.end_headers() | |
| file_handle.seek(start) | |
| return file_handle | |
| def copyfile(self, source, outputfile): | |
| if not getattr(self, "_range", None): | |
| return super().copyfile(source, outputfile) | |
| remaining = self._range[1] - self._range[0] + 1 | |
| while remaining: | |
| chunk = source.read(min(64 * 1024, remaining)) | |
| if not chunk: | |
| break | |
| outputfile.write(chunk) | |
| remaining -= len(chunk) | |
| def main(): | |
| parser = argparse.ArgumentParser(description="Serve static files with HTTP Range support.") | |
| parser.add_argument("--bind", default="0.0.0.0") | |
| parser.add_argument("--port", type=int, default=8123) | |
| parser.add_argument("--directory", default=".") | |
| args = parser.parse_args() | |
| handler = lambda *handler_args, **handler_kwargs: RangeRequestHandler( | |
| *handler_args, directory=args.directory, **handler_kwargs | |
| ) | |
| server = ThreadingHTTPServer((args.bind, args.port), handler) | |
| print(f"Serving {os.path.abspath(args.directory)} on http://{args.bind}:{args.port}") | |
| server.serve_forever() | |
| if __name__ == "__main__": | |
| main() | |