Tasks/serve.py

"""Tasks 项目的 Windows 本地开发服务器控制脚本。"""

from __future__ import annotations

import argparse
import subprocess
import sys
import time
import webbrowser
from pathlib import Path

ROOT = Path(__file__).resolve().parent
SOURCE_DIR = ROOT / "src"
DEFAULT_PORT = 8000
HOST = "127.0.0.1"
REQUIRED_PATHS = (
    SOURCE_DIR / "index.html",
    SOURCE_DIR / "css",
    SOURCE_DIR / "js",
)


def listener_pid(port: int) -> int | None:
    command = (
        f"Get-NetTCPConnection -LocalPort {port} -State Listen "
        "-ErrorAction SilentlyContinue | "
        "Select-Object -First 1 -ExpandProperty OwningProcess"
    )
    result = subprocess.run(
        ["powershell", "-NoProfile", "-Command", command],
        capture_output=True,
        text=True,
        check=False,
    )
    try:
        return int(result.stdout.strip())
    except ValueError:
        return None


def command_line(pid: int) -> str:
    command = f"(Get-CimInstance Win32_Process -Filter 'ProcessId = {pid}').CommandLine"
    result = subprocess.run(
        ["powershell", "-NoProfile", "-Command", command],
        capture_output=True,
        text=True,
        check=False,
    )
    return result.stdout.strip()


def is_tasks_server(pid: int, port: int) -> bool:
    command = command_line(pid).lower()
    return (
        "-m http.server" in command
        and str(SOURCE_DIR).lower() in command
        and str(port) in command
    )


def terminate_process(pid: int) -> bool:
    result = subprocess.run(
        ["taskkill", "/PID", str(pid), "/T", "/F"],
        stdout=subprocess.DEVNULL,
        stderr=subprocess.DEVNULL,
        check=False,
    )
    return result.returncode == 0


def validate_project() -> bool:
    missing = [path.relative_to(ROOT) for path in REQUIRED_PATHS if not path.exists()]
    if not missing:
        return True

    print("Tasks 项目文件不完整，缺少：")
    for path in missing:
        print(f"  - {path}")
    return False


def server_url(port: int) -> str:
    return f"http://{HOST}:{port}/"


def open_browser(port: int) -> None:
    url = server_url(port)
    if not webbrowser.open(url):
        print(f"无法自动打开浏览器，请手动访问：{url}")


def start(port: int) -> None:
    if not validate_project():
        return

    listener = listener_pid(port)
    if listener:
        if is_tasks_server(listener, port):
            print(f"Tasks 开发服务器已运行：{server_url(port)}（PID {listener}）")
            open_browser(port)
            return

        answer = input(
            f"端口 {port} 已被其他程序使用（PID {listener}）。"
            "是否关闭该进程并强制启动？[y/N]: "
        ).strip().lower()
        if answer not in ("y", "yes"):
            print("已取消启动。")
            return
        if not terminate_process(listener):
            print(f"无法关闭 PID {listener}，请以管理员身份运行终端后重试。")
            return
        time.sleep(0.3)
        if listener_pid(port):
            print(f"端口 {port} 仍被占用，无法启动服务。")
            return
        print(f"已关闭 PID {listener}，正在启动 Tasks 开发服务器。")

    process = subprocess.Popen(
        [
            sys.executable,
            "-m",
            "http.server",
            str(port),
            "--bind",
            HOST,
            "--directory",
            str(SOURCE_DIR),
        ],
        cwd=ROOT,
        creationflags=subprocess.CREATE_NEW_PROCESS_GROUP | subprocess.DETACHED_PROCESS,
        stdout=subprocess.DEVNULL,
        stderr=subprocess.DEVNULL,
    )
    time.sleep(0.5)
    if listener_pid(port) == process.pid:
        print(f"Tasks 开发服务器已启动：{server_url(port)}（PID {process.pid}）")
        open_browser(port)
    else:
        print("服务启动失败，请确认 Python 可用且端口未被占用。")


def stop(port: int) -> None:
    listener = listener_pid(port)
    if not listener or not is_tasks_server(listener, port):
        print(f"端口 {port} 上没有运行 Tasks 开发服务器。")
        return

    if terminate_process(listener):
        print("Tasks 开发服务器已停止。")
    else:
        print(f"无法停止 PID {listener}，请以管理员身份运行终端后重试。")


def status(port: int) -> None:
    listener = listener_pid(port)
    if listener and is_tasks_server(listener, port):
        print(f"Tasks 开发服务器运行中：{server_url(port)}（PID {listener}）")
    elif listener:
        print(f"端口 {port} 已被其他程序使用（PID {listener}）。")
    else:
        print(f"端口 {port} 上没有运行 Tasks 开发服务器。")


def main() -> None:
    parser = argparse.ArgumentParser(description="控制 Tasks 项目的本地开发服务器。")
    parser.add_argument(
        "command",
        choices=("start", "stop", "status"),
        help="启动、停止或查看服务器状态",
    )
    parser.add_argument(
        "--port",
        type=int,
        default=DEFAULT_PORT,
        help=f"HTTP 端口（默认：{DEFAULT_PORT}）",
    )
    args = parser.parse_args()
    if not 1 <= args.port <= 65535:
        parser.error("端口必须在 1 到 65535 之间")

    {"start": start, "stop": stop, "status": status}[args.command](args.port)


if __name__ == "__main__":
    main()