dekun
d75bf5b0de
Fall back to a pid-file background process after deploy, and detect running state via HTTP port checks in the management menu. Co-authored-by: Cursor <cursoragent@cursor.com>
288 lines
8.6 KiB
Python
288 lines
8.6 KiB
Python
"""LocalNav 部署脚本公共工具。"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import json
|
||
import os
|
||
import platform
|
||
import shutil
|
||
import socket
|
||
import subprocess
|
||
import urllib.error
|
||
import urllib.request
|
||
from pathlib import Path
|
||
|
||
DEFAULT_REPO = "https://git.bz121.com/dekun/LocalNav.git"
|
||
MIN_PYTHON = (3, 10)
|
||
APP_NAME = "nav-site"
|
||
|
||
|
||
def config_path() -> Path:
|
||
if platform.system() == "Windows":
|
||
base = Path(os.environ.get("USERPROFILE", Path.home())) / ".localnav"
|
||
else:
|
||
base = Path(os.environ.get("XDG_CONFIG_HOME", Path.home() / ".config")) / "localnav"
|
||
base.mkdir(parents=True, exist_ok=True)
|
||
return base / "install.json"
|
||
|
||
|
||
def load_install_config() -> dict:
|
||
path = config_path()
|
||
if not path.is_file():
|
||
return {}
|
||
try:
|
||
return json.loads(path.read_text(encoding="utf-8"))
|
||
except (json.JSONDecodeError, OSError):
|
||
return {}
|
||
|
||
|
||
def save_install_config(project_dir: Path, repo: str) -> None:
|
||
data = {
|
||
"dir": str(project_dir.resolve()),
|
||
"repo": repo.strip(),
|
||
}
|
||
config_path().write_text(json.dumps(data, indent=2) + "\n", encoding="utf-8")
|
||
|
||
|
||
def default_install_dir() -> Path:
|
||
cfg = load_install_config()
|
||
if cfg.get("dir"):
|
||
return Path(cfg["dir"]).expanduser()
|
||
if os.environ.get("NAV_INSTALL_DIR"):
|
||
return Path(os.environ["NAV_INSTALL_DIR"]).expanduser()
|
||
if platform.system() == "Windows":
|
||
return Path.cwd() / "LocalNav"
|
||
return Path("/opt/LocalNav")
|
||
|
||
|
||
def default_repo() -> str:
|
||
cfg = load_install_config()
|
||
return (cfg.get("repo") or os.environ.get("NAV_INSTALL_REPO") or DEFAULT_REPO).strip()
|
||
|
||
|
||
def is_localnav_project(path: Path) -> bool:
|
||
return (path / "app.py").is_file() and (path / "scripts" / "deploy.py").is_file()
|
||
|
||
|
||
def resolve_project_dir(explicit: str | None = None) -> Path | None:
|
||
if explicit and explicit.strip():
|
||
p = Path(explicit.strip()).expanduser().resolve()
|
||
return p if is_localnav_project(p) else None
|
||
cfg_dir = load_install_config().get("dir", "").strip()
|
||
if cfg_dir and is_localnav_project(Path(cfg_dir)):
|
||
return Path(cfg_dir).resolve()
|
||
here = Path.cwd()
|
||
if is_localnav_project(here):
|
||
return here.resolve()
|
||
return None
|
||
|
||
|
||
def pid_file_path(project_dir: Path) -> Path:
|
||
return project_dir / "logs" / "nav-site.pid"
|
||
|
||
|
||
def get_nav_port(project_dir: Path | None = None) -> str:
|
||
if project_dir:
|
||
env_file = project_dir / ".env"
|
||
if env_file.is_file():
|
||
for line in env_file.read_text(encoding="utf-8-sig").splitlines():
|
||
s = line.strip()
|
||
if s.startswith("NAV_PORT="):
|
||
val = s.split("=", 1)[1].strip().strip("\"'")
|
||
if val:
|
||
return val
|
||
return os.environ.get("NAV_PORT", "5070")
|
||
|
||
|
||
def http_ready(port: str, timeout: float = 2.0) -> bool:
|
||
url = f"http://127.0.0.1:{port}/login"
|
||
try:
|
||
with urllib.request.urlopen(url, timeout=timeout) as resp:
|
||
return resp.status == 200
|
||
except (urllib.error.URLError, TimeoutError, OSError):
|
||
return False
|
||
|
||
|
||
def port_open(port: str) -> bool:
|
||
try:
|
||
p = int(port)
|
||
except ValueError:
|
||
return False
|
||
try:
|
||
with socket.create_connection(("127.0.0.1", p), timeout=1.5):
|
||
return True
|
||
except OSError:
|
||
return False
|
||
|
||
|
||
def read_background_pid(project_dir: Path) -> int | None:
|
||
path = pid_file_path(project_dir)
|
||
if not path.is_file():
|
||
return None
|
||
try:
|
||
return int(path.read_text(encoding="utf-8").strip())
|
||
except (ValueError, OSError):
|
||
return None
|
||
|
||
|
||
def is_process_running(pid: int) -> bool:
|
||
if pid <= 0:
|
||
return False
|
||
if platform.system() == "Windows":
|
||
try:
|
||
proc = subprocess.run(
|
||
["tasklist", "/FI", f"PID eq {pid}"],
|
||
capture_output=True,
|
||
text=True,
|
||
check=False,
|
||
)
|
||
return str(pid) in (proc.stdout or "")
|
||
except OSError:
|
||
return False
|
||
try:
|
||
os.kill(pid, 0)
|
||
return True
|
||
except OSError:
|
||
return False
|
||
|
||
|
||
def stop_background_process(project_dir: Path) -> bool:
|
||
pid = read_background_pid(project_dir)
|
||
if not pid or not is_process_running(pid):
|
||
pid_file_path(project_dir).unlink(missing_ok=True)
|
||
return False
|
||
try:
|
||
if platform.system() == "Windows":
|
||
subprocess.run(["taskkill", "/PID", str(pid), "/F"], check=False)
|
||
else:
|
||
os.kill(pid, 15)
|
||
except OSError:
|
||
pass
|
||
pid_file_path(project_dir).unlink(missing_ok=True)
|
||
return True
|
||
|
||
|
||
def pm2_available() -> str | None:
|
||
return shutil.which("pm2")
|
||
|
||
|
||
def pm2_has_app(name: str = APP_NAME) -> bool:
|
||
pm2 = pm2_available()
|
||
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 pm2_app_status(name: str = APP_NAME) -> str | None:
|
||
pm2 = pm2_available()
|
||
if not pm2:
|
||
return None
|
||
try:
|
||
proc = subprocess.run(
|
||
[pm2, "jlist"],
|
||
check=True,
|
||
capture_output=True,
|
||
text=True,
|
||
)
|
||
for app in json.loads(proc.stdout or "[]"):
|
||
if app.get("name") == name:
|
||
return str(app.get("pm2_env", {}).get("status") or "unknown")
|
||
except (subprocess.CalledProcessError, json.JSONDecodeError, OSError):
|
||
return None
|
||
return None
|
||
|
||
|
||
def systemd_unit_ready(name: str = APP_NAME) -> 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 systemd_active(name: str = APP_NAME) -> bool:
|
||
systemctl = shutil.which("systemctl")
|
||
if not systemctl:
|
||
return False
|
||
try:
|
||
proc = subprocess.run(
|
||
[systemctl, "is-active", f"{name}.service"],
|
||
check=True,
|
||
capture_output=True,
|
||
text=True,
|
||
)
|
||
return (proc.stdout or "").strip() == "active"
|
||
except (subprocess.CalledProcessError, OSError):
|
||
return False
|
||
|
||
|
||
def service_status_text(project_dir: Path | None) -> str:
|
||
if not project_dir:
|
||
return "未安装"
|
||
|
||
port = get_nav_port(project_dir)
|
||
if http_ready(port):
|
||
pm2_state = pm2_app_status()
|
||
if pm2_state:
|
||
return f"运行中 · pm2 {pm2_state} · :{port}"
|
||
if systemd_unit_ready() and systemd_active():
|
||
return f"运行中 · systemd · :{port}"
|
||
pid = read_background_pid(project_dir)
|
||
if pid and is_process_running(pid):
|
||
return f"运行中 · 后台 pid {pid} · :{port}"
|
||
return f"运行中 · 端口 {port}"
|
||
|
||
pm2_state = pm2_app_status()
|
||
if pm2_state:
|
||
return f"pm2 {pm2_state}（未响应 :{port}）"
|
||
if systemd_unit_ready() and systemd_active():
|
||
return f"systemd active（未响应 :{port}）"
|
||
if systemd_unit_ready():
|
||
return "systemd 已配置但未运行"
|
||
pid = read_background_pid(project_dir)
|
||
if pid and is_process_running(pid):
|
||
return f"后台 pid {pid}（未响应 :{port}）"
|
||
if port_open(port):
|
||
return f"端口 {port} 已占用（进程未知）"
|
||
return "未运行"
|
||
|
||
|
||
def stop_service(project_dir: Path | None = None) -> list[str]:
|
||
"""停止并移除进程管理中的 nav-site，返回已执行操作说明。"""
|
||
actions: list[str] = []
|
||
pm2 = pm2_available()
|
||
if pm2 and pm2_has_app():
|
||
subprocess.run([pm2, "stop", APP_NAME], check=False)
|
||
subprocess.run([pm2, "delete", APP_NAME], check=False)
|
||
subprocess.run([pm2, "save"], check=False)
|
||
actions.append("pm2 stop/delete nav-site")
|
||
|
||
systemctl = shutil.which("systemctl")
|
||
if systemctl and systemd_unit_ready():
|
||
subprocess.run([systemctl, "stop", f"{APP_NAME}.service"], check=False)
|
||
subprocess.run([systemctl, "disable", f"{APP_NAME}.service"], check=False)
|
||
actions.append("systemctl stop/disable nav-site")
|
||
|
||
if project_dir and stop_background_process(project_dir):
|
||
actions.append("已停止后台 nav-site 进程")
|
||
|
||
return actions
|