594 lines
18 KiB
Python
Executable File
594 lines
18 KiB
Python
Executable File
#!/usr/bin/env python3
|
||
"""宿舍报修系统 —— 统一项目管理 CLI
|
||
|
||
用法:
|
||
./cli.py setup 安装全部依赖
|
||
./cli.py start [backend|frontend|all] 启动服务
|
||
./cli.py stop [backend|frontend|all] 停止服务
|
||
./cli.py restart [backend|frontend|all] 重启服务
|
||
./cli.py status 查看服务状态
|
||
./cli.py logs [backend|frontend] [-f] 查看日志
|
||
./cli.py seed [--force] [--db PATH] 写入演示数据
|
||
./cli.py lint 运行代码检查
|
||
./cli.py typecheck TypeScript 类型检查
|
||
./cli.py build 生产构建（前端）
|
||
./cli.py clean [--runtime|--db|--deps|--all] 清理临时文件
|
||
./cli.py env 创建 .env 配置
|
||
./cli.py docker [up|down|build] Docker 快捷操作
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import argparse
|
||
import json
|
||
import os
|
||
import signal
|
||
import socket
|
||
import subprocess
|
||
import sys
|
||
import time
|
||
from dataclasses import dataclass
|
||
from pathlib import Path
|
||
from typing import Any
|
||
|
||
IS_WINDOWS = sys.platform == "win32"
|
||
|
||
# Auto-activate virtualenv if present
|
||
_VENV = Path(__file__).resolve().parent / ".venv"
|
||
_VENV_PYTHON = _VENV / ("Scripts" if IS_WINDOWS else "bin") / ("python.exe" if IS_WINDOWS else "python")
|
||
if _VENV_PYTHON.is_file() and Path(sys.executable).resolve() != _VENV_PYTHON.resolve():
|
||
os.execv(str(_VENV_PYTHON), [str(_VENV_PYTHON), *sys.argv])
|
||
|
||
PROJECT_ROOT = Path(__file__).resolve().parent
|
||
BACKEND_DIR = PROJECT_ROOT / "backend"
|
||
FRONTEND_DIR = PROJECT_ROOT / "frontend"
|
||
RUNTIME_DIR = PROJECT_ROOT / ".runtime" / "dev-services"
|
||
STATE_PATH = RUNTIME_DIR / "state.json"
|
||
DATETIME_FORMAT = "%Y-%m-%d %H:%M:%S"
|
||
|
||
# ── Color helpers ──────────────────────────────────────────────
|
||
_BOLD = "\033[1m"
|
||
_GREEN = "\033[32m"
|
||
_YELLOW = "\033[33m"
|
||
_RED = "\033[31m"
|
||
_CYAN = "\033[36m"
|
||
_RESET = "\033[0m"
|
||
|
||
|
||
def _green(s: str) -> str:
|
||
return f"{_GREEN}{s}{_RESET}"
|
||
|
||
|
||
def _yellow(s: str) -> str:
|
||
return f"{_YELLOW}{s}{_RESET}"
|
||
|
||
|
||
def _red(s: str) -> str:
|
||
return f"{_RED}{s}{_RESET}"
|
||
|
||
|
||
def _cyan(s: str) -> str:
|
||
return f"{_CYAN}{s}{_RESET}"
|
||
|
||
|
||
def _bold(s: str) -> str:
|
||
return f"{_BOLD}{s}{_RESET}"
|
||
|
||
|
||
# ── Configuration ──────────────────────────────────────────────
|
||
|
||
|
||
@dataclass(frozen=True)
|
||
class Config:
|
||
backend_host: str = "127.0.0.1"
|
||
backend_port: int = 8000
|
||
frontend_host: str = "127.0.0.1"
|
||
frontend_port: int = 5173
|
||
default_db: Path = BACKEND_DIR / "data" / "dorm_repair.sqlite3"
|
||
|
||
@property
|
||
def backend_url(self) -> str:
|
||
return f"http://{self.backend_host}:{self.backend_port}"
|
||
|
||
@property
|
||
def frontend_url(self) -> str:
|
||
return f"http://{self.frontend_host}:{self.frontend_port}"
|
||
|
||
@property
|
||
def backend_health_url(self) -> str:
|
||
return f"{self.backend_url}/api/health"
|
||
|
||
|
||
CONFIG = Config()
|
||
|
||
|
||
# ── Service definitions ────────────────────────────────────────
|
||
|
||
|
||
@dataclass(frozen=True)
|
||
class Service:
|
||
name: str
|
||
command: list[str]
|
||
cwd: Path
|
||
url: str
|
||
host: str
|
||
port: int
|
||
env: dict[str, str] | None = None
|
||
|
||
@property
|
||
def log_path(self) -> Path:
|
||
return RUNTIME_DIR / f"{self.name}.log"
|
||
|
||
|
||
SERVICES = {
|
||
"backend": Service(
|
||
name="backend",
|
||
command=[
|
||
"uv",
|
||
"run",
|
||
"uvicorn",
|
||
"app.main:app",
|
||
"--reload",
|
||
"--host",
|
||
CONFIG.backend_host,
|
||
"--port",
|
||
str(CONFIG.backend_port),
|
||
],
|
||
cwd=BACKEND_DIR,
|
||
url=CONFIG.backend_health_url,
|
||
host=CONFIG.backend_host,
|
||
port=CONFIG.backend_port,
|
||
env={"UV_CACHE_DIR": str(RUNTIME_DIR / "uv-cache")},
|
||
),
|
||
"frontend": Service(
|
||
name="frontend",
|
||
command=[
|
||
"pnpm",
|
||
"dev",
|
||
"--host",
|
||
CONFIG.frontend_host,
|
||
"--port",
|
||
str(CONFIG.frontend_port),
|
||
"--strictPort",
|
||
],
|
||
cwd=FRONTEND_DIR,
|
||
url=CONFIG.frontend_url + "/",
|
||
host=CONFIG.frontend_host,
|
||
port=CONFIG.frontend_port,
|
||
),
|
||
}
|
||
|
||
|
||
# ── State management ───────────────────────────────────────────
|
||
|
||
|
||
def _load_state() -> dict[str, Any]:
|
||
if not STATE_PATH.exists():
|
||
return {}
|
||
try:
|
||
return json.loads(STATE_PATH.read_text(encoding="utf-8"))
|
||
except json.JSONDecodeError:
|
||
return {}
|
||
|
||
|
||
def _save_state(state: dict[str, Any]) -> None:
|
||
RUNTIME_DIR.mkdir(parents=True, exist_ok=True)
|
||
STATE_PATH.write_text(json.dumps(state, indent=2, ensure_ascii=False), encoding="utf-8")
|
||
|
||
|
||
# ── Process helpers ────────────────────────────────────────────
|
||
|
||
|
||
def _pid_running(pid: int) -> bool:
|
||
if IS_WINDOWS:
|
||
result = subprocess.run(
|
||
["tasklist", "/FI", f"PID eq {pid}", "/NH"],
|
||
capture_output=True,
|
||
text=True,
|
||
)
|
||
return str(pid) in result.stdout
|
||
try:
|
||
os.kill(pid, 0)
|
||
except ProcessLookupError:
|
||
return False
|
||
except PermissionError:
|
||
return True
|
||
return True
|
||
|
||
|
||
def _service_pid(state: dict[str, Any], svc: Service) -> int | None:
|
||
raw = state.get(svc.name, {}).get("pid")
|
||
if not isinstance(raw, int):
|
||
return None
|
||
return raw if _pid_running(raw) else None
|
||
|
||
|
||
def _port_open(host: str, port: int) -> bool:
|
||
with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock:
|
||
sock.settimeout(0.2)
|
||
return sock.connect_ex((host, port)) == 0
|
||
|
||
|
||
# ── Service lifecycle ──────────────────────────────────────────
|
||
|
||
|
||
def _start_service(svc: Service, state: dict[str, Any]) -> None:
|
||
pid = _service_pid(state, svc)
|
||
if pid is not None:
|
||
print(f" {svc.name}: 已在运行 pid={pid}")
|
||
return
|
||
|
||
if _port_open(svc.host, svc.port):
|
||
print(f" {_yellow(svc.name)}: 端口 {svc.host}:{svc.port} 已被占用")
|
||
return
|
||
|
||
RUNTIME_DIR.mkdir(parents=True, exist_ok=True)
|
||
log_file = svc.log_path.open("ab")
|
||
popen_kwargs: dict[str, Any] = {
|
||
"cwd": svc.cwd,
|
||
"env": {**os.environ, **(svc.env or {})},
|
||
"stdin": subprocess.DEVNULL,
|
||
"stdout": log_file,
|
||
"stderr": subprocess.STDOUT,
|
||
}
|
||
if IS_WINDOWS and hasattr(subprocess, "CREATE_NEW_PROCESS_GROUP"):
|
||
popen_kwargs["creationflags"] = subprocess.CREATE_NEW_PROCESS_GROUP
|
||
else:
|
||
popen_kwargs["start_new_session"] = True
|
||
proc = subprocess.Popen(svc.command, **popen_kwargs)
|
||
state[svc.name] = {
|
||
"pid": proc.pid,
|
||
"command": svc.command,
|
||
"cwd": str(svc.cwd),
|
||
"log": str(svc.log_path),
|
||
"url": svc.url,
|
||
"started_at": time.strftime(DATETIME_FORMAT),
|
||
}
|
||
print(f" {_green(svc.name)}: 已启动 pid={proc.pid}")
|
||
|
||
|
||
def _stop_service(svc: Service, state: dict[str, Any]) -> None:
|
||
pid = _service_pid(state, svc)
|
||
if pid is None:
|
||
print(f" {svc.name}: 未运行")
|
||
state.pop(svc.name, None)
|
||
return
|
||
|
||
print(f" {svc.name}: 正在停止 pid={pid}")
|
||
|
||
if IS_WINDOWS:
|
||
subprocess.run(["taskkill", "/PID", str(pid), "/T"], capture_output=True)
|
||
else:
|
||
try:
|
||
os.killpg(pid, signal.SIGTERM)
|
||
except ProcessLookupError:
|
||
state.pop(svc.name, None)
|
||
return
|
||
|
||
deadline = time.time() + 8
|
||
while time.time() < deadline:
|
||
if not _pid_running(pid):
|
||
state.pop(svc.name, None)
|
||
print(f" {svc.name}: 已停止")
|
||
return
|
||
time.sleep(0.2)
|
||
|
||
print(f" {_yellow(svc.name)}: 未响应，强制终止")
|
||
if IS_WINDOWS:
|
||
subprocess.run(["taskkill", "/F", "/PID", str(pid), "/T"], capture_output=True)
|
||
else:
|
||
try:
|
||
os.killpg(pid, signal.SIGKILL)
|
||
except ProcessLookupError:
|
||
pass
|
||
state.pop(svc.name, None)
|
||
|
||
|
||
def _status_service(svc: Service, state: dict[str, Any]) -> None:
|
||
pid = _service_pid(state, svc)
|
||
if pid is None:
|
||
suffix = " (端口被占用)" if _port_open(svc.host, svc.port) else ""
|
||
print(f" {svc.name}: 未运行{suffix}")
|
||
return
|
||
print(f" {_green(svc.name)}: 运行中 pid={pid} url={svc.url}")
|
||
|
||
|
||
def _pick_services(names: list[str]) -> list[Service]:
|
||
if not names or "all" in names:
|
||
return list(SERVICES.values())
|
||
selected: list[Service] = []
|
||
for name in names:
|
||
svc = SERVICES.get(name)
|
||
if svc is None:
|
||
choices = ", ".join(["all", *SERVICES])
|
||
raise SystemExit(f"未知服务 '{name}'，可选: {choices}")
|
||
selected.append(svc)
|
||
return selected
|
||
|
||
|
||
# ── Commands ───────────────────────────────────────────────────
|
||
|
||
|
||
def cmd_setup() -> int:
|
||
"""安装后端和前端依赖。"""
|
||
print(_bold("安装后端依赖 (uv sync) ..."))
|
||
subprocess.run(["uv", "sync"], cwd=BACKEND_DIR, check=True)
|
||
print(_bold("安装前端依赖 (pnpm install) ..."))
|
||
subprocess.run(["pnpm", "install"], cwd=FRONTEND_DIR, check=True)
|
||
print(_green("安装完成。"))
|
||
return 0
|
||
|
||
|
||
def cmd_start(services: list[str]) -> int:
|
||
state = _load_state()
|
||
svcs = _pick_services(services)
|
||
for svc in svcs:
|
||
_start_service(svc, state)
|
||
_save_state(state)
|
||
if svcs:
|
||
_print_urls(svcs)
|
||
return 0
|
||
|
||
|
||
def _print_urls(svcs: list[Service]) -> None:
|
||
print()
|
||
for svc in svcs:
|
||
label = "后端" if svc.name == "backend" else "前端"
|
||
print(f" {label}: {_cyan(svc.url)}")
|
||
print()
|
||
|
||
|
||
def cmd_stop(services: list[str]) -> int:
|
||
state = _load_state()
|
||
for svc in reversed(_pick_services(services)):
|
||
_stop_service(svc, state)
|
||
_save_state(state)
|
||
return 0
|
||
|
||
|
||
def cmd_restart(services: list[str]) -> int:
|
||
state = _load_state()
|
||
svcs = _pick_services(services)
|
||
for svc in reversed(svcs):
|
||
_stop_service(svc, state)
|
||
for svc in svcs:
|
||
_start_service(svc, state)
|
||
_save_state(state)
|
||
if svcs:
|
||
_print_urls(svcs)
|
||
return 0
|
||
|
||
|
||
def cmd_status() -> int:
|
||
state = _load_state()
|
||
for svc in SERVICES.values():
|
||
_status_service(svc, state)
|
||
return 0
|
||
|
||
|
||
def cmd_logs(service: str, follow: bool) -> int:
|
||
svc = SERVICES.get(service)
|
||
if svc is None:
|
||
raise SystemExit(f"未知服务 '{service}'，可选: backend, frontend")
|
||
log_path = svc.log_path
|
||
if not log_path.exists():
|
||
print(f"日志文件不存在: {log_path}")
|
||
return 0
|
||
|
||
if follow:
|
||
_follow_log(log_path)
|
||
else:
|
||
content = log_path.read_text(encoding="utf-8", errors="replace")
|
||
for line in content.splitlines()[-50:]:
|
||
print(line)
|
||
return 0
|
||
|
||
|
||
def _follow_log(path: Path) -> None:
|
||
if IS_WINDOWS:
|
||
with path.open("r", encoding="utf-8", errors="replace") as f:
|
||
f.seek(0, 2)
|
||
while True:
|
||
line = f.readline()
|
||
if line:
|
||
print(line, end="")
|
||
else:
|
||
time.sleep(0.5)
|
||
else:
|
||
subprocess.run(["tail", "-n", "50", "-f", str(path)])
|
||
|
||
|
||
def cmd_seed(force: bool, db_path: str | None) -> int:
|
||
"""Seed demo data via standalone seed script."""
|
||
db = db_path or str(CONFIG.default_db)
|
||
seed_script = PROJECT_ROOT / "scripts" / "_seed.py"
|
||
print(_bold("Seeding demo data ..."))
|
||
subprocess.run(
|
||
["uv", "run", "python", str(seed_script), "--db", db] + (["--force"] if force else []),
|
||
cwd=BACKEND_DIR,
|
||
check=True,
|
||
)
|
||
print(f" student01 / Student123 {_cyan('(student)')}")
|
||
print(f" admin01 / Admin123 {_cyan('(admin)')}")
|
||
print(" 8 sample repair orders covering all statuses")
|
||
return 0
|
||
|
||
|
||
def cmd_lint() -> int:
|
||
print(_bold("Backend lint (ruff) ..."))
|
||
subprocess.run(["uv", "run", "ruff", "check", "backend"], check=False)
|
||
subprocess.run(["uv", "run", "ruff", "format", "--check", "backend"], check=False)
|
||
print(_bold("Frontend lint (eslint) ..."))
|
||
subprocess.run(["pnpm", "lint"], cwd=FRONTEND_DIR, check=False)
|
||
return 0
|
||
|
||
|
||
def cmd_typecheck() -> int:
|
||
print(_bold("TypeScript type check ..."))
|
||
subprocess.run(["pnpm", "typecheck"], cwd=FRONTEND_DIR, check=False)
|
||
return 0
|
||
|
||
|
||
def cmd_build() -> int:
|
||
print(_bold("Building frontend ..."))
|
||
subprocess.run(["pnpm", "build"], cwd=FRONTEND_DIR, check=True)
|
||
print(_green(f"Output: {FRONTEND_DIR / 'dist'}/"))
|
||
return 0
|
||
|
||
|
||
def cmd_clean(runtime: bool, db: bool, deps: bool, all_: bool) -> int:
|
||
if all_:
|
||
runtime = db = deps = True
|
||
|
||
if runtime:
|
||
if RUNTIME_DIR.exists():
|
||
import shutil
|
||
|
||
shutil.rmtree(RUNTIME_DIR)
|
||
print(f"Removed {RUNTIME_DIR}")
|
||
else:
|
||
print("No runtime dir to clean.")
|
||
|
||
if db:
|
||
db_path = CONFIG.default_db
|
||
if db_path.exists():
|
||
db_path.unlink()
|
||
print(f"Removed {db_path}")
|
||
else:
|
||
print("No database to clean.")
|
||
|
||
if deps:
|
||
for d in [BACKEND_DIR / ".venv", FRONTEND_DIR / "node_modules"]:
|
||
if d.exists():
|
||
import shutil
|
||
|
||
shutil.rmtree(d)
|
||
print(f"Removed {d}")
|
||
|
||
print(_green("Clean complete."))
|
||
return 0
|
||
|
||
|
||
def cmd_env() -> int:
|
||
for template, target in [
|
||
(BACKEND_DIR / ".env.example", BACKEND_DIR / ".env"),
|
||
(FRONTEND_DIR / ".env.example", FRONTEND_DIR / ".env"),
|
||
]:
|
||
if not template.exists():
|
||
print(f"Template not found: {template}")
|
||
continue
|
||
if target.exists():
|
||
print(f"{_yellow('Skipped')} {target} (already exists)")
|
||
continue
|
||
target.write_text(template.read_text(encoding="utf-8"), encoding="utf-8")
|
||
print(f"Created {target} from {template.name}")
|
||
print(_green("Environment files ready. Edit them if needed:"))
|
||
print(f" {BACKEND_DIR / '.env'}")
|
||
print(f" {FRONTEND_DIR / '.env'}")
|
||
return 0
|
||
|
||
|
||
def cmd_docker(action: str) -> int:
|
||
compose_file = PROJECT_ROOT / "docker-compose.yml"
|
||
if not compose_file.exists():
|
||
raise SystemExit("docker-compose.yml not found.")
|
||
if action == "up":
|
||
subprocess.run(["docker", "compose", "up", "-d"], check=True)
|
||
elif action == "down":
|
||
subprocess.run(["docker", "compose", "down"], check=True)
|
||
elif action == "build":
|
||
subprocess.run(["docker", "compose", "build"], check=True)
|
||
else:
|
||
raise SystemExit(f"Unknown docker action: {action}")
|
||
return 0
|
||
|
||
|
||
# ── Argument parsing ───────────────────────────────────────────
|
||
|
||
|
||
def _parse_args(argv: list[str] | None = None) -> argparse.Namespace:
|
||
parser = argparse.ArgumentParser(
|
||
description="宿舍报修系统 —— 项目管理 CLI",
|
||
formatter_class=argparse.RawDescriptionHelpFormatter,
|
||
)
|
||
sub = parser.add_subparsers(dest="command", required=True)
|
||
|
||
sub.add_parser("setup", help="安装全部依赖（uv sync + pnpm install）")
|
||
|
||
for cmd_name in ("start", "stop", "restart"):
|
||
label = {"start": "启动", "stop": "停止", "restart": "重启"}[cmd_name]
|
||
p = sub.add_parser(cmd_name, help=f"{label}服务")
|
||
p.add_argument(
|
||
"services",
|
||
nargs="*",
|
||
default=["all"],
|
||
help="目标服务：all, backend, frontend（默认: all）",
|
||
)
|
||
|
||
sub.add_parser("status", help="查看服务状态")
|
||
|
||
log_p = sub.add_parser("logs", help="查看服务日志")
|
||
log_p.add_argument("service", choices=["backend", "frontend"], help="服务名")
|
||
log_p.add_argument("-f", "--follow", action="store_true", help="实时跟踪日志")
|
||
|
||
seed_p = sub.add_parser("seed", help="写入演示数据")
|
||
seed_p.add_argument("--force", action="store_true", help="覆盖已有数据")
|
||
seed_p.add_argument("--db", type=str, default=None, help="SQLite 数据库路径")
|
||
|
||
sub.add_parser("lint", help="运行代码检查（ruff + eslint）")
|
||
sub.add_parser("typecheck", help="TypeScript 类型检查（vue-tsc）")
|
||
sub.add_parser("build", help="生产构建（前端）")
|
||
|
||
clean_p = sub.add_parser("clean", help="清理临时文件")
|
||
clean_p.add_argument("--runtime", action="store_true", help="删除 .runtime/")
|
||
clean_p.add_argument("--db", action="store_true", help="删除 SQLite 数据库")
|
||
clean_p.add_argument("--deps", action="store_true", help="删除 .venv 和 node_modules")
|
||
clean_p.add_argument("--all", action="store_true", help="清理全部")
|
||
|
||
sub.add_parser("env", help="从 .env.example 创建 .env 配置文件")
|
||
|
||
docker_p = sub.add_parser("docker", help="Docker 快捷操作")
|
||
docker_p.add_argument("action", choices=["up", "down", "build"], help="Docker 操作")
|
||
|
||
return parser.parse_args(argv)
|
||
|
||
|
||
# ── Main ───────────────────────────────────────────────────────
|
||
|
||
|
||
def main(argv: list[str] | None = None) -> int:
|
||
args = _parse_args(argv)
|
||
cmd = args.command
|
||
|
||
if cmd == "setup":
|
||
return cmd_setup()
|
||
elif cmd == "start":
|
||
return cmd_start(args.services)
|
||
elif cmd == "stop":
|
||
return cmd_stop(args.services)
|
||
elif cmd == "restart":
|
||
return cmd_restart(args.services)
|
||
elif cmd == "status":
|
||
return cmd_status()
|
||
elif cmd == "logs":
|
||
return cmd_logs(args.service, args.follow)
|
||
elif cmd == "seed":
|
||
return cmd_seed(args.force, args.db)
|
||
elif cmd == "lint":
|
||
return cmd_lint()
|
||
elif cmd == "typecheck":
|
||
return cmd_typecheck()
|
||
elif cmd == "build":
|
||
return cmd_build()
|
||
elif cmd == "clean":
|
||
return cmd_clean(args.runtime, args.db, args.deps, args.all)
|
||
elif cmd == "env":
|
||
return cmd_env()
|
||
elif cmd == "docker":
|
||
return cmd_docker(args.action)
|
||
return 1
|
||
|
||
|
||
if __name__ == "__main__":
|
||
raise SystemExit(main())
|