feat: add previewer static serving

2026-10-11 18:19:46 +08:00
parent 0ec8ea81d8
commit 5ba1d8e3d6
9 changed files with 325 additions and 0 deletions
@@ -0,0 +1,177 @@
+#!/usr/bin/env python3
+"""Expose a static file's parent directory through a Tailscale Serve HTTP route."""
+
+from __future__ import annotations
+
+import argparse
+import ipaddress
+import json
+import re
+import shutil
+import subprocess
+import sys
+from pathlib import Path, PurePosixPath
+from urllib.parse import quote
+
+
+class PreviewError(RuntimeError):
+    pass
+
+
+def run_tailscale(*args: str) -> subprocess.CompletedProcess[str]:
+    try:
+        return subprocess.run(
+            ["tailscale", *args],
+            check=False,
+            capture_output=True,
+            text=True,
+        )
+    except OSError as exc:
+        raise PreviewError(f"无法运行 tailscale：{exc}") from exc
+
+
+def require_tailscale() -> None:
+    if shutil.which("tailscale") is None:
+        raise PreviewError("未找到 tailscale，请先安装并登录 Tailscale")
+
+
+def tailscale_ipv4() -> str:
+    result = run_tailscale("ip", "-4")
+    if result.returncode != 0:
+        detail = (result.stderr or result.stdout).strip()
+        raise PreviewError(f"Tailscale 节点不可用：{detail or 'tailscale ip -4 失败'}")
+    for line in result.stdout.splitlines():
+        candidate = line.strip()
+        try:
+            address = ipaddress.ip_address(candidate)
+        except ValueError:
+            continue
+        if address.version == 4:
+            return candidate
+    raise PreviewError("tailscale 没有返回 IPv4 地址；请确认节点已连接 tailnet")
+
+
+def serve_status() -> tuple[dict, str]:
+    result = run_tailscale("serve", "status", "--json")
+    if result.returncode != 0:
+        detail = (result.stderr or result.stdout).strip()
+        raise PreviewError(f"无法读取 Tailscale Serve 状态：{detail or 'status 失败'}")
+    try:
+        data = json.loads(result.stdout or "{}")
+    except json.JSONDecodeError as exc:
+        raise PreviewError("Tailscale Serve 返回了无法解析的 JSON 状态") from exc
+    if not isinstance(data, dict):
+        raise PreviewError("Tailscale Serve 状态不是 JSON 对象")
+
+    plain = run_tailscale("serve", "status")
+    if plain.returncode != 0:
+        detail = (plain.stderr or plain.stdout).strip()
+        raise PreviewError(f"无法读取 Tailscale Serve 文本状态：{detail or 'status 失败'}")
+    return data, plain.stdout
+
+
+def status_blob(data: dict, plain: str) -> str:
+    return json.dumps(data, ensure_ascii=False, sort_keys=True) + "\n" + plain
+
+
+def normalize_mount_path(value: str) -> str:
+    path = value.strip()
+    if not path.startswith("/") or path == "/":
+        raise PreviewError("mount path 必须以 / 开头且不能是根路径")
+    if "?" in path or "#" in path or "//" in path:
+        raise PreviewError("mount path 不能包含 query、fragment 或空路径段")
+    parts = PurePosixPath(path).parts
+    if any(part in {".", ".."} for part in parts):
+        raise PreviewError("mount path 不能包含 . 或 ..")
+    return "/" + "/".join(parts[1:])
+
+
+def default_mount_path(directory: Path) -> str:
+    slug = re.sub(r"[^a-zA-Z0-9._-]+", "-", directory.name).strip("-")
+    if not slug:
+        raise PreviewError("无法从文件目录生成 mount path，请显式提供 --mount-path")
+    return f"/preview/{slug}"
+
+
+def route_matches(blob: str, mount_path: str, directory: Path) -> bool:
+    return mount_path in blob and str(directory) in blob
+
+
+def route_conflicts(blob: str, mount_path: str, directory: Path) -> bool:
+    return mount_path in blob and str(directory) not in blob
+
+
+def configure_route(directory: Path, mount_path: str, port: int, data: dict, plain: str) -> None:
+    blob = status_blob(data, plain)
+    if route_conflicts(blob, mount_path, directory):
+        raise PreviewError(
+            f"mount path {mount_path} 已被其它目录占用；请改用唯一的 --mount-path，不会覆盖已有路由"
+        )
+
+    if route_matches(blob, mount_path, directory):
+        return
+
+    result = run_tailscale(
+        "serve",
+        "--bg",
+        f"--http={port}",
+        f"--set-path={mount_path}",
+        "--yes",
+        str(directory),
+    )
+    if result.returncode != 0:
+        detail = (result.stderr or result.stdout).strip()
+        raise PreviewError(f"Tailscale Serve 配置失败：{detail or 'serve 命令失败'}")
+
+    data, plain = serve_status()
+    blob = status_blob(data, plain)
+    if route_conflicts(blob, mount_path, directory) or not route_matches(blob, mount_path, directory):
+        raise PreviewError("Serve 命令返回成功，但状态中没有确认目标路由；停止输出 URL")
+
+
+def build_result(file_path: Path, directory: Path, mount_path: str, ip: str, port: int) -> dict[str, object]:
+    relative = file_path.relative_to(directory).as_posix()
+    suffix = f":{port}" if port != 80 else ""
+    url = f"http://{ip}{suffix}{mount_path}/{quote(relative, safe='/')}"
+    return {
+        "file": str(file_path),
+        "directory": str(directory),
+        "mountPath": mount_path,
+        "ip": ip,
+        "port": port,
+        "url": url,
+        "stopCommand": f"tailscale serve --http={port} --set-path={mount_path} off",
+    }
+
+
+def parse_args(argv: list[str]) -> argparse.Namespace:
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("file", type=Path, help="要展示的静态文件")
+    parser.add_argument("--mount-path", help="Tailscale Serve 的 URL 路径")
+    parser.add_argument("--port", type=int, default=80, help="HTTP 端口，默认 80")
+    return parser.parse_args(argv)
+
+
+def main(argv: list[str] | None = None) -> int:
+    args = parse_args(argv or sys.argv[1:])
+    try:
+        if not 1 <= args.port <= 65535:
+            raise PreviewError("port 必须在 1 到 65535 之间")
+        require_tailscale()
+        file_path = args.file.expanduser().resolve()
+        if not file_path.is_file():
+            raise PreviewError(f"输入不是普通文件或不存在：{file_path}")
+        directory = file_path.parent
+        mount_path = normalize_mount_path(args.mount_path) if args.mount_path else default_mount_path(directory)
+        ip = tailscale_ipv4()
+        data, plain = serve_status()
+        configure_route(directory, mount_path, args.port, data, plain)
+        print(json.dumps(build_result(file_path, directory, mount_path, ip, args.port), ensure_ascii=False))
+        return 0
+    except PreviewError as exc:
+        print(f"previewer: {exc}", file=sys.stderr)
+        return 2
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())