#!/usr/bin/env python3
"""Serve files from one directory to unauthenticated LAN clients.

The HTTP API intentionally has a small surface:
* GET / (and /list) lists direct children of the share root.
* GET /<dir>/ lists direct children of a nested directory (HTML or JSON).
* GET /<path> downloads a file; GET /<dir> or /<dir>.zip downloads a ZIP archive.

All resolved filesystem targets remain confined to the configured share root.
File and directory opens use O_NOFOLLOW so concurrent symlink swaps cannot
escape the share root after path validation.
"""

from __future__ import annotations

import argparse
from functools import partial
import html
import json
import logging
import mimetypes
import os
from pathlib import Path
import re
import shutil
import signal
import stat
import tempfile
import threading
from http import HTTPStatus
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from typing import Callable, Dict, Iterable, List, Optional, Tuple
from urllib.parse import parse_qs, quote, unquote_to_bytes, urlsplit
import zipfile


DEFAULT_HOST = "0.0.0.0"
DEFAULT_PORT = 18001
COPY_CHUNK_SIZE = 1024 * 1024
MAX_ARCHIVE_FILE_COUNT = 5000
MAX_ARCHIVE_UNCOMPRESSED_BYTES = 512 * 1024 * 1024
MAX_ARCHIVE_DEPTH = 32
MAX_CONCURRENT_ARCHIVES = 2
REQUEST_TIMEOUT_SECONDS = 30

# Keep service internals out of the public share listing/download surface.
RESERVED_SHARE_NAMES = frozenset({
    ".omc",
    "__pycache__",
})

_ARCHIVE_SEMAPHORE = threading.Semaphore(MAX_CONCURRENT_ARCHIVES)
_CONTROL_CHARS = re.compile(r"[\x00-\x1f\x7f]")


class ShareHTTPServer(ThreadingHTTPServer):
    """Threaded HTTP server with safe socket/thread defaults."""

    allow_reuse_address = True
    daemon_threads = True
    request_queue_size = 32

    def get_request(self):  # type: ignore[override]
        request, client_address = super().get_request()
        request.settimeout(REQUEST_TIMEOUT_SECONDS)
        return request, client_address


def is_within_root(candidate: Path, root: Path) -> bool:
    """Return whether an already-resolved path remains inside ``root``."""

    try:
        candidate.relative_to(root)
    except ValueError:
        return False
    return True


def safe_attachment_name(name: str) -> str:
    """Create an ASCII fallback suitable for a Content-Disposition header."""

    return "".join(
        character if 32 <= ord(character) < 127 and character not in '"\\'
        else "_"
        for character in name
    ) or "download"


def content_disposition(filename: str) -> str:
    """Produce a standards-compatible attachment disposition value."""

    fallback = safe_attachment_name(filename)
    encoded = quote(filename, safe="")
    return "attachment; filename=\"{}\"; filename*=UTF-8''{}".format(
        fallback,
        encoded,
    )


def sanitized_log_text(value: object) -> str:
    """Strip control characters so request data cannot rewrite logs."""

    return _CONTROL_CHARS.sub("?", str(value))


def decoded_request_path(raw_path: str) -> Optional[Tuple[str, ...]]:
    """Validate and decode an origin-form request target into path segments.

    ``None`` represents an invalid or intentionally unsupported path. The root
    path is represented by an empty tuple so it can be requested as a ZIP.
    """

    if not raw_path.startswith("/") or raw_path.startswith("//"):
        return None

    try:
        decoded = unquote_to_bytes(raw_path).decode("utf-8", errors="strict")
    except UnicodeDecodeError:
        return None

    if "\x00" in decoded:
        return None

    segments = tuple(segment for segment in decoded.split("/") if segment)
    if any(segment in (".", "..") for segment in segments):
        return None
    if any("\\" in segment or "\x00" in segment for segment in segments):
        return None
    return segments


def _close_fd(file_descriptor: Optional[int]) -> None:
    if file_descriptor is not None and file_descriptor >= 0:
        try:
            os.close(file_descriptor)
        except OSError:
            pass


