#!/usr/bin/env python3
"""Zero-dependency HTTP/1.1 static file server.

This server demonstrates a minimal but production-ready pattern for:
- raw socket creation and listening on IPv4 TCP sockets
- manual HTTP request parsing from byte streams
- per-connection threading for concurrency
- safe static file serving with directory traversal protections
- graceful shutdown using socket timeouts and a stop event

The project intentionally uses only the Python standard library.
"""

from __future__ import annotations

import argparse
import html
import mimetypes
import os
import re
import socket
import sys
import threading
import time
from dataclasses import dataclass, field
from datetime import datetime, timezone
from pathlib import Path
from typing import Callable, Dict, Iterable, List, Optional, Tuple
from urllib.parse import parse_qs, unquote, urlsplit, urlparse

DEFAULT_HOST = "0.0.0.0"
DEFAULT_PORT = 8080
READ_CHUNK_SIZE = 4096
MAX_HEADER_BYTES = 64 * 1024
CONNECTION_TIMEOUT_SECONDS = 5.0


@dataclass
class HTTPRequest:
    """Normalized request object used by the router."""

    method: str
    path: str
    uri_params: Dict[str, str] = field(default_factory=dict)
    query_params: Dict[str, str] = field(default_factory=dict)
    headers: Dict[str, str] = field(default_factory=dict)
    body: bytes = b""


@dataclass
class HTTPResponse:
    """Simple HTTP response container."""

    status_code: int
    reason: str
    headers: Dict[str, str] = field(default_factory=dict)
    body: bytes = b""

    def to_bytes(self) -> bytes:
        """Serialize the response into a valid HTTP/1.1 payload."""
        header_lines = [
            f"HTTP/1.1 {self.status_code} {self.reason}",
            "Server: ZeroDepHTTP/1.0",
            f"Date: {http_date()}",
            f"Content-Length: {len(self.body)}",
            "Connection: close",
        ]
        for key, value in self.headers.items():
            header_lines.append(f"{key}: {value}")
        header_lines.append("")
        header_lines.append("")
        return "\r\n".join(header_lines).encode("latin-1") + self.body


@dataclass(frozen=True)
class Request:
    """Represents a parsed HTTP request."""

    method: str
    target: str
    http_version: str
    headers: Dict[str, str]
    body: bytes = b""


@dataclass
class Response:
    """Represents an HTTP response that will be framed into raw bytes."""

    status_code: int
    reason: str
    headers: Dict[str, str] = field(default_factory=dict)
    body: bytes = b""

    def to_bytes(self) -> bytes:
        """Convert the response object into a valid HTTP/1.1 response."""
        header_lines = [
            f"HTTP/1.1 {self.status_code} {self.reason}",
            "Server: ZeroDepHTTP/1.0",
            f"Date: {http_date()}",
            f"Content-Length: {len(self.body)}",
            "Connection: close",
        ]

        for key, value in self.headers.items():
            header_lines.append(f"{key}: {value}")

        header_lines.append("")
        header_lines.append("")
        response = "\r\n".join(header_lines).encode("latin-1")
        return response + self.body


class HTTPRequestError(ValueError):
    """Raised when the raw socket stream does not form a valid HTTP request."""


class Router:
    """Standard-library router with dynamic URI parameter matching."""

    def __init__(self, static_dir: str):
        self.static_dir = Path(static_dir).resolve()
        self.dynamic_routes: List[Tuple[str, re.Pattern, List[str], Callable]] = []

    def add_route(self, method: str, path_pattern: str, handler: Callable):
        method = method.upper()
        param_names = re.findall(r"<([^>]+)>", path_pattern)
        regex_pattern = re.sub(r"<[^>]+>", r"([^/]+)", path_pattern)
        compiled_regex = re.compile(f"^{regex_pattern}$")
        self.dynamic_routes.append((method, compiled_regex, param_names, handler))

    def handle(self, request: HTTPRequest) -> HTTPResponse:
        for method, regex, param_names, handler in self.dynamic_routes:
            if method == request.method:
                match = regex.match(request.path)
                if match:
                    request.uri_params = dict(zip(param_names, match.groups()))
                    return handler(request)
        return self._serve_static(request)

    def _serve_static(self, request: HTTPRequest) -> HTTPResponse:
        if request.method not in {"GET", "HEAD"}:
            return HTTPResponse(405, "Method Not Allowed", body=b"405 Method Not Allowed")

        rel_path = request.path.lstrip("/")
        if not rel_path:
            rel_path = "index.html"

        safe_path = (self.static_dir / rel_path).resolve()
        if not str(safe_path).startswith(str(self.static_dir)):
            return HTTPResponse(403, "Forbidden", body=b"403 Access Denied")

        if not safe_path.exists() or not safe_path.is_file():
            return HTTPResponse(404, "Not Found", body=b"404 Resource Not Found")

        mime_type, _ = mimetypes.guess_type(str(safe_path))
        mime_type = mime_type or "application/octet-stream"
        with open(safe_path, "rb") as fh:
            content = fh.read()
        body = b"" if request.method == "HEAD" else content
        return HTTPResponse(
            status_code=200,
            reason="OK",
            headers={"Content-Type": mime_type},
            body=body,
        )


