0ce86645b1
- Added a new webhook integration for Feishu/Lark groups, allowing for custom bot notifications. - Refactored deployment scripts to unify the deployment process, improving maintainability and reducing redundancy. - Updated the experience index to include recent entries and improved documentation for various roles. Made-with: Cursor
357 lines
12 KiB
Python
357 lines
12 KiB
Python
#!/usr/bin/env python3
|
||
# -*- coding: utf-8 -*-
|
||
"""
|
||
soul-admin 静态站点部署（统一主入口）
|
||
|
||
流程：本地构建 dist -> 上传 zip -> 服务器解压到 dist2 -> dist/dist2 无缝切换。
|
||
默认 profile=prod（正式环境）；deploy.py 作为 dev 包装入口调用本脚本。
|
||
"""
|
||
|
||
from __future__ import print_function
|
||
|
||
import argparse
|
||
import os
|
||
import shlex
|
||
import subprocess
|
||
import sys
|
||
import tempfile
|
||
import zipfile
|
||
|
||
try:
|
||
import paramiko
|
||
except ImportError:
|
||
print("错误: 请先安装 paramiko")
|
||
print(" pip install paramiko")
|
||
sys.exit(1)
|
||
|
||
|
||
DEFAULT_SSH_PORT = int(os.environ.get("DEPLOY_SSH_PORT", "22022"))
|
||
DEFAULT_WWW_USER = "www:www"
|
||
|
||
PROFILE_PRESETS = {
|
||
"prod": {
|
||
"title": "soul-admin",
|
||
"default_base_path": "/www/wwwroot/self/soul-admin",
|
||
"build_cmd": ["pnpm", "build"],
|
||
"build_desc": "pnpm build（正式环境）",
|
||
},
|
||
"dev": {
|
||
"title": "soul-admin-dev",
|
||
"default_base_path": "/www/wwwroot/self/soul-admin-dev",
|
||
"build_cmd": ["pnpm", "run", "build:dev"],
|
||
"build_desc": "pnpm run build:dev（测试环境）",
|
||
},
|
||
}
|
||
|
||
|
||
def resolve_profile(profile_name):
|
||
profile = (profile_name or "prod").strip().lower()
|
||
if profile not in PROFILE_PRESETS:
|
||
raise ValueError("不支持的 profile: %s" % profile_name)
|
||
return profile
|
||
|
||
|
||
def _resolve_base_path(profile, cli_base_path=None):
|
||
preset = PROFILE_PRESETS[profile]
|
||
if cli_base_path:
|
||
return cli_base_path.rstrip("/")
|
||
|
||
profile_env_key = "DEPLOY_BASE_PATH_%s" % profile.upper()
|
||
base = (os.environ.get(profile_env_key) or "").strip()
|
||
if not base:
|
||
base = (os.environ.get("DEPLOY_BASE_PATH") or "").strip()
|
||
if not base:
|
||
base = preset["default_base_path"]
|
||
return base.rstrip("/")
|
||
|
||
|
||
def get_cfg(profile, cli_base_path=None):
|
||
base = _resolve_base_path(profile, cli_base_path=cli_base_path)
|
||
www_user = (os.environ.get("DEPLOY_WWW_USER") or DEFAULT_WWW_USER).strip()
|
||
return {
|
||
"profile": profile,
|
||
"title": PROFILE_PRESETS[profile]["title"],
|
||
"build_cmd": list(PROFILE_PRESETS[profile]["build_cmd"]),
|
||
"build_desc": PROFILE_PRESETS[profile]["build_desc"],
|
||
"host": os.environ.get("DEPLOY_HOST", "43.139.27.93"),
|
||
"user": os.environ.get("DEPLOY_USER", "root"),
|
||
"password": os.environ.get("DEPLOY_PASSWORD", "Zhiqun1984"),
|
||
"ssh_key": os.environ.get("DEPLOY_SSH_KEY", ""),
|
||
"base_path": base,
|
||
"dist_path": base + "/dist",
|
||
"dist2_path": base + "/dist2",
|
||
"www_user": www_user,
|
||
}
|
||
|
||
|
||
def resolve_project_root():
|
||
script_dir = os.path.dirname(os.path.abspath(__file__))
|
||
if os.path.isfile(os.path.join(script_dir, "package.json")):
|
||
return script_dir
|
||
return os.path.dirname(script_dir)
|
||
|
||
|
||
def connect_ssh(cfg, timeout=30):
|
||
client = paramiko.SSHClient()
|
||
client.set_missing_host_key_policy(paramiko.AutoAddPolicy())
|
||
if cfg.get("ssh_key") and os.path.isfile(cfg["ssh_key"]):
|
||
client.connect(
|
||
cfg["host"],
|
||
port=DEFAULT_SSH_PORT,
|
||
username=cfg["user"],
|
||
key_filename=cfg["ssh_key"],
|
||
timeout=timeout,
|
||
banner_timeout=timeout,
|
||
)
|
||
else:
|
||
client.connect(
|
||
cfg["host"],
|
||
port=DEFAULT_SSH_PORT,
|
||
username=cfg["user"],
|
||
password=cfg["password"],
|
||
timeout=timeout,
|
||
banner_timeout=timeout,
|
||
)
|
||
return client
|
||
|
||
|
||
def ensure_dist_ready(root):
|
||
return os.path.isdir(os.path.join(root, "dist")) and os.path.isfile(
|
||
os.path.join(root, "dist", "index.html")
|
||
)
|
||
|
||
|
||
def run_build(root, cfg):
|
||
print("[1/4] 本地构建 %s ..." % cfg["build_desc"])
|
||
use_shell = sys.platform == "win32"
|
||
try:
|
||
r = subprocess.run(
|
||
cfg["build_cmd"],
|
||
cwd=root,
|
||
shell=use_shell,
|
||
timeout=300,
|
||
capture_output=True,
|
||
text=True,
|
||
encoding="utf-8",
|
||
errors="replace",
|
||
)
|
||
if r.returncode != 0:
|
||
print(" [失败] 构建失败，退出码:", r.returncode)
|
||
for line in (r.stdout or "").strip().split("\n")[-10:]:
|
||
if line:
|
||
print(" " + line)
|
||
for line in (r.stderr or "").strip().split("\n")[-10:]:
|
||
if line:
|
||
print(" " + line)
|
||
return False
|
||
except subprocess.TimeoutExpired:
|
||
print(" [失败] 构建超时")
|
||
return False
|
||
except FileNotFoundError:
|
||
print(" [失败] 未找到 pnpm，请安装: npm install -g pnpm")
|
||
return False
|
||
except Exception as e:
|
||
print(" [失败] 构建异常:", str(e))
|
||
return False
|
||
|
||
if not ensure_dist_ready(root):
|
||
print(" [失败] 未找到 dist/index.html")
|
||
return False
|
||
print(" [成功] 构建完成")
|
||
return True
|
||
|
||
|
||
def pack_dist_zip(root, profile):
|
||
print("[2/4] 打包 dist 为 zip ...")
|
||
dist_dir = os.path.join(root, "dist")
|
||
if not os.path.isdir(dist_dir):
|
||
print(" [失败] 未找到 dist 目录")
|
||
return None
|
||
|
||
index_html = os.path.join(dist_dir, "index.html")
|
||
if not os.path.isfile(index_html):
|
||
print(" [失败] 未找到 dist/index.html，请先执行构建")
|
||
return None
|
||
|
||
zip_name = "soul_admin_%s_deploy.zip" % profile
|
||
zip_path = os.path.join(tempfile.gettempdir(), zip_name)
|
||
try:
|
||
with zipfile.ZipFile(zip_path, "w", zipfile.ZIP_DEFLATED) as zf:
|
||
for dirpath, _dirs, filenames in os.walk(dist_dir):
|
||
for name in filenames:
|
||
full = os.path.join(dirpath, name)
|
||
arcname = os.path.relpath(full, dist_dir).replace("\\", "/")
|
||
zf.write(full, arcname)
|
||
print(
|
||
" [成功] 打包完成: %s (%.2f MB)"
|
||
% (zip_path, os.path.getsize(zip_path) / 1024 / 1024)
|
||
)
|
||
return zip_path
|
||
except Exception as e:
|
||
print(" [失败] 打包异常:", str(e))
|
||
return None
|
||
|
||
|
||
def upload_zip_and_extract_to_dist2(cfg, zip_path):
|
||
print("[3/4] SSH 上传 zip 并解压到 dist2 ...")
|
||
if not cfg.get("password") and not cfg.get("ssh_key"):
|
||
print(" [失败] 请设置 DEPLOY_PASSWORD 或 DEPLOY_SSH_KEY")
|
||
return False
|
||
|
||
zip_size_mb = os.path.getsize(zip_path) / (1024 * 1024)
|
||
remote_zip = cfg["base_path"] + "/soul_admin_deploy.zip"
|
||
client = None
|
||
try:
|
||
print(" 正在连接 %s@%s:%s ..." % (cfg["user"], cfg["host"], DEFAULT_SSH_PORT))
|
||
client = connect_ssh(cfg, timeout=30)
|
||
print(" [OK] SSH 已连接，正在上传 zip（%.1f MB）..." % zip_size_mb)
|
||
sftp = client.open_sftp()
|
||
sftp.put(zip_path, remote_zip)
|
||
sftp.close()
|
||
|
||
dist2 = cfg["dist2_path"]
|
||
cmd = (
|
||
"rm -rf {dist2} && mkdir -p {dist2} && "
|
||
"unzip -o -q {remote_zip} -d {dist2} && rm -f {remote_zip} && echo OK"
|
||
).format(
|
||
dist2=shlex.quote(dist2),
|
||
remote_zip=shlex.quote(remote_zip),
|
||
)
|
||
stdin, stdout, stderr = client.exec_command(cmd, timeout=120)
|
||
out = stdout.read().decode("utf-8", errors="replace").strip()
|
||
err = stderr.read().decode("utf-8", errors="replace").strip()
|
||
exit_status = stdout.channel.recv_exit_status()
|
||
if err:
|
||
print(" 服务器 stderr: %s" % err[:500])
|
||
if exit_status != 0 or "OK" not in out:
|
||
print(" [失败] 解压失败，退出码: %s" % exit_status)
|
||
if out:
|
||
print(" stdout: %s" % out[:300])
|
||
return False
|
||
print(" [成功] 已解压到: %s" % dist2)
|
||
return True
|
||
except Exception as e:
|
||
print(" [失败] SSH 错误: %s" % str(e))
|
||
return False
|
||
finally:
|
||
if client:
|
||
client.close()
|
||
|
||
|
||
def remote_swap_dist(cfg):
|
||
print("[4/4] 服务器切换目录: dist→dist1, dist2→dist ...")
|
||
client = None
|
||
try:
|
||
client = connect_ssh(cfg, timeout=15)
|
||
base = shlex.quote(cfg["base_path"])
|
||
cmd = (
|
||
"cd {base} && "
|
||
"(test -d dist && (mv dist dist1 && mv dist2 dist && rm -rf dist1) || mv dist2 dist) "
|
||
"&& echo OK"
|
||
).format(base=base)
|
||
stdin, stdout, stderr = client.exec_command(cmd, timeout=60)
|
||
out = stdout.read().decode("utf-8", errors="replace").strip()
|
||
err = stderr.read().decode("utf-8", errors="replace").strip()
|
||
exit_status = stdout.channel.recv_exit_status()
|
||
if exit_status != 0 or "OK" not in out:
|
||
print(" [失败] 切换失败 (退出码: %s)" % exit_status)
|
||
if err:
|
||
print(" 服务器 stderr: %s" % err[:300])
|
||
if out and "OK" not in out:
|
||
print(" 服务器 stdout: %s" % out[:300])
|
||
return False
|
||
|
||
print(" [成功] 新版本已切换至: %s" % cfg["dist_path"])
|
||
www_user = cfg.get("www_user")
|
||
if www_user:
|
||
chown_cmd = "chown -R {user} {dist} && echo OK".format(
|
||
user=shlex.quote(www_user),
|
||
dist=shlex.quote(cfg["dist_path"]),
|
||
)
|
||
stdin, stdout, stderr = client.exec_command(chown_cmd, timeout=60)
|
||
chown_out = stdout.read().decode("utf-8", errors="replace").strip()
|
||
chown_err = stderr.read().decode("utf-8", errors="replace").strip()
|
||
if stdout.channel.recv_exit_status() != 0 or "OK" not in chown_out:
|
||
print(" [警告] chown 失败，站点可能无法访问: %s" % (chown_err or chown_out))
|
||
else:
|
||
print(" [成功] 已设置属主: %s" % www_user)
|
||
return True
|
||
except Exception as e:
|
||
print(" [失败] SSH 错误: %s" % str(e))
|
||
return False
|
||
finally:
|
||
if client:
|
||
client.close()
|
||
|
||
|
||
def create_parser(default_profile):
|
||
parser = argparse.ArgumentParser(
|
||
description="soul-admin 静态站点部署（dist2 解压后目录切换，无缝更新）",
|
||
formatter_class=argparse.RawDescriptionHelpFormatter,
|
||
)
|
||
parser.add_argument(
|
||
"--profile",
|
||
choices=tuple(PROFILE_PRESETS.keys()),
|
||
default=default_profile,
|
||
help="部署档位：prod=正式环境，dev=测试环境",
|
||
)
|
||
parser.add_argument(
|
||
"--base-path",
|
||
default="",
|
||
help="覆盖部署目录（默认按 profile 取 DEPLOY_BASE_PATH 或预设路径）",
|
||
)
|
||
parser.add_argument("--no-build", action="store_true", help="跳过本地构建，直接上传现有 dist")
|
||
return parser
|
||
|
||
|
||
def main(argv=None, default_profile="prod"):
|
||
default_profile = resolve_profile(default_profile)
|
||
parser = create_parser(default_profile=default_profile)
|
||
args = parser.parse_args(argv)
|
||
|
||
profile = resolve_profile(args.profile)
|
||
root = resolve_project_root()
|
||
cfg = get_cfg(profile=profile, cli_base_path=(args.base_path or "").strip())
|
||
|
||
print("=" * 60)
|
||
print(" %s 部署（dist/dist2 无缝切换）" % cfg["title"])
|
||
print("=" * 60)
|
||
print(" profile: %s" % cfg["profile"])
|
||
print(" 服务器: %s@%s:%s" % (cfg["user"], cfg["host"], DEFAULT_SSH_PORT))
|
||
print(" 站点目录: %s" % cfg["dist_path"])
|
||
print("=" * 60)
|
||
|
||
if not args.no_build:
|
||
if not run_build(root, cfg):
|
||
return 1
|
||
else:
|
||
if not ensure_dist_ready(root):
|
||
print("[错误] 未找到 dist/index.html，请先执行构建或去掉 --no-build")
|
||
return 1
|
||
print("[1/4] 跳过本地构建")
|
||
|
||
zip_path = pack_dist_zip(root, profile=profile)
|
||
if not zip_path:
|
||
return 1
|
||
|
||
try:
|
||
if not upload_zip_and_extract_to_dist2(cfg, zip_path):
|
||
return 1
|
||
finally:
|
||
try:
|
||
if zip_path and os.path.isfile(zip_path):
|
||
os.remove(zip_path)
|
||
except Exception:
|
||
pass
|
||
|
||
if not remote_swap_dist(cfg):
|
||
return 1
|
||
|
||
print("")
|
||
print(" 部署完成！站点目录: %s" % cfg["dist_path"])
|
||
return 0
|
||
|
||
|
||
if __name__ == "__main__":
|
||
sys.exit(main())
|