def open_within_root(
    root: Path,
    segments: Iterable[str],
) -> Optional[Tuple[int, os.stat_result, str]]:
    """Open a path under ``root`` with O_NOFOLLOW and return (fd, stat, kind).

    Intermediate components must be directories. The final target must be a regular
    file or directory. Symlinks are never followed. On success the caller owns
    the returned file descriptor.
    """

    segment_list = list(segments)
    if segment_list and (
        segment_list[0] in RESERVED_SHARE_NAMES or segment_list[0].startswith(".")
    ):
        return None
    flags_base = os.O_RDONLY | os.O_NOFOLLOW | os.O_CLOEXEC
    current_fd = -1
    try:
        current_fd = os.open(
            str(root),
            os.O_RDONLY | os.O_DIRECTORY | os.O_CLOEXEC,
        )

        if not segment_list:
            status = os.fstat(current_fd)
            if not stat.S_ISDIR(status.st_mode):
                _close_fd(current_fd)
                return None
            result = (current_fd, status, "directory")
            current_fd = -1
            return result

        for index, segment in enumerate(segment_list):
            is_last = index == len(segment_list) - 1
            try:
                next_fd = os.open(segment, flags_base, dir_fd=current_fd)
            except OSError:
                _close_fd(current_fd)
                current_fd = -1
                return None

            try:
                status = os.fstat(next_fd)
            except OSError:
                _close_fd(next_fd)
                _close_fd(current_fd)
                current_fd = -1
                return None

            if not is_last:
                if not stat.S_ISDIR(status.st_mode):
                    _close_fd(next_fd)
                    _close_fd(current_fd)
                    current_fd = -1
                    return None
                _close_fd(current_fd)
                current_fd = next_fd
                continue

            if stat.S_ISREG(status.st_mode):
                kind = "file"
            elif stat.S_ISDIR(status.st_mode):
                kind = "directory"
            else:
                _close_fd(next_fd)
                _close_fd(current_fd)
                current_fd = -1
                return None

            _close_fd(current_fd)
            current_fd = -1
            return next_fd, status, kind

        _close_fd(current_fd)
        current_fd = -1
        return None
    except OSError:
        _close_fd(current_fd)
        return None


def human_size(size: Optional[int]) -> str:
    """Format a byte count for the listing UI."""

    if size is None:
        return "—"
    units = ("B", "KB", "MB", "GB", "TB")
    value = float(size)
    for unit in units:
        if value < 1024.0 or unit == units[-1]:
            if unit == "B":
                return "{} {}".format(int(value), unit)
            return "{:.1f} {}".format(value, unit)
        value /= 1024.0
    return "{} B".format(size)


def list_directory_entries(
    root: Path,
    segments: Tuple[str, ...] = (),
) -> Optional[list]:
    """Return visible direct children of ``root/segments``.

    Uses O_NOFOLLOW opens so concurrent symlink swaps cannot escape the share.
    Returns ``None`` when the target is missing, not a directory, or reserved.
    """

    opened = open_within_root(root, segments)
    if opened is None:
        return None
    directory_fd, _status, kind = opened
    if kind != "directory":
        _close_fd(directory_fd)
        return None

    entries = []
    try:
        with os.scandir(directory_fd) as directory_entries:
            for entry in directory_entries:
                try:
                    # Do not follow symlinks when classifying entries.
                    if entry.is_symlink():
                        continue
                    if entry.is_file(follow_symlinks=False):
                        kind = "file"
                        name = entry.name
                        try:
                            size = entry.stat(follow_symlinks=False).st_size
                        except OSError:
                            size = None
                    elif entry.is_dir(follow_symlinks=False):
                        kind = "directory"
                        name = entry.name
                        size = None
                    else:
                        continue
                except OSError:
                    continue
                try:
                    name.encode("utf-8")
                except UnicodeEncodeError:
                    continue
                if name in RESERVED_SHARE_NAMES or name.startswith("."):
                    continue
                item = {"name": name, "kind": kind}
                if kind == "file" and size is not None:
                    item["size"] = size
                entries.append(item)
    finally:
        _close_fd(directory_fd)
    return sorted(entries, key=lambda entry: entry["name"])


def list_directory_entries(
    root: Path,
    segments: Tuple[str, ...] = (),
) -> Optional[list]:
    """Return visible direct children of ``root/segments``.

    Uses O_NOFOLLOW opens so concurrent symlink swaps cannot escape the share.
    Returns ``None`` when the target is missing, not a directory, or reserved.
    """

    opened = open_within_root(root, segments)
    if opened is None:
        return None
    directory_fd, _status, kind = opened
    if kind != "directory":
        _close_fd(directory_fd)
        return None

    entries = []
    try:
        with os.scandir(directory_fd) as directory_entries:
            for entry in directory_entries:
                try:
                    # Do not follow symlinks when classifying entries.
                    if entry.is_symlink():
                        continue
                    if entry.is_file(follow_symlinks=False):
                        kind = "file"
                        name = entry.name
                        try:
                            size = entry.stat(follow_symlinks=False).st_size
                        except OSError:
                            size = None
                    elif entry.is_dir(follow_symlinks=False):
                        kind = "directory"
                        name = entry.name
                        size = None
                    else:
                        continue
                except OSError:
                    continue
                try:
                    name.encode("utf-8")
                except UnicodeEncodeError:
                    continue
                if name in RESERVED_SHARE_NAMES or name.startswith("."):
                    continue
                item = {"name": name, "kind": kind}
                if kind == "file" and size is not None:
                    item["size"] = size
                entries.append(item)
    finally:
        _close_fd(directory_fd)
    return sorted(entries, key=lambda entry: entry["name"])


def top_level_entries(root: Path) -> list:
    """Return visible direct children of the share root (compat wrapper)."""

    entries = list_directory_entries(root, ())
    return entries if entries is not None else []


def _encoded_path_prefix(segments: Tuple[str, ...]) -> str:
    """Build a URL path prefix for the given segments (no trailing slash)."""

    if not segments:
        return ""
    return "/" + "/".join(quote(segment, safe="") for segment in segments)