class ThreadedHTTPServer:
    """Minimal multithreaded HTTP server built on raw TCP sockets.

    This class intentionally avoids third-party frameworks and uses the
    standard library's threading and socket modules to create one thread per
    connection.
    """

    def __init__(self, host: str, port: int, document_root: Path):
        self.host = host
        self.port = port
        self.document_root = document_root.resolve()
        self.router = Router(str(self.document_root))
        self._register_routes()
        self._stop_event = threading.Event()
        self._lock = threading.Lock()
        self._active_threads: set[threading.Thread] = set()
        self._listener = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        self._listener.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
        self._listener.bind((self.host, self.port))
        self._listener.listen(128)

    def _register_routes(self):
        self.router.add_route(
            "GET",
            "/api/health",
            lambda req: HTTPResponse(
                200,
                "OK",
                {"Content-Type": "application/json"},
                b'{"status": "ok"}',
            ),
        )

        def get_user_handler(req: HTTPRequest):
            user_id = req.uri_params.get("user_id")
            fmt = req.query_params.get("format", "basic")
            data = {
                "user_id": user_id,
                "status": "active",
                "format": fmt,
            }
            if fmt == "full":
                payload = '{"user_id": "%s", "status": "active", "format": "%s"}' % (user_id, fmt)
            else:
                payload = '{"user_id": "%s", "status": "active"}' % user_id
            return HTTPResponse(
                200,
                "OK",
                {"Content-Type": "application/json"},
                payload.encode("utf-8"),
            )

        self.router.add_route("GET", "/api/users/<user_id>", get_user_handler)

        def get_order_item(req: HTTPRequest):
            order_id = req.uri_params.get("order_id")
            item_id = req.uri_params.get("item_id")
            text = f"Order: {order_id}, Item: {item_id}"
            return HTTPResponse(200, "OK", {"Content-Type": "text/plain"}, text.encode("utf-8"))

        self.router.add_route("GET", "/api/orders/<order_id>/items/<item_id>", get_order_item)

    def serve_forever(self) -> None:
        """Accept client connections until shutdown is requested."""
        print(
            f"Serving {self.document_root} on http://{self.host}:{self.port}",
            flush=True,
        )
        self._listener.settimeout(0.5)

        while not self._stop_event.is_set():
            try:
                conn, addr = self._listener.accept()
            except socket.timeout:
                continue
            except OSError:
                break

            thread = threading.Thread(
                target=self._handle_client,
                args=(conn, addr),
                daemon=True,
            )
            with self._lock:
                self._active_threads.add(thread)
            thread.start()

        self._listener.close()

    def shutdown(self) -> None:
        """Signal the server to stop accepting new connections."""
        self._stop_event.set()
        try:
            self._listener.shutdown(socket.SHUT_RDWR)
        except OSError:
            pass
        self._listener.close()

    def _handle_client(self, conn: socket.socket, addr: Tuple[str, int]) -> None:
        """Process one client connection from raw bytes to HTTP response."""
        try:
            conn.settimeout(CONNECTION_TIMEOUT_SECONDS)
            raw_request = self._read_client_bytes(conn)
            if not raw_request:
                return

            request = self._parse_http_request(raw_request)
            if request is None:
                response = self._response_from_status(400, "Bad Request", b"Bad request")
            else:
                response = self._dispatch_request(request)

            conn.sendall(response.to_bytes())
        except (BrokenPipeError, ConnectionResetError, OSError, ValueError):
            pass
        except Exception:
            try:
                error_body = b"Internal Server Error"
                conn.sendall(
                    self._response_from_status(500, "Internal Server Error", error_body).to_bytes()
                )
            except Exception:
                pass
        finally:
            conn.close()
            with self._lock:
                self._active_threads.discard(threading.current_thread())

    def _read_client_bytes(self, conn: socket.socket) -> bytes:
        """Read a complete HTTP request from the socket.

        This method accumulates bytes until it sees the header terminator,
        then ensures that any Content-Length body has also been received.
        """
        buffer = bytearray()

        while True:
            chunk = conn.recv(READ_CHUNK_SIZE)
            if not chunk:
                break
            buffer.extend(chunk)

            if b"\r\n\r\n" not in buffer:
                if len(buffer) > MAX_HEADER_BYTES:
                    raise HTTPRequestError("Headers too large")
                continue

            header_end = buffer.find(b"\r\n\r\n")
            header_block = bytes(buffer[: header_end + 4])
            headers = self._parse_headers(header_block)
            content_length = 0
            if "Content-Length" in headers:
                try:
                    content_length = int(headers["Content-Length"])
                except ValueError as exc:  # pragma: no cover - defensive
                    raise HTTPRequestError("Malformed Content-Length header") from exc

            total_needed = header_end + 4 + content_length
            if len(buffer) >= total_needed:
                return bytes(buffer[:total_needed])

            if len(buffer) > MAX_HEADER_BYTES + content_length:
                raise HTTPRequestError("Request too large")

    def _parse_headers(self, raw_headers: bytes) -> Dict[str, str]:
        """Parse an HTTP header block into a case-insensitive dictionary.

        The raw block may include the request line first, followed by the actual
        headers. We ignore the request line when detecting Content-Length and
        other header values.
        """
        headers: Dict[str, str] = {}
        if not raw_headers:
            return headers

        for line in raw_headers.decode("latin-1").split("\r\n"):
            if not line or line.startswith("\r"):
                continue
            if ":" not in line:
                parts = line.split()
                if len(parts) >= 3 and parts[0].upper() in {
                    "GET",
                    "HEAD",
                    "POST",
                    "PUT",
                    "DELETE",
                    "OPTIONS",
                    "PATCH",
                    "TRACE",
                    "CONNECT",
                }:
                    continue
                raise HTTPRequestError(f"Malformed header line: {line!r}")
            key, value = line.split(":", 1)
            headers[key.strip().lower()] = value.strip()
        return headers

    def _parse_http_request(self, raw_request: bytes) -> Optional[Request]:
        """Parse a raw HTTP request into a Request object."""
        if not raw_request:
            return None

        header_block, _, body = raw_request.partition(b"\r\n\r\n")
        lines = header_block.decode("latin-1").split("\r\n")
        if not lines:
            return None

        request_line = lines[0].strip()
        parts = request_line.split()
        if len(parts) != 3:
            return None

        method, target, http_version = parts
        if not method or not target or not http_version:
            return None

        parsed_url = urlparse(target)
        query_params = {k: v[0] for k, v in parse_qs(parsed_url.query).items()}
        path = unquote(parsed_url.path)

        raw_headers = lines[1:]
        headers: Dict[str, str] = {}
        for line in raw_headers:
            if not line:
                continue
            if ":" not in line:
                return None
            name, value = line.split(":", 1)
            headers[name.strip().lower()] = value.strip()

        content_length = headers.get("content-length", "0")
        try:
            content_length_int = int(content_length)
        except ValueError:
            return None

        if len(body) < content_length_int:
            raise HTTPRequestError("Request body incomplete")

        request = Request(
            method=method.upper(),
            target=target,
            http_version=http_version.upper(),
            headers=headers,
            body=body[:content_length_int],
        )
        normalized = HTTPRequest(
            method=request.method,
            path=path,
            query_params=query_params,
            headers=headers,
            body=request.body,
        )
        return request

    def _dispatch_request(self, request: Request) -> Response:
        """Route a parsed request to the correct static file behavior."""
        if request.method not in {"GET", "HEAD"}:
            return self._response_from_status(405, "Method Not Allowed", b"Method Not Allowed")

        if not request.target.startswith("/"):
            return self._response_from_status(400, "Bad Request", b"Bad Request")

        parsed = urlparse(request.target)
        path = unquote(parsed.path)
        http_request = HTTPRequest(
            method=request.method,
            path=path,
            query_params={k: v[0] for k, v in parse_qs(parsed.query).items()},
            headers=request.headers,
            body=request.body,
        )
        dynamic_response = self.router.handle(http_request)
        if dynamic_response.status_code != 404 or path.startswith("/api/"):
            return dynamic_response

        safe_path = self._resolve_path(request.target)
        if safe_path is None:
            return self._response_from_status(403, "Forbidden", b"Forbidden")

        if safe_path.is_dir():
            index_path = safe_path / "index.html"
            if index_path.exists() and index_path.is_file():
                safe_path = index_path
            else:
                listing = self._directory_listing_html(safe_path)
                return Response(
                    status_code=200,
                    reason="OK",
                    headers={"Content-Type": "text/html; charset=utf-8"},
                    body=listing.encode("utf-8") if request.method == "GET" else b"",
                )

        if not safe_path.exists() or not safe_path.is_file():
            return self._response_from_status(404, "Not Found", b"Not Found")

        data = safe_path.read_bytes()
        content_type, _ = mimetypes.guess_type(str(safe_path))
        if content_type is None:
            content_type = "application/octet-stream"

        payload = data if request.method == "GET" else b""
        return Response(
            status_code=200,
            reason="OK",
            headers={"Content-Type": content_type},
            body=payload,
        )

    def _resolve_path(self, target: str) -> Optional[Path]:
        """Return a path that is guaranteed to stay within the document root."""
        parsed = urlsplit(target)
        path = unquote(parsed.path)
        if not path.startswith("/"):
            path = "/" + path

        candidate = (self.document_root / path.lstrip("/")).resolve()
        try:
            candidate.relative_to(self.document_root)
        except ValueError:
            return None
        return candidate

    def _directory_listing_html(self, directory: Path) -> str:
        """Create a simple HTML directory listing for directory requests."""
        entries = [
            '<!DOCTYPE html><html><head><meta charset="utf-8"><title>Index of {}</title></head><body>'.format(
                html.escape(str(directory.relative_to(self.document_root)))
            )
        ]
        entries.append('<h1>Index of /{}</h1>'.format(html.escape(str(directory.relative_to(self.document_root)))))
        entries.append('<ul>')
        for child in sorted(directory.iterdir(), key=lambda p: p.name.lower()):
            href = "/" + str(child.relative_to(self.document_root)).replace('\\', '/')
            entries.append(f'<li><a href="{html.escape(href)}">{html.escape(child.name)}</a></li>')
        entries.append('</ul></body></html>')
        return "".join(entries)

    @staticmethod
    def _response_from_status(status_code: int, reason: str, body: bytes) -> Response:
        """Create a simple response object for error statuses."""
        return Response(
            status_code=status_code,
            reason=reason,
            headers={"Content-Type": "text/plain; charset=utf-8"},
            body=body,
        )


