dekun
fa90a29576
Avoid unbound BASH_SOURCE under set -u, attach stdin to /dev/tty for interactive prompts, and handle empty menu input without looping. Co-authored-by: Cursor <cursoragent@cursor.com>
340 lines
9.9 KiB
Python
340 lines
9.9 KiB
Python
#!/usr/bin/env python3
|
||
"""LocalNav 交互式部署管理：安装 / 卸载 / 更新。"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import argparse
|
||
import platform
|
||
import shutil
|
||
import subprocess
|
||
import sys
|
||
from pathlib import Path
|
||
|
||
from nav_common import (
|
||
APP_NAME,
|
||
MIN_PYTHON,
|
||
default_install_dir,
|
||
default_repo,
|
||
is_localnav_project,
|
||
load_install_config,
|
||
pm2_app_status,
|
||
resolve_project_dir,
|
||
save_install_config,
|
||
service_status_text,
|
||
stop_service,
|
||
)
|
||
|
||
SCRIPT_DIR = Path(__file__).resolve().parent
|
||
|
||
|
||
def _ok(msg: str) -> None:
|
||
print(f"[OK] {msg}")
|
||
|
||
|
||
def _warn(msg: str) -> None:
|
||
print(f"[WARN] {msg}")
|
||
|
||
|
||
def _fail(msg: str) -> None:
|
||
print(f"[FAIL] {msg}", file=sys.stderr)
|
||
sys.exit(1)
|
||
|
||
|
||
def _pause() -> None:
|
||
if not sys.stdin.isatty():
|
||
return
|
||
try:
|
||
input("\n按回车返回菜单...")
|
||
except (EOFError, KeyboardInterrupt):
|
||
print()
|
||
|
||
|
||
def _require_tty(action: str) -> None:
|
||
if sys.stdin.isatty():
|
||
return
|
||
print(f"[FAIL] 当前环境无法交互式{action}（stdin 不是终端）", file=sys.stderr)
|
||
print("请使用以下方式之一：", file=sys.stderr)
|
||
print(" bash scripts/install.sh", file=sys.stderr)
|
||
print(" python3 scripts/navctl.py install -y", file=sys.stderr)
|
||
print(" curl -fsSL .../install.sh | bash -s -- install -y", file=sys.stderr)
|
||
sys.exit(1)
|
||
|
||
|
||
def prompt_value(label: str, default: str) -> str:
|
||
try:
|
||
raw = input(f"{label} [{default}]: ").strip()
|
||
except (EOFError, KeyboardInterrupt):
|
||
print()
|
||
_fail("已取消")
|
||
return raw or default
|
||
|
||
|
||
def confirm(prompt: str, *, default_no: bool = True) -> bool:
|
||
suffix = " [y/N]: " if default_no else " [Y/n]: "
|
||
try:
|
||
raw = input(prompt + suffix).strip().lower()
|
||
except (EOFError, KeyboardInterrupt):
|
||
print()
|
||
return False
|
||
if not raw:
|
||
return not default_no
|
||
return raw in ("y", "yes", "是")
|
||
|
||
|
||
def check_python() -> str:
|
||
ver = sys.version_info
|
||
if ver < MIN_PYTHON:
|
||
_fail(f"需要 Python {MIN_PYTHON[0]}.{MIN_PYTHON[1]}+，当前 {ver.major}.{ver.minor}")
|
||
return sys.executable
|
||
|
||
|
||
def check_git() -> str:
|
||
git = shutil.which("git")
|
||
if not git:
|
||
_fail("未检测到 git，请先安装：sudo apt install -y git")
|
||
return git
|
||
|
||
|
||
def clone_project(git_exe: str, repo: str, dest: Path) -> None:
|
||
if dest.exists() and not dest.is_dir():
|
||
_fail(f"安装路径已存在且不是目录: {dest}")
|
||
|
||
if is_localnav_project(dest):
|
||
_ok(f"目录已是 LocalNav 项目: {dest}")
|
||
return
|
||
|
||
if dest.exists():
|
||
if (dest / ".git").is_dir():
|
||
_fail(f"目录已存在且不是 LocalNav 项目: {dest}")
|
||
_fail(f"目录已存在: {dest}")
|
||
|
||
dest.parent.mkdir(parents=True, exist_ok=True)
|
||
_ok(f"正在克隆 {repo} -> {dest}")
|
||
subprocess.run([git_exe, "clone", repo, str(dest)], check=True)
|
||
if not is_localnav_project(dest):
|
||
_fail("克隆完成但未找到 app.py，请检查仓库地址")
|
||
|
||
|
||
def git_pull(project_dir: Path) -> None:
|
||
git = check_git()
|
||
if not (project_dir / ".git").is_dir():
|
||
_fail(f"目录不是 Git 仓库，无法更新: {project_dir}")
|
||
_ok(f"正在拉取最新代码: {project_dir}")
|
||
subprocess.run([git, "-C", str(project_dir), "pull", "--ff-only"], check=True)
|
||
|
||
|
||
def run_deploy(project_dir: Path, py_exe: str) -> None:
|
||
deploy_script = project_dir / "scripts" / "deploy.py"
|
||
if not deploy_script.is_file():
|
||
_fail(f"未找到部署脚本: {deploy_script}")
|
||
subprocess.run([py_exe, str(deploy_script)], cwd=project_dir, check=True)
|
||
|
||
|
||
def action_install(*, repo: str = "", dest: str = "", yes: bool = False) -> None:
|
||
if not yes:
|
||
_require_tty("安装")
|
||
print()
|
||
print("=" * 50)
|
||
print(" 一键部署安装")
|
||
print("=" * 50)
|
||
|
||
py_exe = check_python()
|
||
git_exe = check_git()
|
||
|
||
repo = repo.strip() or default_repo()
|
||
if yes:
|
||
project_dir = Path(dest).expanduser() if dest.strip() else default_install_dir()
|
||
else:
|
||
repo = prompt_value("Git 仓库地址", repo)
|
||
dest_str = dest.strip() or prompt_value("安装目录", str(default_install_dir()))
|
||
project_dir = Path(dest_str).expanduser()
|
||
|
||
project_dir = project_dir.resolve()
|
||
print()
|
||
print(f" 仓库: {repo}")
|
||
print(f" 目录: {project_dir}")
|
||
print()
|
||
|
||
clone_project(git_exe, repo, project_dir)
|
||
run_deploy(project_dir, py_exe)
|
||
save_install_config(project_dir, repo)
|
||
_ok(f"安装完成: {project_dir}")
|
||
|
||
|
||
def action_update(*, dest: str = "") -> None:
|
||
print()
|
||
print("=" * 50)
|
||
print(" 更新")
|
||
print("=" * 50)
|
||
|
||
project_dir = resolve_project_dir(dest)
|
||
if not project_dir:
|
||
dest_str = dest.strip() or prompt_value("安装目录", str(default_install_dir()))
|
||
project_dir = Path(dest_str).expanduser().resolve()
|
||
if not is_localnav_project(project_dir):
|
||
_fail(f"未找到 LocalNav 安装: {project_dir}")
|
||
|
||
py_exe = check_python()
|
||
git_pull(project_dir)
|
||
run_deploy(project_dir, py_exe)
|
||
cfg = load_install_config()
|
||
save_install_config(project_dir, cfg.get("repo") or default_repo())
|
||
_ok(f"更新完成: {project_dir}")
|
||
|
||
|
||
def action_uninstall(*, dest: str = "") -> None:
|
||
_require_tty("卸载")
|
||
print()
|
||
print("=" * 50)
|
||
print(" 一键卸载")
|
||
print("=" * 50)
|
||
|
||
project_dir = resolve_project_dir(dest)
|
||
if not project_dir:
|
||
dest_str = dest.strip() or prompt_value("安装目录", str(default_install_dir()))
|
||
candidate = Path(dest_str).expanduser().resolve()
|
||
if candidate.exists() and is_localnav_project(candidate):
|
||
project_dir = candidate
|
||
elif not candidate.exists():
|
||
_warn(f"目录不存在: {candidate}")
|
||
project_dir = None
|
||
else:
|
||
_fail(f"目录不是 LocalNav 项目: {candidate}")
|
||
|
||
print()
|
||
if project_dir:
|
||
print(f" 安装目录: {project_dir}")
|
||
print(f" 服务状态: {service_status_text(project_dir)}")
|
||
else:
|
||
print(" 未检测到已安装的 LocalNav 项目")
|
||
print()
|
||
|
||
if not confirm("确认停止 nav-site 服务？", default_no=True):
|
||
_warn("已取消卸载")
|
||
return
|
||
|
||
actions = stop_service()
|
||
if actions:
|
||
for item in actions:
|
||
_ok(item)
|
||
else:
|
||
_warn("未发现 pm2 / systemd 中的 nav-site 进程")
|
||
|
||
if not project_dir:
|
||
_ok("卸载完成（仅停止服务）")
|
||
return
|
||
|
||
if confirm("是否删除整个安装目录（含数据库，不可恢复）？", default_no=True):
|
||
if not confirm("再次确认：删除 " + str(project_dir) + " ？", default_no=True):
|
||
_warn("已保留安装目录")
|
||
else:
|
||
shutil.rmtree(project_dir)
|
||
_ok(f"已删除目录: {project_dir}")
|
||
cfg = load_install_config()
|
||
if cfg.get("dir") == str(project_dir):
|
||
from nav_common import config_path
|
||
|
||
try:
|
||
config_path().unlink(missing_ok=True)
|
||
except OSError:
|
||
pass
|
||
else:
|
||
_warn(f"已保留安装目录: {project_dir}")
|
||
|
||
_ok("卸载完成")
|
||
|
||
|
||
def render_menu(project_dir: Path | None) -> None:
|
||
print()
|
||
print("=" * 50)
|
||
print(" LocalNav 部署管理")
|
||
print("=" * 50)
|
||
if project_dir:
|
||
print(f" 安装目录: {project_dir}")
|
||
else:
|
||
print(f" 安装目录: 未安装（默认 {default_install_dir()}）")
|
||
print(f" 服务状态: {service_status_text(project_dir)}")
|
||
if pm2_app_status():
|
||
print(f" PM2 进程: {APP_NAME}")
|
||
print("-" * 50)
|
||
print(" 1 一键部署安装")
|
||
print(" 2 一键卸载")
|
||
print(" 3 更新")
|
||
print(" 0 退出")
|
||
print("-" * 50)
|
||
|
||
|
||
def interactive_menu() -> None:
|
||
_require_tty("打开菜单")
|
||
print(f"LocalNav 部署管理 · {platform.system()} {platform.release()}")
|
||
while True:
|
||
project_dir = resolve_project_dir()
|
||
render_menu(project_dir)
|
||
try:
|
||
choice = input("请选择: ").strip()
|
||
except (EOFError, KeyboardInterrupt):
|
||
print("\n再见。")
|
||
return
|
||
|
||
if not choice:
|
||
_warn("请输入 0-3")
|
||
_pause()
|
||
continue
|
||
|
||
if choice == "1":
|
||
try:
|
||
action_install()
|
||
except subprocess.CalledProcessError as exc:
|
||
_fail(f"安装失败: {exc}")
|
||
_pause()
|
||
elif choice == "2":
|
||
try:
|
||
action_uninstall()
|
||
except subprocess.CalledProcessError as exc:
|
||
_fail(f"卸载失败: {exc}")
|
||
_pause()
|
||
elif choice == "3":
|
||
try:
|
||
action_update()
|
||
except subprocess.CalledProcessError as exc:
|
||
_fail(f"更新失败: {exc}")
|
||
_pause()
|
||
elif choice == "0":
|
||
print("再见。")
|
||
return
|
||
else:
|
||
_warn("无效选项，请输入 0-3")
|
||
_pause()
|
||
|
||
|
||
def parse_args() -> argparse.Namespace:
|
||
parser = argparse.ArgumentParser(description="LocalNav 部署管理")
|
||
parser.add_argument(
|
||
"command",
|
||
nargs="?",
|
||
choices=("install", "uninstall", "update", "menu"),
|
||
help="直接执行指定操作；省略则进入交互菜单",
|
||
)
|
||
parser.add_argument("--repo", default="", help="Git 仓库地址")
|
||
parser.add_argument("--dir", default="", help="安装目录")
|
||
parser.add_argument("-y", "--yes", action="store_true", help="安装时使用默认路径，减少提问")
|
||
return parser.parse_args()
|
||
|
||
|
||
def main() -> None:
|
||
args = parse_args()
|
||
if args.command == "install":
|
||
action_install(repo=args.repo, dest=args.dir, yes=args.yes)
|
||
elif args.command == "uninstall":
|
||
action_uninstall(dest=args.dir)
|
||
elif args.command == "update":
|
||
action_update(dest=args.dir)
|
||
elif args.command == "menu":
|
||
interactive_menu()
|
||
else:
|
||
interactive_menu()
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|