def render_listing_html(
    entries: list,
    base_url: str,
    path_segments: Tuple[str, ...] = (),
) -> str:
    """Render a browser-friendly listing with browse + download actions."""

    base = base_url.rstrip("/")
    path_prefix = _encoded_path_prefix(path_segments)
    current_path_label = "/".join(path_segments) if path_segments else "/"
    json_href = (path_prefix + "/" if path_segments else "/") + "?format=json"

    # Breadcrumb: 根目录 / nested / deeper
    breadcrumb_parts = [
        '<a class="crumb" href="{}/">根目录</a>'.format(html.escape(base, quote=True))
        if path_segments
        else '<span class="crumb current">根目录</span>'
    ]
    accumulated: List[str] = []
    for index, segment in enumerate(path_segments):
        accumulated.append(segment)
        crumb_path = _encoded_path_prefix(tuple(accumulated)) + "/"
        if index == len(path_segments) - 1:
            breadcrumb_parts.append(
                '<span class="crumb current">{}</span>'.format(html.escape(segment))
            )
        else:
            breadcrumb_parts.append(
                '<a class="crumb" href="{}{}">{}</a>'.format(
                    html.escape(base, quote=True),
                    html.escape(crumb_path, quote=True),
                    html.escape(segment),
                )
            )
    breadcrumb_html = ' <span class="sep">/</span> '.join(breadcrumb_parts)

    rows = []
    if path_segments:
        parent_segments = path_segments[:-1]
        parent_href = (
            base + _encoded_path_prefix(parent_segments) + "/"
            if parent_segments
            else base + "/"
        )
        rows.append(
            """
            <tr class="parent-row">
              <td class="icon">⬆</td>
              <td class="name">
                <a class="file-link" href="{href}" title="返回上级">..</a>
                <div class="meta">上级目录</div>
              </td>
              <td class="url-cell"><code class="url muted-url">—</code></td>
              <td class="actions"></td>
            </tr>
            """.format(href=html.escape(parent_href, quote=True))
        )

    for entry in entries:
        name = entry["name"]
        kind = entry["kind"]
        entry_prefix = path_prefix + "/" + quote(name, safe="")
        if kind == "directory":
            # Click name to browse; ZIP stays a separate download URL.
            browse_href = base + entry_prefix + "/"
            zip_href = base + entry_prefix + ".zip"
            href = browse_href
            wget_cmd = "wget '{}'".format(zip_href)
            kind_label = "目录 · 点击进入"
            size_label = "可浏览 / ZIP 打包"
            icon = "📁"
            link_title = "点击进入文件夹"
            secondary_btn = (
                '<a class="btn secondary link-btn" href="{zip_attr}">下载 ZIP</a>'
                '<button type="button" class="btn secondary" data-copy="{zip_attr}">复制 ZIP</button>'
                '<button type="button" class="btn secondary" data-copy="{wget_attr}">复制 wget</button>'
            ).format(
                zip_attr=html.escape(zip_href, quote=True),
                wget_attr=html.escape(wget_cmd, quote=True),
            )
            url_display = zip_href
            url_copy = zip_href
        else:
            file_href = base + entry_prefix
            href = file_href
            wget_cmd = "wget '{}'".format(file_href)
            kind_label = "文件"
            size_label = human_size(entry.get("size"))
            icon = "📄"
            link_title = "点击下载"
            secondary_btn = (
                '<button type="button" class="btn" data-copy="{href_attr}">复制链接</button>'
                '<button type="button" class="btn secondary" data-copy="{wget_attr}">复制 wget</button>'
            ).format(
                href_attr=html.escape(file_href, quote=True),
                wget_attr=html.escape(wget_cmd, quote=True),
            )
            url_display = file_href
            url_copy = file_href

        rows.append(
            """
            <tr>
              <td class="icon">{icon}</td>
              <td class="name">
                <a class="file-link" href="{href}" title="{link_title}">{name_html}</a>
                <div class="meta">{kind_label} · {size_label}</div>
              </td>
              <td class="url-cell">
                <code class="url" data-copy="{url_attr}">{url_html}</code>
              </td>
              <td class="actions">
                {secondary_btn}
              </td>
            </tr>
            """.format(
                icon=icon,
                href=html.escape(href, quote=True),
                link_title=html.escape(link_title, quote=True),
                name_html=html.escape(name),
                kind_label=html.escape(kind_label),
                size_label=html.escape(size_label),
                url_attr=html.escape(url_copy, quote=True),
                url_html=html.escape(url_display),
                secondary_btn=secondary_btn,
            )
        )

    if not entries and not path_segments:
        body_rows = (
            '<tr><td colspan="4" class="empty">目录为空，把文件放进共享目录后刷新。</td></tr>'
        )
    elif not entries:
        body_rows = "\n".join(rows) + (
            '<tr><td colspan="4" class="empty">此文件夹为空。</td></tr>'
        )
    else:
        body_rows = "\n".join(rows)

    title = (
        "局域网文件共享 · {}".format(current_path_label)
        if path_segments
        else "局域网文件共享"
    )

    return """<!DOCTYPE html>
<html lang="zh-CN">
<head>
  <meta charset="utf-8">
  <meta name="viewport" content="width=device-width, initial-scale=1">
  <title>{title}</title>
  <style>
    :root {{
      --bg: #0f172a;
      --panel: #111827;
      --card: #1f2937;
      --text: #e5e7eb;
      --muted: #94a3b8;
      --accent: #38bdf8;
      --accent-2: #22c55e;
      --border: #334155;
      --btn: #0ea5e9;
      --btn-hover: #0284c7;
    }}
    * {{ box-sizing: border-box; }}
    body {{
      margin: 0;
      font-family: "Segoe UI", "PingFang SC", "Microsoft YaHei", sans-serif;
      background:
        radial-gradient(circle at top right, rgba(56,189,248,.18), transparent 28%),
        radial-gradient(circle at left, rgba(34,197,94,.12), transparent 24%),
        var(--bg);
      color: var(--text);
      min-height: 100vh;
    }}
    .wrap {{
      max-width: 1100px;
      margin: 0 auto;
      padding: 28px 18px 48px;
    }}
    .hero {{
      background: linear-gradient(135deg, rgba(30,41,59,.95), rgba(15,23,42,.92));
      border: 1px solid var(--border);
      border-radius: 18px;
      padding: 22px 24px;
      box-shadow: 0 16px 40px rgba(0,0,0,.28);
      margin-bottom: 18px;
    }}
    h1 {{
      margin: 0 0 8px;
      font-size: 1.55rem;
      letter-spacing: .02em;
    }}
    .subtitle {{
      margin: 0;
      color: var(--muted);
      line-height: 1.5;
    }}
    .breadcrumb {{
      margin-top: 14px;
      padding: 10px 14px;
      border-radius: 12px;
      background: rgba(15,23,42,.7);
      border: 1px solid var(--border);
      font-size: .95rem;
      line-height: 1.6;
      word-break: break-all;
    }}
    .crumb {{
      color: var(--accent);
      text-decoration: none;
      font-weight: 600;
    }}
    .crumb:hover {{ text-decoration: underline; }}
    .crumb.current {{ color: #f8fafc; }}
    .sep {{ color: var(--muted); margin: 0 2px; }}
    .toolbar {{
      display: flex;
      flex-wrap: wrap;
      gap: 10px;
      margin-top: 16px;
      align-items: center;
    }}
    .chip {{
      display: inline-flex;
      align-items: center;
      gap: 8px;
      background: rgba(15,23,42,.7);
      border: 1px solid var(--border);
      border-radius: 999px;
      padding: 8px 12px;
      color: var(--muted);
      font-size: .92rem;
    }}
    .chip code {{
      color: var(--accent);
      background: transparent;
      font-size: .9rem;
    }}
    .chip a {{ color: var(--accent); text-decoration: none; }}
    .panel {{
      background: rgba(17,24,39,.9);
      border: 1px solid var(--border);
      border-radius: 18px;
      overflow: hidden;
      box-shadow: 0 12px 30px rgba(0,0,0,.22);
    }}
    table {{
      width: 100%;
      border-collapse: collapse;
    }}
    th, td {{
      padding: 14px 16px;
      border-bottom: 1px solid rgba(51,65,85,.8);
      vertical-align: middle;
      text-align: left;
    }}
    th {{
      font-size: .78rem;
      text-transform: uppercase;
      letter-spacing: .08em;
      color: var(--muted);
      background: rgba(15,23,42,.85);
      position: sticky;
      top: 0;
    }}
    tr:last-child td {{ border-bottom: none; }}
    tr:hover td {{ background: rgba(30,41,59,.55); }}
    tr.parent-row td {{ background: rgba(15,23,42,.35); }}
    .icon {{ width: 42px; font-size: 1.2rem; }}
    .name {{ min-width: 180px; }}
    .file-link {{
      color: #f8fafc;
      text-decoration: none;
      font-weight: 600;
      word-break: break-all;
    }}
    .file-link:hover {{ color: var(--accent); text-decoration: underline; }}
    .meta {{
      margin-top: 4px;
      color: var(--muted);
      font-size: .85rem;
    }}
    .url-cell {{ width: 38%; }}
    .url {{
      display: block;
      padding: 10px 12px;
      border-radius: 10px;
      background: #0b1220;
      border: 1px solid var(--border);
      color: #7dd3fc;
      font-size: .86rem;
      word-break: break-all;
      user-select: all;
      cursor: text;
    }}
    .muted-url {{ color: var(--muted); cursor: default; }}
    .actions {{
      white-space: nowrap;
      width: 1%;
    }}
    .btn, a.link-btn {{
      appearance: none;
      border: none;
      border-radius: 10px;
      background: var(--btn);
      color: white;
      font-weight: 600;
      padding: 8px 12px;
      margin: 0 6px 6px 0;
      cursor: pointer;
      font-size: .86rem;
      text-decoration: none;
      display: inline-block;
    }}
    .btn:hover, a.link-btn:hover {{ background: var(--btn-hover); }}
    .btn.secondary, a.link-btn.secondary {{
      background: #334155;
    }}
    .btn.secondary:hover, a.link-btn.secondary:hover {{ background: #475569; }}
    .btn.copied {{ background: var(--accent-2); }}
    .empty {{
      text-align: center;
      color: var(--muted);
      padding: 36px 16px !important;
    }}
    .footer {{
      margin-top: 14px;
      color: var(--muted);
      font-size: .88rem;
    }}
    .toast {{
      position: fixed;
      right: 18px;
      bottom: 18px;
      background: #14532d;
      color: #dcfce7;
      border: 1px solid #22c55e;
      padding: 10px 14px;
      border-radius: 12px;
      opacity: 0;
      transform: translateY(8px);
      transition: .18s ease;
      pointer-events: none;
      z-index: 20;
    }}
    .toast.show {{
      opacity: 1;
      transform: translateY(0);
    }}
    @media (max-width: 820px) {{
      th:nth-child(3), td.url-cell {{ display: none; }}
      .actions .btn, .actions a.link-btn {{ display: block; width: 100%; margin-right: 0; }}
    }}
  </style>
</head>
<body>
  <div class="wrap">
    <section class="hero">
      <h1>局域网文件共享</h1>
      <p class="subtitle">点击文件夹进入浏览；点击文件名直接下载。目录也可单独打包 ZIP。</p>
      <div class="breadcrumb">{breadcrumb}</div>
      <div class="toolbar">
        <div class="chip">路径 <code>{path_html}</code></div>
        <div class="chip">条目 <strong style="color:#e2e8f0">{count}</strong></div>
        <div class="chip">基础地址 <code>{base_html}</code></div>
        <div class="chip">JSON <a href="{json_href_attr}">{json_href_html}</a></div>
      </div>
    </section>
    <section class="panel">
      <table>
        <thead>
          <tr>
            <th></th>
            <th>名称</th>
            <th>下载地址</th>
            <th>操作</th>
          </tr>
        </thead>
        <tbody>
          {rows}
        </tbody>
      </table>
    </section>
    <p class="footer">提示：目录名点进去浏览内部文件；「下载 ZIP / 复制 wget」用于整夹打包。手机上可先复制链接再粘贴到下载器。</p>
  </div>
  <div class="toast" id="toast">已复制</div>
  <script>
    (function () {{
      var toast = document.getElementById('toast');
      var timer = null;
      function showToast(text) {{
        toast.textContent = text || '已复制';
        toast.classList.add('show');
        if (timer) clearTimeout(timer);
        timer = setTimeout(function () {{ toast.classList.remove('show'); }}, 1400);
      }}
      function copyText(text, button) {{
        function done() {{
          if (button) {{
            button.classList.add('copied');
            var old = button.textContent;
            button.textContent = '已复制';
            setTimeout(function () {{
              button.classList.remove('copied');
              button.textContent = old;
            }}, 1200);
          }}
          showToast('已复制到剪贴板');
        }}
        if (navigator.clipboard && navigator.clipboard.writeText) {{
          navigator.clipboard.writeText(text).then(done).catch(function () {{
            window.prompt('复制以下内容', text);
          }});
        }} else {{
          var area = document.createElement('textarea');
          area.value = text;
          document.body.appendChild(area);
          area.select();
          try {{ document.execCommand('copy'); done(); }}
          catch (e) {{ window.prompt('复制以下内容', text); }}
          document.body.removeChild(area);
        }}
      }}
      document.addEventListener('click', function (event) {{
        var target = event.target;
        if (!(target instanceof Element)) return;
        var button = target.closest('[data-copy]');
        if (!button) return;
        var value = button.getAttribute('data-copy');
        if (!value) return;
        copyText(value, button.tagName === 'BUTTON' ? button : null);
      }});
    }})();
  </script>
</body>
</html>
""".format(
        title=html.escape(title),
        breadcrumb=breadcrumb_html,
        path_html=html.escape(current_path_label),
        count=len(entries),
        base_html=html.escape(base),
        json_href_attr=html.escape(json_href, quote=True),
        json_href_html=html.escape(json_href),
        rows=body_rows,
    )


