#!/usr/bin/env python3 """启动两个本地服务，等待 HTTP 就绪后执行检查并清理进程。""" from __future__ import annotations import argparse import os import shutil import signal import subprocess import sys import tempfile import time import urllib.error import urllib.request from dataclasses import dataclass from pathlib import Path @dataclass class Service: name: str command: str cwd: Path url: str process: subprocess.Popen[bytes] | None = None log_path: Path | None = None log_file: object | None = None def parse_args() -> argparse.Namespace: parser = argparse.ArgumentParser( description="启动前后端命令，检查 HTTP 地址并清理进程。" ) parser.add_argument("--backend-cmd", required=True, help="后端启动命令") parser.add_argument("--backend-dir", default=".", help="后端工作目录") parser.add_argument("--backend-url", required=True, help="后端健康检查地址") parser.add_argument("--frontend-cmd", required=True, help="前端启动命令") parser.add_argument("--frontend-dir", default=".", help="前端工作目录") parser.add_argument("--frontend-url", required=True, help="前端就绪检查地址") parser.add_argument( "--check-url", action="append", default=[], metavar="LABEL=URL", help="额外的 GET 检查，可重复传入", ) parser.add_argument( "--timeout", type=float, default=60.0, help="每个服务的就绪超时时间，单位为秒，默认 60", ) parser.add_argument( "--interval", type=float, default=1.0, help="每次就绪检查的间隔秒数，默认 1", ) parser.add_argument( "--log-dir", type=Path, help="日志目录，运行结束后保留", ) parser.add_argument( "--keep-logs", action="store_true", help="运行结束后保留临时日志", ) parser.add_argument( "--keep-running", action="store_true", help="验证结束后保持已启动的服务运行", ) return parser.parse_args() def resolve_dir(value: str) -> Path: path = Path(value).resolve() if not path.is_dir(): raise RuntimeError(f"工作目录不存在：{path}") return path def start_service(service: Service, log_dir: Path) -> None: service.log_path = log_dir / f"{service.name}.log" service.log_file = service.log_path.open("w", encoding="utf-8", buffering=1) creationflags = 0 start_new_session = False if os.name == "nt": creationflags = getattr(subprocess, "CREATE_NEW_PROCESS_GROUP", 0) else: start_new_session = True try: service.process = subprocess.Popen( service.command, cwd=service.cwd, shell=True, stdout=service.log_file, stderr=subprocess.STDOUT, creationflags=creationflags, start_new_session=start_new_session, ) except Exception: service.log_file.close() service.log_file = None raise print(f"启动 {service.name}：进程={service.process.pid}，工作目录={service.cwd}") def request(url: str, timeout: float) -> tuple[bool, str]: request_obj = urllib.request.Request(url, headers={"User-Agent": "fullstack-start-verify/1.0"}) try: with urllib.request.urlopen(request_obj, timeout=timeout) as response: status = response.status return 200 <= status < 400, str(status) except urllib.error.HTTPError as error: return 200 <= error.code < 400, str(error.code) except (urllib.error.URLError, TimeoutError, OSError) as error: reason = getattr(error, "reason", error) return False, type(reason).__name__ def wait_for_service(service: Service, timeout: float, interval: float) -> None: if service.process is None: raise RuntimeError(f"{service.name} 未启动") deadline = time.monotonic() + timeout last_reason = "not checked" while time.monotonic() < deadline: exit_code = service.process.poll() if exit_code is not None: raise RuntimeError(f"{service.name} 在就绪检查前退出，退出码为 {exit_code}") ok, reason = request(service.url, min(interval, 5.0)) last_reason = reason if ok: print(f"通过 {service.name}：{service.url} [{reason}]") return time.sleep(max(interval, 0.05)) raise RuntimeError(f"{service.name} 在 {timeout:g} 秒内未就绪 [{last_reason}]") def parse_extra_checks(values: list[str]) -> list[tuple[str, str]]: checks: list[tuple[str, str]] = [] for value in values: if "=" in value: label, url = value.split("=", 1) else: label, url = value, value label = label.strip() or url.strip() url = url.strip() if not url.startswith(("http://", "https://")): raise RuntimeError(f"检查地址无效：{url}") checks.append((label, url)) return checks def run_extra_checks(checks: list[tuple[str, str]]) -> None: for label, url in checks: ok, reason = request(url, 10.0) if not ok: raise RuntimeError(f"检查失败：{label} {url} [{reason}]") print(f"通过 {label}：{url} [{reason}]") def stop_service(service: Service) -> None: process = service.process if process is None or process.poll() is not None: return if os.name == "nt": subprocess.run( ["taskkill", "/PID", str(process.pid), "/T", "/F"], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, check=False, ) try: process.wait(timeout=5) except subprocess.TimeoutExpired: process.kill() process.wait(timeout=5) except OSError: # taskkill 已经结束进程，但 Windows 可能同时使句柄失效。 process.returncode = -1 else: try: os.killpg(process.pid, signal.SIGTERM) except ProcessLookupError: return try: process.wait(timeout=5) except subprocess.TimeoutExpired: os.killpg(process.pid, signal.SIGKILL) def tail(path: Path | None, lines: int = 20) -> list[str]: if path is None or not path.exists(): return [] return path.read_text(encoding="utf-8", errors="replace").splitlines()[-lines:] def main() -> int: args = parse_args() temp_log_dir: Path | None = None log_dir = args.log_dir if log_dir is None: temp_log_dir = Path(tempfile.mkdtemp(prefix="fullstack-start-verify-")) log_dir = temp_log_dir log_dir = log_dir.resolve() log_dir.mkdir(parents=True, exist_ok=True) services = [ Service("backend", args.backend_cmd, resolve_dir(args.backend_dir), args.backend_url), Service("frontend", args.frontend_cmd, resolve_dir(args.frontend_dir), args.frontend_url), ] failure: Exception | None = None try: for service in services: start_service(service, log_dir) for service in services: wait_for_service(service, args.timeout, args.interval) run_extra_checks(parse_extra_checks(args.check_url)) print("全栈验证通过（FULLSTACK_VERIFY=PASS）") return 0 except (OSError, RuntimeError, ValueError) as error: failure = error print(f"全栈验证失败：{error}", file=sys.stderr) for service in services: for line in tail(service.log_path): print(f"日志 {service.name}：{line}", file=sys.stderr) return 1 finally: if not args.keep_running: for service in reversed(services): stop_service(service) for service in services: if service.log_file is not None: service.log_file.close() if args.keep_running: print(f"服务仍在运行，日志目录：{log_dir}") elif temp_log_dir is not None and not args.keep_logs: shutil.rmtree(temp_log_dir, ignore_errors=True) elif temp_log_dir is not None: print(f"日志已保留：{temp_log_dir}") elif failure is not None: print(f"日志已保留：{log_dir}") if __name__ == "__main__": raise SystemExit(main())