264 lines
8.7 KiB
Python
264 lines
8.7 KiB
Python
#!/usr/bin/env python3
|
||
# -*- coding: utf-8 -*-
|
||
"""
|
||
wzdj-options 管理端部署脚本（new/wz-admin）
|
||
|
||
目标目录（默认）: /www/wwwroot/self/wzdj-options/admin
|
||
流程: 本地 npm run build -> 打包 dist -> 宝塔 API 上传/解压
|
||
"""
|
||
|
||
from __future__ import print_function
|
||
|
||
import argparse
|
||
import hashlib
|
||
import json
|
||
import os
|
||
import shutil
|
||
import subprocess
|
||
import sys
|
||
import tarfile
|
||
import tempfile
|
||
import time
|
||
|
||
try:
|
||
import requests
|
||
try:
|
||
import urllib3
|
||
urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)
|
||
except Exception:
|
||
pass
|
||
except ImportError:
|
||
requests = None
|
||
|
||
|
||
DEPLOY_PROJECT_PATH = "/www/wwwroot/self/wzdj-options/admin"
|
||
# 与 new/wz-api/master.py 等处一致；可被环境变量 BT_API_KEY 覆盖
|
||
BT_API_KEY_DEFAULT = "qcWubCdlfFjS2b2DMT1lzPFaDfmv1cBT"
|
||
|
||
|
||
def get_cfg():
|
||
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 {
|
||
"project_path": os.environ.get("DEPLOY_PROJECT_PATH", DEPLOY_PROJECT_PATH),
|
||
"bt_panel_url": bt_url,
|
||
"bt_api_key": (os.environ.get("BT_API_KEY") or BT_API_KEY_DEFAULT).strip(),
|
||
}
|
||
|
||
|
||
def run_frontend_build(root):
|
||
print("[1/3] 本地构建管理端 ...")
|
||
if sys.platform == "win32":
|
||
npm_exe = shutil.which("npm.cmd") or shutil.which("npm")
|
||
if npm_exe:
|
||
cmd = [npm_exe, "run", "build"]
|
||
use_shell = False
|
||
else:
|
||
cmd = "npm run build"
|
||
use_shell = True
|
||
else:
|
||
cmd = [shutil.which("npm") or "npm", "run", "build"]
|
||
use_shell = False
|
||
try:
|
||
r = subprocess.run(cmd, cwd=root, shell=use_shell, timeout=900, capture_output=True, text=True, encoding="utf-8", errors="replace")
|
||
if r.returncode != 0:
|
||
print(" [失败] 构建失败，退出码:", r.returncode)
|
||
if r.stderr:
|
||
for line in (r.stderr or "").strip().split("\n")[-20:]:
|
||
print(" " + line)
|
||
return None
|
||
dist_dir = os.path.join(root, "dist")
|
||
if not os.path.isdir(dist_dir):
|
||
print(" [失败] 未找到 dist 目录")
|
||
return None
|
||
print(" [成功] 构建完成: %s" % dist_dir)
|
||
return dist_dir
|
||
except Exception as e:
|
||
print(" [失败] 构建异常:", str(e))
|
||
return None
|
||
|
||
|
||
def pack_dist(dist_dir):
|
||
print("[2/3] 打包 dist 文件 ...")
|
||
tarball = os.path.join(tempfile.gettempdir(), "wzdj_admin_dist.tar.gz")
|
||
try:
|
||
with tarfile.open(tarball, "w:gz") as tf:
|
||
for name in os.listdir(dist_dir):
|
||
full = os.path.join(dist_dir, name)
|
||
tf.add(full, arcname=name)
|
||
print(" [成功] 打包完成: %s (%.2f MB)" % (tarball, os.path.getsize(tarball) / 1024 / 1024))
|
||
return tarball
|
||
except Exception as e:
|
||
print(" [失败] 打包异常:", str(e))
|
||
return None
|
||
|
||
|
||
def _bt_signed_post(base_url, key, path, extra_data, timeout=30):
|
||
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):
|
||
url = (cfg.get("bt_panel_url") or "").rstrip("/")
|
||
key = cfg.get("bt_api_key") or ""
|
||
if not url or not key:
|
||
print("[失败] 缺少 BT_API_KEY 或 BT_PANEL_URL")
|
||
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()
|
||
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:
|
||
return False
|
||
print("[宝塔API 探活] 成功（GetDiskInfo）")
|
||
return True
|
||
except Exception as e:
|
||
print("[宝塔API 探活] 异常:", e)
|
||
return False
|
||
|
||
|
||
def bt_upload_file_resumable(cfg, local_path, remote_dir, remote_name):
|
||
url = (cfg.get("bt_panel_url") or "").rstrip("/")
|
||
key = cfg.get("bt_api_key") or ""
|
||
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
|
||
upload_url = url + "/files?action=upload"
|
||
headers = {"User-Agent": "Mozilla/5.0"}
|
||
with open(local_path, "rb") as fp:
|
||
while f_start < total:
|
||
fp.seek(f_start)
|
||
buf = fp.read(min(chunk_size, total - f_start))
|
||
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)
|
||
raw = (r.text or "").strip()
|
||
if raw.isdigit():
|
||
nxt = int(raw)
|
||
if nxt <= f_start:
|
||
return False
|
||
f_start = nxt
|
||
continue
|
||
j = _bt_parse_json_response(r)
|
||
if isinstance(j, dict) and j.get("status") is True:
|
||
return True
|
||
return False
|
||
return False
|
||
|
||
|
||
def bt_unzip_remote(cfg, sfile, dfile):
|
||
url = (cfg.get("bt_panel_url") or "").rstrip("/")
|
||
key = cfg.get("bt_api_key") or ""
|
||
for type1 in ("tar.gz", "tgz", "tar"):
|
||
j = _bt_parse_json_response(_bt_signed_post(url, key, "/files?action=UnZip", {"sfile": sfile, "dfile": dfile, "type1": type1, "coding": "UTF-8"}, timeout=180))
|
||
if isinstance(j, dict) and j.get("status") is True:
|
||
return True
|
||
return False
|
||
|
||
|
||
def bt_delete_file(cfg, remote_path):
|
||
url = (cfg.get("bt_panel_url") or "").rstrip("/")
|
||
key = cfg.get("bt_api_key") or ""
|
||
j = _bt_parse_json_response(_bt_signed_post(url, key, "/files?action=DeleteFile", {"path": remote_path}))
|
||
return isinstance(j, dict) and j.get("status") is True
|
||
|
||
|
||
def upload_and_extract(cfg, tarball_path):
|
||
print("[3/3] 上传并解压到宝塔目录 ...")
|
||
project_path = cfg["project_path"].rstrip("/")
|
||
tar_name = "wzdj_admin_dist.tar.gz"
|
||
remote_tar = project_path + "/" + tar_name
|
||
if not bt_upload_file_resumable(cfg, tarball_path, project_path, tar_name):
|
||
print(" [失败] 上传失败")
|
||
return False
|
||
if not bt_unzip_remote(cfg, remote_tar, project_path + "/"):
|
||
print(" [失败] 解压失败")
|
||
return False
|
||
bt_delete_file(cfg, remote_tar)
|
||
return True
|
||
|
||
|
||
def main():
|
||
parser = argparse.ArgumentParser(description="wzdj-options 管理端部署脚本（宝塔 API）")
|
||
parser.add_argument("--no-build", action="store_true", help="跳过本地构建（直接打包已有 dist）")
|
||
parser.add_argument("--skip-bt-ping", action="store_true", help="跳过宝塔探活")
|
||
args = parser.parse_args()
|
||
|
||
root = os.path.dirname(os.path.abspath(__file__))
|
||
cfg = get_cfg()
|
||
print("=" * 60)
|
||
print("wzdj-options 管理端部署")
|
||
print("目标目录:", cfg["project_path"])
|
||
print("=" * 60)
|
||
|
||
if not requests:
|
||
print("[失败] 需要 requests：pip install requests")
|
||
return 1
|
||
if not args.skip_bt_ping and not bt_panel_ping(cfg):
|
||
return 1
|
||
|
||
dist_dir = os.path.join(root, "dist")
|
||
if not args.no_build:
|
||
d = run_frontend_build(root)
|
||
if not d:
|
||
return 1
|
||
dist_dir = d
|
||
elif not os.path.isdir(dist_dir):
|
||
print("[失败] 未找到 dist 目录")
|
||
return 1
|
||
|
||
tarball = pack_dist(dist_dir)
|
||
if not tarball:
|
||
return 1
|
||
ok = upload_and_extract(cfg, tarball)
|
||
try:
|
||
os.remove(tarball)
|
||
except Exception:
|
||
pass
|
||
if not ok:
|
||
return 1
|
||
print("")
|
||
print("部署完成:", cfg["project_path"])
|
||
return 0
|
||
|
||
|
||
if __name__ == "__main__":
|
||
sys.exit(main())
|
||
|