591 lines
20 KiB
Python
591 lines
20 KiB
Python
#!/usr/bin/env python3
|
||
# -*- coding: utf-8 -*-
|
||
"""
|
||
soul-admin 静态站点部署（统一主入口，全程宝塔面板 API，无需 SSH）
|
||
|
||
流程与 soul-api/master.py 对齐：
|
||
探活(GetDiskInfo) → 本地 pnpm 构建 dist → 打包 zip → /files?action=upload
|
||
→ /files?action=UnZip 到 dist2 → MvFile 切换 dist/dist2（等价原 dist→dist1→删）→ DeleteFile 删包
|
||
|
||
环境变量（与 soul-api 共用宝塔相关项）：
|
||
BT_PANEL_URL、BT_API_KEY、BT_UPLOAD_CHUNK_MB（可选，默认 4）
|
||
DEPLOY_HOST（未设 BT_PANEL_URL 时默认 https://{DEPLOY_HOST}:9988）
|
||
DEPLOY_BASE_PATH / DEPLOY_BASE_PATH_PROD / DEPLOY_BASE_PATH_DEV
|
||
|
||
deploy.py 作为 dev 包装入口：default_profile=dev。
|
||
"""
|
||
|
||
from __future__ import print_function
|
||
|
||
import argparse
|
||
import hashlib
|
||
import json
|
||
import os
|
||
import subprocess
|
||
import sys
|
||
import tempfile
|
||
import time
|
||
import zipfile
|
||
|
||
try:
|
||
import requests
|
||
|
||
try:
|
||
import urllib3
|
||
|
||
urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)
|
||
except Exception:
|
||
pass
|
||
except ImportError:
|
||
requests = None
|
||
|
||
# 与 soul-api/master.py 一致，便于同一套面板密钥
|
||
BT_API_KEY_DEFAULT = "qcWubCdlfFjS2b2DMT1lzPFaDfmv1cBT"
|
||
|
||
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)
|
||
host = os.environ.get("DEPLOY_HOST", "43.139.27.93")
|
||
bt_url = (os.environ.get("BT_PANEL_URL") or "").strip().rstrip("/")
|
||
if not bt_url:
|
||
bt_url = "https://%s:9988" % host
|
||
return {
|
||
"profile": profile,
|
||
"title": PROFILE_PRESETS[profile]["title"],
|
||
"build_cmd": list(PROFILE_PRESETS[profile]["build_cmd"]),
|
||
"build_desc": PROFILE_PRESETS[profile]["build_desc"],
|
||
"host": host,
|
||
"base_path": base,
|
||
"dist_path": base + "/dist",
|
||
"dist2_path": base + "/dist2",
|
||
"dist1_path": base + "/dist1",
|
||
"bt_panel_url": bt_url,
|
||
"bt_api_key": os.environ.get("BT_API_KEY", BT_API_KEY_DEFAULT),
|
||
}
|
||
|
||
|
||
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 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/5] 本地构建 %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/5] 打包 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
|
||
|
||
|
||
# ---------- 宝塔 API（与 soul-api/master.py 同源精简） ----------
|
||
|
||
|
||
def _bt_signed_post(base_url, key, path, extra_data, timeout=20):
|
||
req_time = int(time.time())
|
||
sk_md5 = hashlib.md5(key.encode()).hexdigest()
|
||
req_token = hashlib.md5(("%s%s" % (req_time, sk_md5)).encode()).hexdigest()
|
||
data = {"request_time": req_time, "request_token": req_token}
|
||
data.update(extra_data or {})
|
||
return requests.post(base_url + path, data=data, timeout=timeout, verify=False)
|
||
|
||
|
||
def _bt_parse_json_response(r):
|
||
if r is None or r.status_code != 200:
|
||
return None
|
||
ct = (r.headers.get("content-type") or "").lower()
|
||
if "json" in ct:
|
||
try:
|
||
return r.json()
|
||
except Exception:
|
||
pass
|
||
t = (r.text or "").lstrip()
|
||
if t.startswith("{"):
|
||
try:
|
||
return json.loads(r.text)
|
||
except Exception:
|
||
pass
|
||
return None
|
||
|
||
|
||
def bt_panel_ping(cfg):
|
||
if not requests:
|
||
return True
|
||
url = (cfg.get("bt_panel_url") or "").rstrip("/")
|
||
key = cfg.get("bt_api_key") or ""
|
||
if not url or not key:
|
||
print("[宝塔API 探活] 缺少 BT_PANEL_URL 或 BT_API_KEY，已跳过")
|
||
return True
|
||
req_time = int(time.time())
|
||
sk_md5 = hashlib.md5(key.encode()).hexdigest()
|
||
req_token = hashlib.md5(("%s%s" % (req_time, sk_md5)).encode()).hexdigest()
|
||
try:
|
||
r = requests.post(
|
||
url + "/system?action=GetDiskInfo",
|
||
data={"request_time": req_time, "request_token": req_token},
|
||
timeout=20,
|
||
verify=False,
|
||
)
|
||
if r.status_code != 200:
|
||
print("[宝塔API 探活] HTTP %s" % r.status_code)
|
||
return False
|
||
ct = (r.headers.get("content-type") or "").lower()
|
||
j = {}
|
||
if "json" in ct:
|
||
try:
|
||
j = r.json()
|
||
except Exception:
|
||
j = {}
|
||
else:
|
||
t = (r.text or "").lstrip()
|
||
if t.startswith("{"):
|
||
try:
|
||
j = json.loads(r.text)
|
||
except Exception:
|
||
j = {}
|
||
if isinstance(j, dict) and j.get("status") is False:
|
||
print("[宝塔API 探活] %s" % (j.get("msg") or j))
|
||
return False
|
||
print("[宝塔API 探活] 成功（GetDiskInfo）", flush=True)
|
||
return True
|
||
except Exception as e:
|
||
print("[宝塔API 探活] 异常: %s" % e)
|
||
return False
|
||
|
||
|
||
def _bt_upload_headers():
|
||
return {
|
||
"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 "
|
||
"(KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36"
|
||
}
|
||
|
||
|
||
def _bt_parse_upload_response(r):
|
||
if r is None or r.status_code != 200:
|
||
return "err", None, "HTTP %s" % (r.status_code if r else "?")
|
||
raw = (r.text or "").strip()
|
||
if raw.isdigit():
|
||
return "progress", int(raw), None
|
||
j = _bt_parse_json_response(r)
|
||
if isinstance(j, dict):
|
||
if j.get("status") is True:
|
||
return "ok", None, None
|
||
return "err", None, j.get("msg") or str(j)
|
||
return "err", None, raw[:300] if raw else "empty body"
|
||
|
||
|
||
def bt_upload_file_resumable(cfg, local_path, remote_dir, remote_name):
|
||
if not requests:
|
||
print(" [失败] 需要 requests：pip install requests")
|
||
return False
|
||
url = (cfg.get("bt_panel_url") or "").rstrip("/")
|
||
key = cfg.get("bt_api_key") or ""
|
||
if not url or not key:
|
||
print(" [失败] 缺少 BT_PANEL_URL 或 BT_API_KEY")
|
||
return False
|
||
total = os.path.getsize(local_path)
|
||
chunk_mb = int(os.environ.get("BT_UPLOAD_CHUNK_MB", "4"))
|
||
chunk_size = max(1, chunk_mb) * 1024 * 1024
|
||
f_start = 0
|
||
last_start = -1
|
||
upload_url = url + "/files?action=upload"
|
||
headers = _bt_upload_headers()
|
||
print(
|
||
" [宝塔API] 上传 %s → %s/%s（共 %.2f MB，每块 %d MB）…"
|
||
% (
|
||
os.path.basename(local_path),
|
||
remote_dir,
|
||
remote_name,
|
||
total / 1024 / 1024,
|
||
chunk_mb,
|
||
)
|
||
)
|
||
try:
|
||
with open(local_path, "rb") as fp:
|
||
while f_start < total:
|
||
if f_start == last_start:
|
||
print(" [失败] 上传停滞 offset=%s（请删服务器上同名 .upload.tmp 后重试）" % f_start)
|
||
return False
|
||
last_start = f_start
|
||
fp.seek(f_start)
|
||
buf = fp.read(min(chunk_size, total - f_start))
|
||
if not buf:
|
||
print(" [失败] 读取本地文件失败 offset=%s" % f_start)
|
||
return False
|
||
req_time = int(time.time())
|
||
sk_md5 = hashlib.md5(key.encode()).hexdigest()
|
||
req_token = hashlib.md5(("%s%s" % (req_time, sk_md5)).encode()).hexdigest()
|
||
data = {
|
||
"request_time": str(req_time),
|
||
"request_token": req_token,
|
||
"f_path": remote_dir,
|
||
"f_name": remote_name,
|
||
"f_size": str(total),
|
||
"f_start": str(f_start),
|
||
}
|
||
files = {"blob": ("blob", buf, "application/octet-stream")}
|
||
r = requests.post(
|
||
upload_url,
|
||
data=data,
|
||
files=files,
|
||
headers=headers,
|
||
timeout=600,
|
||
verify=False,
|
||
)
|
||
kind, prog, err = _bt_parse_upload_response(r)
|
||
if kind == "ok":
|
||
print(" [成功] 宝塔上传完成")
|
||
return True
|
||
if kind == "progress":
|
||
if prog <= f_start:
|
||
print(" [失败] 上传偏移未前进: server=%s local=%s" % (prog, f_start))
|
||
return False
|
||
f_start = prog
|
||
continue
|
||
print(" [失败] 上传接口: %s" % (err or kind))
|
||
return False
|
||
print(" [失败] 上传未正常结束（offset=%s size=%s）" % (f_start, total))
|
||
return False
|
||
except Exception as e:
|
||
print(" [失败] 上传异常: %s" % e)
|
||
return False
|
||
|
||
|
||
def bt_delete_file(cfg, remote_path):
|
||
url = (cfg.get("bt_panel_url") or "").rstrip("/")
|
||
key = cfg.get("bt_api_key") or ""
|
||
if not url or not key or not remote_path:
|
||
return False
|
||
j = _bt_parse_json_response(
|
||
_bt_signed_post(url, key, "/files?action=DeleteFile", {"path": remote_path})
|
||
)
|
||
if isinstance(j, dict) and j.get("status") is True:
|
||
return True
|
||
if isinstance(j, dict) and j.get("msg"):
|
||
print(" [宝塔API] DeleteFile: %s" % j.get("msg"))
|
||
return False
|
||
|
||
|
||
def bt_delete_dir(cfg, remote_dir):
|
||
"""递归删除目录（不存在时可能返回失败，调用方可忽略）。"""
|
||
url = (cfg.get("bt_panel_url") or "").rstrip("/")
|
||
key = cfg.get("bt_api_key") or ""
|
||
if not url or not key or not remote_dir:
|
||
return False
|
||
j = _bt_parse_json_response(
|
||
_bt_signed_post(
|
||
url,
|
||
key,
|
||
"/files?action=DeleteDir",
|
||
{"path": remote_dir},
|
||
timeout=120,
|
||
)
|
||
)
|
||
if isinstance(j, dict) and j.get("status") is True:
|
||
return True
|
||
return False
|
||
|
||
|
||
def bt_create_dir(cfg, path):
|
||
url = (cfg.get("bt_panel_url") or "").rstrip("/")
|
||
key = cfg.get("bt_api_key") or ""
|
||
if not url or not key or not path:
|
||
return False
|
||
j = _bt_parse_json_response(
|
||
_bt_signed_post(url, key, "/files?action=CreateDir", {"path": path})
|
||
)
|
||
if isinstance(j, dict) and j.get("status") is True:
|
||
return True
|
||
msg = (j.get("msg") if isinstance(j, dict) else "") or ""
|
||
if msg and ("已存在" in msg or "exist" in msg.lower()):
|
||
return True
|
||
if isinstance(j, dict) and j.get("msg"):
|
||
print(" [宝塔API] CreateDir: %s" % j.get("msg"))
|
||
return False
|
||
|
||
|
||
def bt_mv_file(cfg, sfile, dfile):
|
||
url = (cfg.get("bt_panel_url") or "").rstrip("/")
|
||
key = cfg.get("bt_api_key") or ""
|
||
if not url or not key:
|
||
return False, "no url/key"
|
||
j = _bt_parse_json_response(
|
||
_bt_signed_post(
|
||
url,
|
||
key,
|
||
"/files?action=MvFile",
|
||
{"sfile": sfile, "dfile": dfile},
|
||
timeout=120,
|
||
)
|
||
)
|
||
if isinstance(j, dict) and j.get("status") is True:
|
||
return True, None
|
||
msg = j.get("msg") if isinstance(j, dict) else str(j)
|
||
return False, msg
|
||
|
||
|
||
def bt_unzip_remote_zip(cfg, sfile, dfile):
|
||
url = (cfg.get("bt_panel_url") or "").rstrip("/")
|
||
key = cfg.get("bt_api_key") or ""
|
||
coding = "UTF-8"
|
||
for type1 in ("zip", "ZIP"):
|
||
j = _bt_parse_json_response(
|
||
_bt_signed_post(
|
||
url,
|
||
key,
|
||
"/files?action=UnZip",
|
||
{"sfile": sfile, "dfile": dfile, "type1": type1, "coding": coding},
|
||
timeout=300,
|
||
)
|
||
)
|
||
if isinstance(j, dict) and j.get("status") is True:
|
||
print(" [成功] 宝塔解压完成 (type1=%s)" % type1)
|
||
return True
|
||
if isinstance(j, dict) and j.get("msg"):
|
||
print(" [宝塔API] UnZip type1=%s: %s" % (type1, j.get("msg")))
|
||
return False
|
||
|
||
|
||
def deploy_via_btapi(cfg, zip_path):
|
||
print("[3/5] 宝塔 API：准备 dist2、上传、解压 …")
|
||
if not requests:
|
||
print(" [失败] 需要 requests：pip install requests")
|
||
return False
|
||
|
||
base = cfg["base_path"].rstrip("/")
|
||
dist = cfg["dist_path"].rstrip("/")
|
||
dist2 = cfg["dist2_path"].rstrip("/")
|
||
dist1 = cfg["dist1_path"].rstrip("/")
|
||
remote_zip_name = "soul_admin_deploy.zip"
|
||
remote_zip = base + "/" + remote_zip_name
|
||
dfile_dist2 = dist2 + "/"
|
||
|
||
bt_delete_dir(cfg, dist1)
|
||
bt_delete_dir(cfg, dist2)
|
||
if not bt_create_dir(cfg, dist2):
|
||
print(" [失败] 无法创建 dist2: %s" % dist2)
|
||
return False
|
||
|
||
if not bt_upload_file_resumable(cfg, zip_path, base, remote_zip_name):
|
||
return False
|
||
|
||
print(" [宝塔API] 解压 %s → %s …" % (remote_zip, dfile_dist2))
|
||
if not bt_unzip_remote_zip(cfg, remote_zip, dfile_dist2):
|
||
print(" [提示] 解压失败时可检查面板「文件」权限与压缩包格式")
|
||
return False
|
||
|
||
if bt_delete_file(cfg, remote_zip):
|
||
print(" [已清理] 远程 %s" % remote_zip)
|
||
else:
|
||
print(" [提示] 未能删除远程压缩包，可在面板手动删: %s" % remote_zip)
|
||
|
||
print("[4/5] 宝塔 API：目录切换（等价原 dist→dist1→dist2→dist）…")
|
||
bt_delete_dir(cfg, dist1)
|
||
|
||
ok_move_old, err_old = bt_mv_file(cfg, dist, dist1)
|
||
if not ok_move_old:
|
||
if err_old and isinstance(err_old, str):
|
||
low = err_old.lower()
|
||
if "不存在" not in err_old and "not exist" not in low and "不存在" not in low:
|
||
print(" [宝塔API] 移动 dist→dist1: %s" % err_old)
|
||
|
||
ok_swap, err_swap = bt_mv_file(cfg, dist2, dist)
|
||
if not ok_swap:
|
||
print(" [失败] 将 dist2 切换为 dist 失败: %s" % err_swap)
|
||
print(" [提示] 可到面板「文件」手动把 %s 改名为 dist" % dist2)
|
||
return False
|
||
|
||
if ok_move_old:
|
||
if not bt_delete_dir(cfg, dist1):
|
||
print(" [警告] 未能删除临时目录 dist1: %s（可面板手动删）" % dist1)
|
||
|
||
print(" [成功] 新版本已生效: %s" % dist)
|
||
print("[5/5] 完成（纯 API；若 403 请在面板核对站点目录属主 www）")
|
||
return True
|
||
|
||
|
||
def create_parser(default_profile):
|
||
parser = argparse.ArgumentParser(
|
||
description="soul-admin 静态站点部署（宝塔 API：上传 zip、解压、目录切换）",
|
||
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")
|
||
parser.add_argument(
|
||
"--skip-bt-ping",
|
||
action="store_true",
|
||
help="跳过宝塔面板 API 探活（GetDiskInfo）",
|
||
)
|
||
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 部署（宝塔 API）" % cfg["title"])
|
||
print("=" * 60)
|
||
print(" profile: %s" % cfg["profile"])
|
||
print(" 面板: %s" % cfg["bt_panel_url"])
|
||
print(" 站点根: %s" % cfg["base_path"])
|
||
print(" 静态目录: %s" % cfg["dist_path"])
|
||
print("=" * 60)
|
||
|
||
if not args.skip_bt_ping:
|
||
if not bt_panel_ping(cfg):
|
||
print("[失败] 宝塔 API 探活未通过（可用 --skip-bt-ping 跳过）")
|
||
return 1
|
||
|
||
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/5] 跳过本地构建")
|
||
|
||
zip_path = pack_dist_zip(root, profile=profile)
|
||
if not zip_path:
|
||
return 1
|
||
|
||
try:
|
||
if not deploy_via_btapi(cfg, zip_path):
|
||
return 1
|
||
finally:
|
||
try:
|
||
if zip_path and os.path.isfile(zip_path):
|
||
os.remove(zip_path)
|
||
except Exception:
|
||
pass
|
||
|
||
print("")
|
||
print(" 部署完成！站点目录: %s" % cfg["dist_path"])
|
||
return 0
|
||
|
||
|
||
if __name__ == "__main__":
|
||
sys.exit(main())
|