def safe_zip_arcname(relative_path: str) -> Optional[str]:
    """Reject archive member names that are unsafe for common extractors."""

    normalized = relative_path.replace("\\", "/")
    if normalized.startswith("/") or normalized.startswith("../") or "/../" in "/{}/".format(normalized):
        return None
    if ".." in normalized.split("/"):
        return None
    return normalized


def write_directory_archive(directory_fd: int, archive_label: str) -> str:
    """Create a temporary ZIP from an already-opened directory descriptor.

    Entries are opened with O_NOFOLLOW relative to their parent directory fd so
    symlink swaps cannot pull outside content into the archive.
    """

    del archive_label  # reserved for future naming; path is temp-based
    descriptor, archive_name = tempfile.mkstemp(suffix=".zip", prefix="lan-file-share-")
    os.close(descriptor)
    archive_path = Path(archive_name)
    file_count = 0
    uncompressed_bytes = 0

    try:
        with zipfile.ZipFile(
            str(archive_path),
            mode="w",
            compression=zipfile.ZIP_STORED,
            allowZip64=True,
        ) as archive:
            stack: List[Tuple[int, str, int]] = [(directory_fd, "", 0)]
            owned_fds: List[int] = []
            try:
                while stack:
                    current_fd, prefix, depth = stack.pop()
                    if depth > MAX_ARCHIVE_DEPTH:
                        raise OSError("Archive depth limit exceeded")
                    with os.scandir(current_fd) as directory_entries:
                        for entry in directory_entries:
                            try:
                                if entry.is_symlink():
                                    continue
                            except OSError:
                                continue

                            child_name = entry.name
                            relative = (
                                "{}/{}".format(prefix, child_name) if prefix else child_name
                            )
                            arcname = safe_zip_arcname(relative)
                            if arcname is None:
                                continue

                            try:
                                if entry.is_dir(follow_symlinks=False):
                                    child_fd = os.open(
                                        child_name,
                                        os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW | os.O_CLOEXEC,
                                        dir_fd=current_fd,
                                    )
                                    owned_fds.append(child_fd)
                                    archive.writestr(arcname.rstrip("/") + "/", b"")
                                    stack.append((child_fd, arcname, depth + 1))
                                    continue
                                if not entry.is_file(follow_symlinks=False):
                                    continue
                            except OSError:
                                continue

                            try:
                                child_fd = os.open(
                                    child_name,
                                    os.O_RDONLY | os.O_NOFOLLOW | os.O_CLOEXEC,
                                    dir_fd=current_fd,
                                )
                            except OSError:
                                continue
                            try:
                                status = os.fstat(child_fd)
                                if not stat.S_ISREG(status.st_mode):
                                    continue
                                file_count += 1
                                uncompressed_bytes += status.st_size
                                if file_count > MAX_ARCHIVE_FILE_COUNT:
                                    raise OSError("Archive file count limit exceeded")
                                if uncompressed_bytes > MAX_ARCHIVE_UNCOMPRESSED_BYTES:
                                    raise OSError("Archive size limit exceeded")
                                with os.fdopen(child_fd, "rb") as source:
                                    child_fd = -1  # ownership moved
                                    with archive.open(arcname, mode="w") as target:
                                        shutil.copyfileobj(
                                            source,
                                            target,
                                            length=COPY_CHUNK_SIZE,
                                        )
                            finally:
                                _close_fd(child_fd)
            finally:
                for owned in owned_fds:
                    _close_fd(owned)
    except BaseException:
        archive_path.unlink(missing_ok=True)
        raise

    return str(archive_path)