def http_date() -> str:
    """Return RFC 7231 compliant HTTP date and time."""
    dt = datetime.now(timezone.utc)
    return dt.strftime("%a, %d %b %Y %H:%M:%S GMT")


def build_parser() -> argparse.ArgumentParser:
    """Create the CLI parser for the server."""
    parser = argparse.ArgumentParser(
        description="Zero-dependency HTTP/1.1 static file server using only Python's standard library."
    )
    parser.add_argument(
        "--host",
        default=DEFAULT_HOST,
        help="Host interface to bind (default: %(default)s)",
    )
    parser.add_argument(
        "--port",
        type=int,
        default=DEFAULT_PORT,
        help="TCP port to listen on (default: %(default)s)",
    )
    parser.add_argument(
        "--directory",
        default="public",
        help="Directory to serve as static content (default: %(default)s)",
    )
    return parser


def main() -> int:
    """Entry point for the standalone script."""
    parser = build_parser()
    args = parser.parse_args()

    document_root = Path(args.directory).resolve()
    if not document_root.exists() or not document_root.is_dir():
        print(f"Document root does not exist or is not a directory: {document_root}", file=sys.stderr)
        return 2

    server = ThreadedHTTPServer(args.host, args.port, document_root)

    try:
        server.serve_forever()
    except KeyboardInterrupt:
        print("\nReceived shutdown signal; stopping server.", file=sys.stderr)
    finally:
        server.shutdown()

    return 0


if __name__ == "__main__":
    raise SystemExit(main())