xsk-education-management/scripts/deploy_to_vps.py

#!/usr/bin/env python3
from __future__ import annotations

import argparse
import os
from pathlib import Path
import shlex
import subprocess
import sys
import tempfile


PROJECT_DIR = Path(__file__).resolve().parents[1]
DEFAULT_REMOTE_HOST = "121.199.172.246"
DEFAULT_REMOTE_USER = "root"
DEFAULT_REMOTE_PORT = 22222
DEFAULT_REMOTE_DIR = "/root/新时空教务管理系统"
DEFAULT_IDENTITY_FILE = Path.home() / ".ssh" / "xsk_records_vps_ed25519"


def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(description="部署新时空教务管理系统到 VPS")
    parser.add_argument("--remote-host", default=os.getenv("XSK_REMOTE_HOST", DEFAULT_REMOTE_HOST))
    parser.add_argument("--remote-user", default=os.getenv("XSK_REMOTE_USER", DEFAULT_REMOTE_USER))
    parser.add_argument("--remote-port", type=int, default=int(os.getenv("XSK_REMOTE_PORT", str(DEFAULT_REMOTE_PORT))))
    parser.add_argument("--remote-dir", default=os.getenv("XSK_REMOTE_DIR", DEFAULT_REMOTE_DIR))
    parser.add_argument(
        "--identity-file",
        default=os.getenv("XSK_IDENTITY_FILE", str(DEFAULT_IDENTITY_FILE) if DEFAULT_IDENTITY_FILE.exists() else ""),
        help="SSH 私钥路径；默认兼容使用历史文件 ~/.ssh/xsk_records_vps_ed25519（如果存在）",
    )
    parser.add_argument("--app-port", default=os.getenv("XSK_APP_PORT", "18080"))
    parser.add_argument(
        "--python-image",
        default=os.getenv(
            "XSK_PYTHON_IMAGE",
            "swr.cn-north-4.myhuaweicloud.com/ddn-k8s/docker.io/library/python:3.12-slim",
        ),
        help="Docker 基础镜像；VPS 无法访问 Docker Hub 时可使用镜像站",
    )
    parser.add_argument("--auth-user", default=os.getenv("XSK_WEB_USER", "wolfydw"))
    parser.add_argument("--auth-password", default=os.getenv("XSK_WEB_PASSWORD"))
    parser.add_argument("--use-sshpass", action="store_true", default=os.getenv("XSK_USE_SSHPASS", "") == "1")
    return parser.parse_args()


def run(command: list[str], cwd: Path | None = None) -> None:
    env = os.environ.copy()
    if env.get("XSK_SSH_PASSWORD") and not env.get("SSHPASS"):
        env["SSHPASS"] = env["XSK_SSH_PASSWORD"]
    result = subprocess.run(command, cwd=cwd, text=True, capture_output=True, check=False, env=env)
    if result.returncode != 0:
        redacted = ["***" if part == os.getenv("XSK_SSH_PASSWORD") else part for part in command]
        raise RuntimeError(
            f"命令失败({result.returncode}): {' '.join(shlex.quote(part) for part in redacted)}\n"
            f"STDOUT: {result.stdout.strip()}\nSTDERR: {result.stderr.strip()}"
        )
    if result.stdout.strip():
        print(result.stdout.strip())


def base_remote(args: argparse.Namespace) -> str:
    return f"{args.remote_user}@{args.remote_host}"


def ssh_options(args: argparse.Namespace) -> list[str]:
    options = ["-p", str(args.remote_port), "-o", "StrictHostKeyChecking=accept-new"]
    if args.identity_file:
        options.extend(["-i", str(Path(args.identity_file).expanduser()), "-o", "IdentitiesOnly=yes"])
    return options


def ssh_command(args: argparse.Namespace) -> list[str]:
    command: list[str] = []
    if args.use_sshpass:
        password = os.getenv("XSK_SSH_PASSWORD")
        if not password:
            raise RuntimeError("启用 --use-sshpass 时必须设置 XSK_SSH_PASSWORD")
        command.extend(["sshpass", "-e"])
    command.extend(["ssh", *ssh_options(args), base_remote(args)])
    return command


def rsync_command(args: argparse.Namespace) -> list[str]:
    command: list[str] = []
    if args.use_sshpass:
        password = os.getenv("XSK_SSH_PASSWORD")
        if not password:
            raise RuntimeError("启用 --use-sshpass 时必须设置 XSK_SSH_PASSWORD")
        command.extend(["sshpass", "-e"])
    ssh_command_text = " ".join(shlex.quote(part) for part in ["ssh", *ssh_options(args)])
    command.extend(
        [
            "rsync",
            "-az",
            "--delete",
            "--exclude",
            ".env",
            "--exclude",
            "*.log",
            "-e",
            ssh_command_text,
        ]
    )
    return command


def write_env(args: argparse.Namespace) -> Path:
    if not args.auth_password:
        raise RuntimeError("必须通过 XSK_WEB_PASSWORD 或 --auth-password 设置网页访问密码")
    env_text = "\n".join(
        [
            f"APP_PORT={args.app_port}",
            "TZ=Asia/Shanghai",
            f"PYTHON_IMAGE={args.python_image}",
            f"BASIC_AUTH_USERNAME={args.auth_user}",
            f"BASIC_AUTH_PASSWORD={args.auth_password}",
            "CLASSNOTES_PATH=/data/classnotes.txt",
            "ACCOUNTS_PATH=/data/学生课时账户.md",
            "",
        ]
    )
    handle = tempfile.NamedTemporaryFile("w", encoding="utf-8", delete=False)
    try:
        handle.write(env_text)
        return Path(handle.name)
    finally:
        handle.close()


def main() -> int:
    args = parse_args()
    app_dir = f"{args.remote_dir.rstrip('/')}/app"
    data_dir = f"{args.remote_dir.rstrip('/')}/data"
    env_file = write_env(args)
    try:
        run(ssh_command(args) + [f"mkdir -p {shlex.quote(app_dir)} {shlex.quote(data_dir)}"])
        run(rsync_command(args) + [f"{PROJECT_DIR}/", f"{base_remote(args)}:{app_dir}/"])
        run(rsync_command(args) + [str(env_file), f"{base_remote(args)}:{app_dir}/.env"])
        run(ssh_command(args) + [f"cd {shlex.quote(app_dir)} && docker compose up -d --build --remove-orphans"])
        print(f"部署完成: http://{args.remote_host}:{args.app_port}/")
    finally:
        env_file.unlink(missing_ok=True)
    return 0


if __name__ == "__main__":
    raise SystemExit(main())