class FileShareRequestHandler(BaseHTTPRequestHandler):
    """Serve a single configured share root."""

    server_version = "LanFileShare/1.0"
    protocol_version = "HTTP/1.1"
    timeout = REQUEST_TIMEOUT_SECONDS

    @property
    def share_root(self) -> Path:
        return self.server.share_root  # type: ignore[attr-defined]

    def send_json_error(self, status: HTTPStatus, message: str) -> None:
        """Return a small JSON error without leaking local filesystem details."""

        body = json.dumps({"error": message}).encode("utf-8")
        self.send_response(status)
        self.send_header("Content-Type", "application/json; charset=utf-8")
        self.send_header("Content-Length", str(len(body)))
        self.send_header("Connection", "close")
        self.end_headers()
        self.wfile.write(body)
        self.close_connection = True

    def send_download_headers(
        self,
        *,
        content_type: str,
        content_length: int,
        filename: str,
    ) -> None:
        """Send common response headers for an attachment download."""

        self.send_response(HTTPStatus.OK)
        self.send_header("Content-Type", content_type)
        self.send_header("Content-Length", str(content_length))
        self.send_header("Content-Disposition", content_disposition(filename))
        self.send_header("X-Content-Type-Options", "nosniff")
        self.send_header("Cache-Control", "no-store")
        self.end_headers()

    def do_GET(self) -> None:  # noqa: N802 - required by BaseHTTPRequestHandler
        try:
            parsed = urlsplit(self.path)
        except ValueError:
            self.send_json_error(HTTPStatus.NOT_FOUND, "Resource not found.")
            return

        if parsed.fragment or parsed.netloc:
            self.send_json_error(HTTPStatus.NOT_FOUND, "Resource not found.")
            return

        # Listing lives at the site root. /list remains as a compatibility alias.
        if parsed.path in ("/", "/list"):
            self.send_listing(parsed.query, path_segments=())
            return

        segments = decoded_request_path(parsed.path)
        if segments is None:
            self.send_json_error(HTTPStatus.NOT_FOUND, "Resource not found.")
            return

        # Root ZIP download is intentionally disabled; only explicit children are served.
        if not segments:
            self.send_json_error(HTTPStatus.NOT_FOUND, "Resource not found.")
            return

        # Trailing slash means "browse this directory" (HTML/JSON listing).
        # Without the slash, directories still download as ZIP for wget compatibility.
        wants_listing = parsed.path.endswith("/")
        if wants_listing:
            if parsed.query and not self._wants_json_listing(parsed.query):
                # Unknown query keys on directory listings are ignored only when
                # format is absent/json; other formats already handled above.
                pass
            self.send_listing(parsed.query, path_segments=segments)
            return

        if parsed.query:
            self.send_json_error(HTTPStatus.NOT_FOUND, "Resource not found.")
            return

        # Prefer real files named *.zip; otherwise treat trailing .zip as a
        # directory-archive alias so clients save with a .zip extension.
        opened = open_within_root(self.share_root, segments)
        archive_segments = None
        if opened is None and segments[-1].lower().endswith(".zip") and len(segments[-1]) > 4:
            directory_segments = segments[:-1] + (segments[-1][:-4],)
            directory_opened = open_within_root(self.share_root, directory_segments)
            if directory_opened is not None and directory_opened[2] == "directory":
                opened = directory_opened
                archive_segments = directory_segments

        if opened is None:
            self.send_json_error(HTTPStatus.NOT_FOUND, "Resource not found.")
            return

        file_descriptor, status, kind = opened
        try:
            if kind == "file":
                self.send_file(file_descriptor, status, segments)
                file_descriptor = -1
                return
            if kind == "directory":
                # Always attach a .zip filename for directory archives.
                self.send_directory_archive(
                    file_descriptor,
                    archive_segments or segments,
                )
                file_descriptor = -1
                return
            self.send_json_error(HTTPStatus.NOT_FOUND, "Resource not found.")
        finally:
            _close_fd(file_descriptor)

    def do_HEAD(self) -> None:  # noqa: N802 - required by BaseHTTPRequestHandler
        self.send_method_not_allowed()

    def do_POST(self) -> None:  # noqa: N802 - required by BaseHTTPRequestHandler
        self.send_method_not_allowed()

    def do_PUT(self) -> None:  # noqa: N802 - required by BaseHTTPRequestHandler
        self.send_method_not_allowed()

    def do_DELETE(self) -> None:  # noqa: N802 - required by BaseHTTPRequestHandler
        self.send_method_not_allowed()

    def send_method_not_allowed(self) -> None:
        """Reject unsupported request methods consistently."""

        body = json.dumps({"error": "Only GET is supported."}).encode("utf-8")
        self.send_response(HTTPStatus.METHOD_NOT_ALLOWED)
        self.send_header("Allow", "GET")
        self.send_header("Content-Type", "application/json; charset=utf-8")
        self.send_header("Content-Length", str(len(body)))
        self.send_header("Connection", "close")
        self.end_headers()
        self.wfile.write(body)
        self.close_connection = True

    def _wants_json_listing(self, query: str) -> bool:
        """Prefer JSON only when explicitly requested."""

        if query:
            params = parse_qs(query, keep_blank_values=False)
            formats = [value.lower() for value in params.get("format", [])]
            if "json" in formats:
                return True
            if formats:
                return False
        accept = self.headers.get("Accept", "")
        if "application/json" in accept and "text/html" not in accept:
            return True
        return False

    def _public_base_url(self) -> str:
        """Build absolute http://host:port base from the request Host header."""

        host = self.headers.get("Host", "").strip()
        if not host or any(ch in host for ch in ("/", "\\", " ", "\r", "\n", "\t")):
            host = "127.0.0.1:{}".format(self.server.server_address[1])
        return "http://{}".format(host)

    def send_listing(
        self,
        query: str = "",
        path_segments: Tuple[str, ...] = (),
    ) -> None:
        """Send a browser-friendly HTML listing, or JSON when requested."""

        try:
            entries = list_directory_entries(self.share_root, path_segments)
            if entries is None:
                self.send_json_error(HTTPStatus.NOT_FOUND, "Resource not found.")
                return
            relative_path = "/".join(path_segments)
            if self._wants_json_listing(query):
                # Keep API payload compact for scripts while still including size.
                payload = {
                    "path": relative_path,
                    "entries": [
                        {
                            "name": entry["name"],
                            "kind": entry["kind"],
                            **(
                                {"size": entry["size"]}
                                if "size" in entry
                                else {}
                            ),
                        }
                        for entry in entries
                    ]
                }
                body = json.dumps(payload, ensure_ascii=False, indent=2).encode("utf-8")
                content_type = "application/json; charset=utf-8"
            else:
                body = render_listing_html(
                    entries,
                    self._public_base_url(),
                    path_segments=path_segments,
                ).encode("utf-8")
                content_type = "text/html; charset=utf-8"
        except (OSError, UnicodeEncodeError, TypeError, ValueError):
            logging.exception("Unable to list directory")
            self.send_json_error(HTTPStatus.INTERNAL_SERVER_ERROR, "Unable to list files.")
            return

        self.send_response(HTTPStatus.OK)
        self.send_header("Content-Type", content_type)
        self.send_header("Content-Length", str(len(body)))
        self.send_header("Cache-Control", "no-store")
        self.end_headers()
        self.wfile.write(body)

    def send_file(
        self,
        file_descriptor: int,
        status: os.stat_result,
        segments: Tuple[str, ...],
    ) -> None:
        """Stream a regular file opened with O_NOFOLLOW."""

        filename = segments[-1] if segments else "download"
        try:
            content_type = mimetypes.guess_type(filename)[0]
            self.send_download_headers(
                content_type=content_type or "application/octet-stream",
                content_length=status.st_size,
                filename=filename,
            )
            with os.fdopen(file_descriptor, "rb") as source:
                file_descriptor = -1
                shutil.copyfileobj(source, self.wfile, length=COPY_CHUNK_SIZE)
        except (OSError, BrokenPipeError, ConnectionResetError):
            logging.exception("Unable to send requested file")
            self.close_connection = True
        finally:
            _close_fd(file_descriptor)

    def send_directory_archive(
        self,
        directory_fd: int,
        segments: Tuple[str, ...],
    ) -> None:
        """Build, stream, and remove a ZIP archive for a requested directory."""

        archive_name = "-".join(segments) if segments else self.share_root.name
        if archive_name.lower().endswith(".zip"):
            archive_name = archive_name[:-4]
        archive_path = None
        acquired = _ARCHIVE_SEMAPHORE.acquire(blocking=False)
        if not acquired:
            _close_fd(directory_fd)
            self.send_json_error(
                HTTPStatus.SERVICE_UNAVAILABLE,
                "Too many concurrent archive downloads.",
            )
            return

        try:
            try:
                archive_path = Path(
                    write_directory_archive(directory_fd, archive_name)
                )
            except OSError as error:
                message = str(error)
                if "limit exceeded" in message:
                    self.send_json_error(
                        HTTPStatus.REQUEST_ENTITY_TOO_LARGE,
                        "Archive exceeds configured size or file limits.",
                    )
                else:
                    logging.exception("Unable to create directory archive")
                    self.send_json_error(
                        HTTPStatus.INTERNAL_SERVER_ERROR,
                        "Unable to create archive.",
                    )
                return
            finally:
                _close_fd(directory_fd)
                directory_fd = -1

            try:
                size = archive_path.stat().st_size
                self.send_download_headers(
                    content_type="application/zip",
                    content_length=size,
                    filename=archive_name + ".zip",
                )
                with archive_path.open("rb") as source:
                    shutil.copyfileobj(source, self.wfile, length=COPY_CHUNK_SIZE)
            except (OSError, BrokenPipeError, ConnectionResetError):
                logging.exception("Unable to send directory archive")
                self.close_connection = True
        finally:
            if archive_path is not None:
                archive_path.unlink(missing_ok=True)
            _ARCHIVE_SEMAPHORE.release()
            _close_fd(directory_fd)

    def log_message(self, format_string: str, *arguments: object) -> None:
        """Route request logs through the application logger."""

        safe_arguments = tuple(sanitized_log_text(argument) for argument in arguments)
        logging.info(
            "%s - %s",
            sanitized_log_text(self.client_address[0]),
            sanitized_log_text(format_string % safe_arguments),
        )


