init
This commit is contained in:
@@ -0,0 +1,119 @@
|
||||
"""Windows development server controls for the src application."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import subprocess
|
||||
import sys
|
||||
import time
|
||||
from pathlib import Path
|
||||
|
||||
ROOT = Path(__file__).resolve().parent
|
||||
SOURCE_DIR = ROOT / "src"
|
||||
PORT = 8000
|
||||
|
||||
|
||||
def listener_pid() -> int | None:
|
||||
result = subprocess.run(
|
||||
["powershell", "-NoProfile", "-Command", f"(Get-NetTCPConnection -LocalPort {PORT} -State Listen -ErrorAction SilentlyContinue | Select-Object -First 1 -ExpandProperty OwningProcess)"],
|
||||
capture_output=True,
|
||||
text=True,
|
||||
check=False,
|
||||
)
|
||||
try:
|
||||
return int(result.stdout.strip())
|
||||
except ValueError:
|
||||
return None
|
||||
|
||||
|
||||
def command_line(pid: int) -> str:
|
||||
result = subprocess.run(
|
||||
["powershell", "-NoProfile", "-Command", f"(Get-CimInstance Win32_Process -Filter 'ProcessId = {pid}').CommandLine"],
|
||||
capture_output=True,
|
||||
text=True,
|
||||
check=False,
|
||||
)
|
||||
return result.stdout.strip()
|
||||
|
||||
|
||||
def is_src_server(pid: int) -> bool:
|
||||
command = command_line(pid).lower()
|
||||
return "-m http.server" in command and str(SOURCE_DIR).lower() 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 start() -> None:
|
||||
listener = listener_pid()
|
||||
if listener:
|
||||
if is_src_server(listener):
|
||||
print(f"服务已运行:http://localhost:{PORT}/(PID {listener})")
|
||||
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():
|
||||
print(f"端口 {PORT} 仍被占用,无法启动服务。")
|
||||
return
|
||||
print(f"已关闭 PID {listener},正在启动服务。")
|
||||
|
||||
process = subprocess.Popen(
|
||||
[sys.executable, "-m", "http.server", str(PORT), "--directory", str(SOURCE_DIR)],
|
||||
cwd=ROOT,
|
||||
creationflags=subprocess.CREATE_NEW_PROCESS_GROUP | subprocess.DETACHED_PROCESS,
|
||||
stdout=subprocess.DEVNULL,
|
||||
stderr=subprocess.DEVNULL,
|
||||
)
|
||||
time.sleep(0.3)
|
||||
if listener_pid() == process.pid:
|
||||
print(f"服务已启动:http://localhost:{PORT}/(PID {process.pid})")
|
||||
else:
|
||||
print("服务启动失败,请确认 Python 可用且端口未被占用。")
|
||||
|
||||
|
||||
def stop() -> None:
|
||||
listener = listener_pid()
|
||||
if not listener or not is_src_server(listener):
|
||||
print("src 开发服务器未启动。")
|
||||
return
|
||||
|
||||
if terminate_process(listener):
|
||||
print("服务已停止。")
|
||||
else:
|
||||
print(f"无法停止 PID {listener},请以管理员身份运行终端后重试。")
|
||||
|
||||
|
||||
def status() -> None:
|
||||
listener = listener_pid()
|
||||
if listener and is_src_server(listener):
|
||||
print(f"服务运行中:http://localhost:{PORT}/(PID {listener})")
|
||||
else:
|
||||
print("src 开发服务器未启动。")
|
||||
|
||||
|
||||
def main() -> None:
|
||||
parser = argparse.ArgumentParser(description="控制 src 的本地 HTTP 开发服务器。")
|
||||
parser.add_argument("command", choices=("start", "stop", "status"), help="执行的操作")
|
||||
args = parser.parse_args()
|
||||
{"start": start, "stop": stop, "status": status}[args.command]()
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
Reference in New Issue
Block a user