dekun
f508f210f9
Provide navctl.py with a numbered menu, persist install path in config, and support curl bootstrap without pre-cloning the repo. Co-authored-by: Cursor <cursoragent@cursor.com>
351 lines
10 KiB
Python
351 lines
10 KiB
Python
#!/usr/bin/env python3
|
||
"""LocalNav 项目内部署：环境检测、生成 .env、安装依赖（需已在项目目录内）。"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import argparse
|
||
import json
|
||
import os
|
||
import platform
|
||
import secrets
|
||
import shutil
|
||
import subprocess
|
||
import sys
|
||
import time
|
||
import urllib.error
|
||
import urllib.request
|
||
from pathlib import Path
|
||
|
||
ROOT = Path(__file__).resolve().parent.parent
|
||
ENV_EXAMPLE = ROOT / ".env.example"
|
||
ENV_FILE = ROOT / ".env"
|
||
VENV_DIR = ROOT / ".venv"
|
||
REQUIREMENTS = ROOT / "requirements.txt"
|
||
|
||
MIN_PYTHON = (3, 10)
|
||
|
||
DEFAULTS = {
|
||
"NAV_ADMIN_USERNAME": "admin",
|
||
"NAV_ADMIN_PASSWORD": "admin123",
|
||
"NAV_HUB_AUTO_LOGIN": "1",
|
||
"NAV_COOKIES_INSECURE_HTTP": "1",
|
||
}
|
||
|
||
|
||
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 check_python() -> str:
|
||
ver = sys.version_info
|
||
if ver < MIN_PYTHON:
|
||
_fail(f"需要 Python {MIN_PYTHON[0]}.{MIN_PYTHON[1]}+，当前 {ver.major}.{ver.minor}")
|
||
_ok(f"Python {ver.major}.{ver.minor}.{ver.micro}")
|
||
return sys.executable
|
||
|
||
|
||
def check_pip(py_exe: str) -> None:
|
||
try:
|
||
subprocess.run(
|
||
[py_exe, "-m", "pip", "--version"],
|
||
check=True,
|
||
capture_output=True,
|
||
text=True,
|
||
)
|
||
_ok("pip 可用")
|
||
except (subprocess.CalledProcessError, FileNotFoundError):
|
||
_fail("pip 不可用，请先安装 pip")
|
||
|
||
|
||
def optional_tools() -> None:
|
||
if shutil.which("git"):
|
||
_ok("可选工具 git 已安装")
|
||
else:
|
||
_warn("未检测到 git（可选，首次安装请用 scripts/install.py）")
|
||
if shutil.which("pm2"):
|
||
_ok("可选工具 pm2 已安装")
|
||
else:
|
||
_warn("未检测到 pm2（可选）")
|
||
|
||
|
||
def read_env_lines(path: Path) -> list[str]:
|
||
if not path.is_file():
|
||
return []
|
||
return path.read_text(encoding="utf-8-sig").splitlines()
|
||
|
||
|
||
def parse_env_keys(lines: list[str]) -> dict[str, str]:
|
||
out: dict[str, str] = {}
|
||
for line in lines:
|
||
s = line.strip()
|
||
if not s or s.startswith("#") or "=" not in s:
|
||
continue
|
||
key, _, val = s.partition("=")
|
||
out[key.strip()] = val.strip()
|
||
return out
|
||
|
||
|
||
def write_env(lines: list[str], updates: dict[str, str]) -> None:
|
||
existing = parse_env_keys(lines)
|
||
merged = {**existing, **updates}
|
||
out_lines: list[str] = []
|
||
written: set[str] = set()
|
||
|
||
for line in lines:
|
||
s = line.strip()
|
||
if not s or s.startswith("#") or "=" not in s:
|
||
out_lines.append(line)
|
||
continue
|
||
key, _, _ = s.partition("=")
|
||
key = key.strip()
|
||
if key in merged and key not in written:
|
||
out_lines.append(f"{key}={merged[key]}")
|
||
written.add(key)
|
||
else:
|
||
out_lines.append(line)
|
||
|
||
for key, val in merged.items():
|
||
if key not in written:
|
||
out_lines.append(f"{key}={val}")
|
||
ENV_FILE.write_text("\n".join(out_lines) + "\n", encoding="utf-8")
|
||
|
||
|
||
def ensure_env() -> None:
|
||
if not ENV_FILE.is_file():
|
||
if ENV_EXAMPLE.is_file():
|
||
shutil.copy2(ENV_EXAMPLE, ENV_FILE)
|
||
_ok("已从 .env.example 创建 .env")
|
||
else:
|
||
ENV_FILE.write_text("", encoding="utf-8")
|
||
_ok("已创建空 .env")
|
||
|
||
lines = read_env_lines(ENV_FILE)
|
||
parsed = parse_env_keys(lines)
|
||
updates: dict[str, str] = {}
|
||
|
||
secret = parsed.get("NAV_SECRET_KEY", "").strip()
|
||
if not secret:
|
||
updates["NAV_SECRET_KEY"] = secrets.token_hex(32)
|
||
_ok("已自动生成 NAV_SECRET_KEY")
|
||
|
||
for key, val in DEFAULTS.items():
|
||
if not parsed.get(key, "").strip():
|
||
updates[key] = val
|
||
|
||
if updates:
|
||
write_env(lines, updates)
|
||
_ok(f"已写入 .env 配置项：{', '.join(updates.keys())}")
|
||
else:
|
||
_ok(".env 已完整，未覆盖现有配置")
|
||
|
||
|
||
def venv_python() -> str:
|
||
if platform.system() == "Windows":
|
||
py = VENV_DIR / "Scripts" / "python.exe"
|
||
else:
|
||
py = VENV_DIR / "bin" / "python"
|
||
return str(py)
|
||
|
||
|
||
def ensure_venv(base_py: str) -> str:
|
||
if not VENV_DIR.is_dir():
|
||
subprocess.run([base_py, "-m", "venv", str(VENV_DIR)], check=True)
|
||
_ok(f"已创建虚拟环境 {VENV_DIR}")
|
||
else:
|
||
_ok("虚拟环境已存在")
|
||
return venv_python()
|
||
|
||
|
||
def install_requirements(py_exe: str) -> None:
|
||
cmd = [py_exe, "-m", "pip", "install", "-r", str(REQUIREMENTS)]
|
||
try:
|
||
subprocess.run(cmd, check=True, cwd=ROOT)
|
||
except subprocess.CalledProcessError:
|
||
_warn("默认 pip 源失败，尝试 https://pypi.org/simple")
|
||
subprocess.run([*cmd, "-i", "https://pypi.org/simple"], check=True, cwd=ROOT)
|
||
_ok("依赖安装完成")
|
||
|
||
|
||
def smoke_test(py_exe: str) -> None:
|
||
subprocess.run(
|
||
[
|
||
py_exe,
|
||
"-c",
|
||
"from app import create_app; create_app(); print('app import ok')",
|
||
],
|
||
check=True,
|
||
cwd=ROOT,
|
||
)
|
||
_ok("应用启动检查通过")
|
||
|
||
|
||
def ensure_logs_dir() -> None:
|
||
logs = ROOT / "logs"
|
||
logs.mkdir(parents=True, exist_ok=True)
|
||
|
||
|
||
def pm2_has_app(name: str = "nav-site") -> bool:
|
||
pm2 = shutil.which("pm2")
|
||
if not pm2:
|
||
return False
|
||
try:
|
||
proc = subprocess.run(
|
||
[pm2, "jlist"],
|
||
check=True,
|
||
capture_output=True,
|
||
text=True,
|
||
)
|
||
apps = json.loads(proc.stdout or "[]")
|
||
return any(app.get("name") == name for app in apps)
|
||
except (subprocess.CalledProcessError, json.JSONDecodeError, OSError):
|
||
return False
|
||
|
||
|
||
def systemd_unit_ready(name: str = "nav-site") -> bool:
|
||
systemctl = shutil.which("systemctl")
|
||
if not systemctl:
|
||
return False
|
||
try:
|
||
proc = subprocess.run(
|
||
[systemctl, "list-unit-files", f"{name}.service", "--no-pager", "--no-legend"],
|
||
check=True,
|
||
capture_output=True,
|
||
text=True,
|
||
)
|
||
return bool((proc.stdout or "").strip())
|
||
except (subprocess.CalledProcessError, OSError):
|
||
return False
|
||
|
||
|
||
def wait_for_http(port: str, timeout: float = 15.0) -> bool:
|
||
url = f"http://127.0.0.1:{port}/login"
|
||
deadline = time.time() + timeout
|
||
while time.time() < deadline:
|
||
try:
|
||
with urllib.request.urlopen(url, timeout=2) as resp:
|
||
if resp.status == 200:
|
||
return True
|
||
except (urllib.error.URLError, TimeoutError, OSError):
|
||
time.sleep(0.8)
|
||
return False
|
||
|
||
|
||
def start_service(port: str) -> str | None:
|
||
"""尝试自动启动服务，返回启动方式描述；失败返回 None。"""
|
||
ensure_logs_dir()
|
||
pm2 = shutil.which("pm2")
|
||
ecosystem = ROOT / "ecosystem.config.cjs"
|
||
|
||
if pm2 and ecosystem.is_file():
|
||
if pm2_has_app("nav-site"):
|
||
subprocess.run(
|
||
[pm2, "restart", "nav-site", "--update-env"],
|
||
check=True,
|
||
cwd=ROOT,
|
||
)
|
||
mode = "pm2 restart nav-site"
|
||
else:
|
||
subprocess.run(
|
||
[pm2, "start", str(ecosystem)],
|
||
check=True,
|
||
cwd=ROOT,
|
||
)
|
||
mode = "pm2 start ecosystem.config.cjs"
|
||
subprocess.run([pm2, "save"], cwd=ROOT, check=False)
|
||
if wait_for_http(port):
|
||
return mode
|
||
_warn("PM2 已启动进程，但健康检查未通过，请执行 pm2 logs nav-site 排查")
|
||
return mode
|
||
|
||
systemctl = shutil.which("systemctl")
|
||
if systemctl and systemd_unit_ready("nav-site"):
|
||
subprocess.run([systemctl, "enable", "--now", "nav-site"], check=True)
|
||
if wait_for_http(port):
|
||
return "systemctl enable --now nav-site"
|
||
_warn("systemd 已启动 nav-site，但健康检查未通过，请执行 journalctl -u nav-site 排查")
|
||
return "systemctl enable --now nav-site"
|
||
|
||
return None
|
||
|
||
|
||
def print_summary(started_via: str | None = None) -> None:
|
||
port = os.environ.get("NAV_PORT", "5070")
|
||
print()
|
||
print("=" * 50)
|
||
print("部署完成")
|
||
print(f" 访问地址: http://0.0.0.0:{port}")
|
||
print(" 默认账号: admin / admin123")
|
||
print(" 生产环境请尽快在「系统设置」中修改密码")
|
||
print()
|
||
if started_via:
|
||
print(f" 服务状态: 已自动启动（{started_via}）")
|
||
if shutil.which("pm2"):
|
||
print(" 查看状态: pm2 status")
|
||
print(" 查看日志: pm2 logs nav-site")
|
||
print(" 开机自启: pm2 save && pm2 startup # 按提示执行一次 sudo 命令")
|
||
else:
|
||
print(" 服务状态: 未自动启动（未检测到 pm2 / systemd nav-site）")
|
||
print(" 手动启动：")
|
||
print(f" {venv_python()} app.py")
|
||
if shutil.which("pm2"):
|
||
print(" pm2 start ecosystem.config.cjs")
|
||
print("=" * 50)
|
||
|
||
|
||
def parse_args() -> argparse.Namespace:
|
||
parser = argparse.ArgumentParser(description="LocalNav 项目内部署")
|
||
parser.add_argument(
|
||
"--no-start",
|
||
action="store_true",
|
||
help="仅安装依赖与配置，不自动启动服务",
|
||
)
|
||
return parser.parse_args()
|
||
|
||
|
||
def main() -> None:
|
||
args = parse_args()
|
||
if not (ROOT / "app.py").is_file():
|
||
_fail(
|
||
"请在 LocalNav 项目目录内运行，或使用部署管理脚本：\n"
|
||
" python scripts/navctl.py"
|
||
)
|
||
|
||
print(f"LocalNav 项目部署 · {platform.system()} {platform.release()}")
|
||
print(f"项目目录: {ROOT}")
|
||
print()
|
||
|
||
base_py = check_python()
|
||
check_pip(base_py)
|
||
optional_tools()
|
||
ensure_env()
|
||
py_exe = ensure_venv(base_py)
|
||
install_requirements(py_exe)
|
||
smoke_test(py_exe)
|
||
|
||
started_via = None
|
||
if not args.no_start:
|
||
port = os.environ.get("NAV_PORT", "5070")
|
||
try:
|
||
started_via = start_service(port)
|
||
if started_via:
|
||
_ok(f"服务已自动启动（{started_via}）")
|
||
else:
|
||
_warn("未能自动启动，请按下方说明手动启动")
|
||
except subprocess.CalledProcessError as exc:
|
||
_warn(f"自动启动失败: {exc}")
|
||
|
||
print_summary(started_via)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|