docs/server.py

"""
NAVA Demo Server - serves static files + /api/list?dir=xxx for video listing
"""
import os
import json
import urllib.parse
import http.server as _hs
from http.server import HTTPServer, SimpleHTTPRequestHandler

# Some internal Python builds (Secure_SimpleHTTP) refuse all non-whitelisted
# clients with HTTP 403. We open the whitelist to the full IPv4 range so
# personal machines on the same internal network can access this demo.
if hasattr(_hs, "NET_WHITELIST"):
    _hs.NET_WHITELIST = ["0.0.0.0/0"]
if hasattr(_hs, "IP_WHITELIST"):
    _hs.IP_WHITELIST = []
if hasattr(_hs, "ACL_ON"):
    _hs.ACL_ON = True

DEMO_DIR = os.path.dirname(os.path.abspath(__file__))


class DemoHandler(SimpleHTTPRequestHandler):
    def __init__(self, *args, **kwargs):
        super().__init__(*args, directory=DEMO_DIR, **kwargs)

    def do_GET(self):
        if self.path.startswith("/api/list"):
            self.handle_list()
        else:
            # Support Range requests for video seeking
            range_header = self.headers.get('Range')
            if range_header and not self.path.startswith("/api/"):
                self.handle_range(range_header)
            else:
                super().do_GET()

    def handle_range(self, range_header):
        """Handle HTTP Range requests for video seeking."""
        path = urllib.parse.unquote(urllib.parse.urlparse(self.path).path)
        if path.startswith('/'):
            path = path[1:]
        fpath = os.path.join(DEMO_DIR, path)
        if not os.path.isfile(fpath):
            self.send_error(404)
            return
        file_size = os.path.getsize(fpath)
        # Parse range
        try:
            range_spec = range_header.replace('bytes=', '')
            start_str, end_str = range_spec.split('-')
            start = int(start_str) if start_str else 0
            end = int(end_str) if end_str else file_size - 1
        except Exception:
            start, end = 0, file_size - 1
        end = min(end, file_size - 1)
        length = end - start + 1
        # Guess content type
        import mimetypes
        ctype, _ = mimetypes.guess_type(fpath)
        if not ctype:
            ctype = 'application/octet-stream'
        self.send_response(206)
        self.send_header('Content-Type', ctype)
        self.send_header('Content-Range', f'bytes {start}-{end}/{file_size}')
        self.send_header('Content-Length', str(length))
        self.send_header('Accept-Ranges', 'bytes')
        self.end_headers()
        with open(fpath, 'rb') as f:
            f.seek(start)
            self.wfile.write(f.read(length))

    def handle_list(self):
        parsed = urllib.parse.urlparse(self.path)
        params = urllib.parse.parse_qs(parsed.query)
        subdir = params.get("dir", [""])[0]
        target = os.path.join(DEMO_DIR, subdir)
        if not os.path.isdir(target):
            files = []
        else:
            import subprocess
            items = []
            for f in sorted(os.listdir(target)):
                if not f.endswith(".mp4"):
                    continue
                fpath = os.path.join(target, f)
                try:
                    r = subprocess.run(
                        ['ffprobe', '-v', 'error', '-select_streams', 'v:0',
                         '-show_entries', 'stream=width,height', '-of', 'csv=p=0', fpath],
                        capture_output=True, text=True, timeout=5)
                    w, h = r.stdout.strip().split(',')
                    items.append({"name": f, "w": int(w), "h": int(h)})
                except Exception:
                    items.append({"name": f, "w": 1280, "h": 720})
            # Sort: portrait first
            items.sort(key=lambda x: (0 if x["h"] > x["w"] else 1, x["name"]))
            files = items
        body = json.dumps(files).encode()
        self.send_response(200)
        self.send_header("Content-Type", "application/json")
        self.send_header("Content-Length", str(len(body)))
        self.end_headers()
        self.wfile.write(body)

    def log_message(self, format, *args):
        pass


if __name__ == "__main__":
    import argparse
    import socket
    parser = argparse.ArgumentParser()
    parser.add_argument("--port", "-p", type=int, default=8889)
    parser.add_argument("--host", default=None,
                        help="Bind host. Default: result of `hostname -i` (so personal machines on the same network can reach it).")
    args = parser.parse_args()

    if args.host is None:
        try:
            host = socket.gethostbyname(socket.gethostname())
        except Exception:
            host = "0.0.0.0"
    else:
        host = args.host

    server = HTTPServer((host, args.port), DemoHandler)
    print(f"NAVA Demo Server: http://{host}:{args.port}")
    server.serve_forever()