def create_server(
    root: Path,
    host: str = DEFAULT_HOST,
    port: int = DEFAULT_PORT,
) -> ShareHTTPServer:
    """Create a bound HTTP server restricted to an existing directory root."""

    resolved_root = Path(root).resolve(strict=True)
    if not resolved_root.is_dir():
        raise NotADirectoryError("Share root must be a directory.")

    handler_factory: Callable[..., FileShareRequestHandler] = partial(
        FileShareRequestHandler,
    )
    server = ShareHTTPServer((host, port), handler_factory)
    server.share_root = resolved_root  # type: ignore[attr-defined]
    return server


def configure_logging() -> None:
    """Configure concise logs suitable for terminal or systemd journal output."""

    logging.basicConfig(
        level=logging.INFO,
        format="%(asctime)s %(levelname)s %(message)s",
    )


def main() -> int:
    """Run the production LAN file server until interrupted."""

    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--root", type=Path, default=Path(__file__).parent)
    parser.add_argument("--host", default=DEFAULT_HOST)
    parser.add_argument("--port", type=int, default=DEFAULT_PORT)
    arguments = parser.parse_args()

    configure_logging()
    server = create_server(arguments.root, arguments.host, arguments.port)

    def stop_server(signum: int, _frame: object) -> None:
        logging.info("Received signal %s; stopping LAN file share", signum)
        # shutdown() waits for serve_forever() and must run outside its thread.
        threading.Thread(target=server.shutdown, daemon=True).start()

    signal.signal(signal.SIGINT, stop_server)
    signal.signal(signal.SIGTERM, stop_server)
    logging.info(
        "Serving %s on http://%s:%s",
        server.share_root,
        arguments.host,
        arguments.port,
    )
    try:
        server.serve_forever()
    finally:
        server.server_close()
    return 0


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