fnvtk/soul-yongping
2
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

删除多个完成报告文件，优化项目结构以提升可维护性。

Browse Source
This commit is contained in:
乘风
2026-02-03 15:59:37 +08:00
parent d4ca9573f5
commit a2443c097c
119 changed files with 2119 additions and 8537 deletions
Download Patch File Download Diff File Expand all files Collapse all files

724
scripts/TestDevlop.py Normal file
View File

@@ -0,0 +1,724 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
Soul 创业派对 - 统一部署脚本（支持两种模式）
模式一 devlop（默认）：dist 切换方式
本地 pnpm build → 打包 zip → 上传解压到 dist2 → 宝塔暂停 → dist→dist1, dist2→dist → 重启
用法: python scripts/devlop.py [--no-build]
模式二 deploy：直接覆盖方式
本地打包 tar.gz → SSH 上传解压到项目目录 → 宝塔 API 重启
用法: python scripts/devlop.py --mode deploy [--no-build] [--no-upload] [--no-api]
环境变量（可选）:
DEPLOY_HOST / DEPLOY_USER / DEPLOY_PASSWORD / DEPLOY_SSH_KEY
DEPLOY_PROJECT_PATH # deploy 模式项目路径，默认 /www/wwwroot/soulTest
DEVOP_BASE_PATH # devlop 模式目录，默认 /www/wwwroot/auto-devlop/soulTest
BAOTA_PANEL_URL / BAOTA_API_KEY / DEPLOY_PM2_APP
DEPLOY_PORT / DEPLOY_NODE_VERSION / DEPLOY_NODE_PATH
"""
from __future__ import print_function
import os
import sys
import shutil
import tempfile
import argparse
import json
import zipfile
import tarfile
import subprocess
import time
import hashlib
try:
import paramiko
except ImportError:
print("错误: 请先安装 paramiko")
print(" pip install paramiko")
sys.exit(1)
try:
import requests
import urllib3
urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)
except ImportError:
print("错误: 请先安装 requests")
print(" pip install requests")
sys.exit(1)
# ==================== 配置 ====================
def get_cfg():
"""获取基础部署配置（deploy 模式与 devlop 共用 SSH/宝塔）"""
return {
"host": os.environ.get("DEPLOY_HOST", "42.194.232.22"),
"user": os.environ.get("DEPLOY_USER", "root"),
"password": os.environ.get("DEPLOY_PASSWORD", "Zhiqun1984"),
"ssh_key": os.environ.get("DEPLOY_SSH_KEY", ""),
"project_path": os.environ.get("DEPLOY_PROJECT_PATH", "/www/wwwroot/soulTest"),
"panel_url": os.environ.get("BAOTA_PANEL_URL", "https://42.194.232.22:9988"),
"api_key": os.environ.get("BAOTA_API_KEY", "hsAWqFSi0GOCrunhmYdkxy92tBXfqYjd"),
"pm2_name": os.environ.get("DEPLOY_PM2_APP", "soulTest"),
"site_url": os.environ.get("DEPLOY_SITE_URL", "https://soulTest.quwanzhi.com"),
"port": int(os.environ.get("DEPLOY_PORT", "30066")),
"node_version": os.environ.get("DEPLOY_NODE_VERSION", "v22.14.0"),
"node_path": os.environ.get("DEPLOY_NODE_PATH", "/www/server/nodejs/v22.14.0/bin"),
}
def get_cfg_devlop():
"""devlop 模式配置：在基础配置上增加 base_path / dist / dist2"""
cfg = get_cfg().copy()
cfg["base_path"] = os.environ.get("DEVOP_BASE_PATH", "/www/wwwroot/auto-devlop/soulTest")
cfg["dist_path"] = cfg["base_path"] + "/dist"
cfg["dist2_path"] = cfg["base_path"] + "/dist2"
return cfg
# ==================== 宝塔 API ====================
def _get_sign(api_key):
now_time = int(time.time())
sign_str = str(now_time) + hashlib.md5(api_key.encode("utf-8")).hexdigest()
request_token = hashlib.md5(sign_str.encode("utf-8")).hexdigest()
return now_time, request_token
def _baota_request(panel_url, api_key, path, data=None):
req_time, req_token = _get_sign(api_key)
payload = {"request_time": req_time, "request_token": req_token}
if data:
payload.update(data)
url = panel_url.rstrip("/") + "/" + path.lstrip("/")
try:
r = requests.post(url, data=payload, verify=False, timeout=30)
return r.json() if r.text else {}
except Exception as e:
print(" API 请求失败: %s" % str(e))
return None
def get_node_project_list(panel_url, api_key):
for path in ["/project/nodejs/get_project_list", "/plugin?action=a&name=nodejs&s=get_project_list"]:
result = _baota_request(panel_url, api_key, path)
if result and (result.get("status") is True or "data" in result):
return result.get("data", [])
return None
def get_node_project_status(panel_url, api_key, pm2_name):
projects = get_node_project_list(panel_url, api_key)
if projects:
for p in projects:
if p.get("name") == pm2_name:
return p
return None
def start_node_project(panel_url, api_key, pm2_name):
for path in ["/project/nodejs/start_project", "/plugin?action=a&name=nodejs&s=start_project"]:
result = _baota_request(panel_url, api_key, path, {"project_name": pm2_name})
if result and (result.get("status") is True or result.get("msg") or "成功" in str(result)):
print(" [成功] 启动成功: %s" % pm2_name)
return True
return False
def stop_node_project(panel_url, api_key, pm2_name):
for path in ["/project/nodejs/stop_project", "/plugin?action=a&name=nodejs&s=stop_project"]:
result = _baota_request(panel_url, api_key, path, {"project_name": pm2_name})
if result and (result.get("status") is True or result.get("msg") or "成功" in str(result)):
print(" [成功] 停止成功: %s" % pm2_name)
return True
return False
def restart_node_project(panel_url, api_key, pm2_name):
project_status = get_node_project_status(panel_url, api_key, pm2_name)
if project_status:
print(" 项目状态: %s" % project_status.get("status", "未知"))
for path in ["/project/nodejs/restart_project", "/plugin?action=a&name=nodejs&s=restart_project"]:
result = _baota_request(panel_url, api_key, path, {"project_name": pm2_name})
if result and (result.get("status") is True or result.get("msg") or "成功" in str(result)):
print(" [成功] 重启成功: %s" % pm2_name)
return True
if result and "msg" in result:
print(" API 返回: %s" % result.get("msg"))
print(" [警告] 重启失败，请检查宝塔 Node 插件是否安装、API 密钥是否正确")
return False
def add_or_update_node_project(panel_url, api_key, pm2_name, project_path, port=30006, node_path=None):
port_env = "PORT=%d " % port
run_cmd = port_env + ("%s/node server.js" % node_path if node_path else "node server.js")
payload = {"name": pm2_name, "path": project_path, "run_cmd": run_cmd, "port": str(port)}
for path in ["/project/nodejs/add_project", "/plugin?action=a&name=nodejs&s=add_project"]:
result = _baota_request(panel_url, api_key, path, payload)
if result and result.get("status") is True:
print(" [成功] 项目配置已更新: %s" % pm2_name)
return True
if result and "msg" in result:
print(" API 返回: %s" % result.get("msg"))
return False
# ==================== 本地构建 ====================
def run_build(root):
"""执行本地 pnpm build"""
use_shell = sys.platform == "win32"
standalone = os.path.join(root, ".next", "standalone")
server_js = os.path.join(standalone, "server.js")
try:
r = subprocess.run(
["pnpm", "build"],
cwd=root,
shell=use_shell,
timeout=600,
capture_output=True,
text=True,
encoding="utf-8",
errors="replace",
)
stdout_text = r.stdout or ""
stderr_text = r.stderr or ""
combined = stdout_text + stderr_text
is_windows_symlink_error = (
sys.platform == "win32"
and r.returncode != 0
and ("EPERM" in combined or "symlink" in combined.lower() or "operation not permitted" in combined.lower() or "errno: -4048" in combined)
)
if r.returncode != 0:
if is_windows_symlink_error:
print(" [警告] Windows 符号链接权限错误（EPERM）")
print(" 解决方案：开启开发者模式 / 以管理员运行 / 或使用 --no-build")
if os.path.isdir(standalone) and os.path.isfile(server_js):
print(" [成功] standalone 输出可用，继续部署")
return True
return False
print(" [失败] 构建失败，退出码:", r.returncode)
for line in (stdout_text.strip().split("\n") or [])[-10:]:
print(" " + line)
return False
except subprocess.TimeoutExpired:
print(" [失败] 构建超时（超过10分钟）")
return False
except FileNotFoundError:
print(" [失败] 未找到 pnpm，请安装: npm install -g pnpm")
return False
except Exception as e:
print(" [失败] 构建异常:", str(e))
if os.path.isdir(standalone) and os.path.isfile(server_js):
print(" [提示] 可尝试使用 --no-build 跳过构建")
return False
if not os.path.isdir(standalone) or not os.path.isfile(server_js):
print(" [失败] 未找到 .next/standalone 或 server.js")
return False
print(" [成功] 构建完成")
return True
def clean_standalone_before_build(root, retries=3, delay=2):
"""构建前删除 .next/standalone，避免 Windows EBUSY"""
standalone = os.path.join(root, ".next", "standalone")
if not os.path.isdir(standalone):
return True
for attempt in range(1, retries + 1):
try:
shutil.rmtree(standalone)
print(" [清理] 已删除 .next/standalone（第 %d 次尝试）" % attempt)
return True
except (OSError, PermissionError):
if attempt < retries:
print(" [清理] 被占用，%ds 后重试 (%d/%d) ..." % (delay, attempt, retries))
time.sleep(delay)
else:
print(" [失败] 无法删除 .next/standalone，可改用 --no-build")
return False
return False
# ==================== 打包（deploy 模式：tar.gz） ====================
def _copy_with_dereference(src, dst):
if os.path.islink(src):
link_target = os.readlink(src)
real_path = link_target if os.path.isabs(link_target) else os.path.join(os.path.dirname(src), link_target)
if os.path.exists(real_path):
if os.path.isdir(real_path):
shutil.copytree(real_path, dst, symlinks=False, dirs_exist_ok=True)
else:
shutil.copy2(real_path, dst)
else:
shutil.copy2(src, dst, follow_symlinks=False)
elif os.path.isdir(src):
if os.path.exists(dst):
shutil.rmtree(dst)
shutil.copytree(src, dst, symlinks=False, dirs_exist_ok=True)
else:
shutil.copy2(src, dst)
def pack_standalone_tar(root):
"""打包 standalone 为 tar.gz（deploy 模式用）"""
print("[2/4] 打包 standalone ...")
standalone = os.path.join(root, ".next", "standalone")
static_src = os.path.join(root, ".next", "static")
public_src = os.path.join(root, "public")
ecosystem_src = os.path.join(root, "ecosystem.config.cjs")
if not os.path.isdir(standalone) or not os.path.isdir(static_src):
print(" [失败] 未找到 .next/standalone 或 .next/static")
return None
staging = tempfile.mkdtemp(prefix="soul_deploy_")
try:
for name in os.listdir(standalone):
_copy_with_dereference(os.path.join(standalone, name), os.path.join(staging, name))
node_modules_dst = os.path.join(staging, "node_modules")
pnpm_dir = os.path.join(node_modules_dst, ".pnpm")
if os.path.isdir(pnpm_dir):
for dep in ["styled-jsx"]:
dep_in_root = os.path.join(node_modules_dst, dep)
if not os.path.exists(dep_in_root):
for pnpm_pkg in os.listdir(pnpm_dir):
if pnpm_pkg.startswith(dep + "@"):
src_dep = os.path.join(pnpm_dir, pnpm_pkg, "node_modules", dep)
if os.path.isdir(src_dep):
shutil.copytree(src_dep, dep_in_root, symlinks=False, dirs_exist_ok=True)
break
static_dst = os.path.join(staging, ".next", "static")
if os.path.exists(static_dst):
shutil.rmtree(static_dst)
os.makedirs(os.path.dirname(static_dst), exist_ok=True)
shutil.copytree(static_src, static_dst)
if os.path.isdir(public_src):
shutil.copytree(public_src, os.path.join(staging, "public"), dirs_exist_ok=True)
if os.path.isfile(ecosystem_src):
shutil.copy2(ecosystem_src, os.path.join(staging, "ecosystem.config.cjs"))
pkg_json = os.path.join(staging, "package.json")
if os.path.isfile(pkg_json):
try:
with open(pkg_json, "r", encoding="utf-8") as f:
data = json.load(f)
data.setdefault("scripts", {})["start"] = "node server.js"
with open(pkg_json, "w", encoding="utf-8") as f:
json.dump(data, f, indent=2, ensure_ascii=False)
except Exception:
pass
tarball = os.path.join(tempfile.gettempdir(), "soul_deploy.tar.gz")
with tarfile.open(tarball, "w:gz") as tf:
for name in os.listdir(staging):
tf.add(os.path.join(staging, name), arcname=name)
print(" [成功] 打包完成: %s (%.2f MB)" % (tarball, os.path.getsize(tarball) / 1024 / 1024))
return tarball
except Exception as e:
print(" [失败] 打包异常:", str(e))
return None
finally:
shutil.rmtree(staging, ignore_errors=True)
# ==================== Node 环境检查 & SSH 上传（deploy 模式） ====================
def check_node_environments(cfg):
print("[检查] Node 环境 ...")
client = paramiko.SSHClient()
client.set_missing_host_key_policy(paramiko.AutoAddPolicy())
try:
if cfg.get("ssh_key"):
client.connect(cfg["host"], username=cfg["user"], key_filename=cfg["ssh_key"], timeout=15)
else:
client.connect(cfg["host"], username=cfg["user"], password=cfg["password"], timeout=15)
stdin, stdout, stderr = client.exec_command("which node && node -v", timeout=10)
print(" 默认 Node: %s" % (stdout.read().decode("utf-8", errors="replace").strip() or "未找到"))
node_path = cfg.get("node_path", "/www/server/nodejs/v22.14.0/bin")
stdin, stdout, stderr = client.exec_command("%s/node -v 2>/dev/null" % node_path, timeout=5)
print(" 配置 Node: %s" % (stdout.read().decode("utf-8", errors="replace").strip() or "不可用"))
return True
except Exception as e:
print(" [警告] %s" % str(e))
return False
finally:
client.close()
def upload_and_extract(cfg, tarball_path):
"""SSH 上传 tar.gz 并解压到 project_path（deploy 模式）"""
print("[3/4] SSH 上传并解压 ...")
if not cfg.get("password") and not cfg.get("ssh_key"):
print(" [失败] 请设置 DEPLOY_PASSWORD 或 DEPLOY_SSH_KEY")
return False
client = paramiko.SSHClient()
client.set_missing_host_key_policy(paramiko.AutoAddPolicy())
try:
if cfg.get("ssh_key") and os.path.isfile(cfg["ssh_key"]):
client.connect(cfg["host"], username=cfg["user"], key_filename=cfg["ssh_key"], timeout=15)
else:
client.connect(cfg["host"], username=cfg["user"], password=cfg["password"], timeout=15)
sftp = client.open_sftp()
remote_tar = "/tmp/soulTest_deploy.tar.gz"
remote_script = "/tmp/soulTest_deploy_extract.sh"
sftp.put(tarball_path, remote_tar)
node_path = cfg.get("node_path", "/www/server/nodejs/v22.14.0/bin")
project_path = cfg["project_path"]
script_content = """#!/bin/bash
export PATH=%s:$PATH
cd %s
rm -rf .next public ecosystem.config.cjs server.js package.json 2>/dev/null
tar -xzf %s
rm -f %s
echo OK
""" % (node_path, project_path, remote_tar, remote_tar)
with sftp.open(remote_script, "w") as f:
f.write(script_content)
sftp.close()
client.exec_command("chmod +x %s" % remote_script, timeout=10)
stdin, stdout, stderr = client.exec_command("bash %s" % remote_script, timeout=120)
out = stdout.read().decode("utf-8", errors="replace").strip()
exit_status = stdout.channel.recv_exit_status()
if exit_status != 0 or "OK" not in out:
print(" [失败] 解压失败，退出码:", exit_status)
return False
print(" [成功] 解压完成: %s" % project_path)
return True
except Exception as e:
print(" [失败] SSH 错误:", str(e))
return False
finally:
client.close()
def deploy_via_baota_api(cfg):
"""宝塔 API 重启 Node 项目（deploy 模式）"""
print("[4/4] 宝塔 API 管理 Node 项目 ...")
panel_url, api_key, pm2_name = cfg["panel_url"], cfg["api_key"], cfg["pm2_name"]
project_path = cfg["project_path"]
node_path = cfg.get("node_path", "/www/server/nodejs/v22.14.0/bin")
port = cfg.get("port", 30006)
if not get_node_project_status(panel_url, api_key, pm2_name):
add_or_update_node_project(panel_url, api_key, pm2_name, project_path, port, node_path)
stop_node_project(panel_url, api_key, pm2_name)
time.sleep(2)
ok = restart_node_project(panel_url, api_key, pm2_name)
if not ok:
ok = start_node_project(panel_url, api_key, pm2_name)
if not ok:
print(" 请到宝塔 Node 项目手动重启 %s，路径: %s" % (pm2_name, project_path))
return ok
# ==================== 打包（devlop 模式：zip） ====================
ZIP_EXCLUDE_DIRS = {".cache", "__pycache__", ".git", "node_modules", "cache", "test", "tests", "coverage", ".nyc_output", ".turbo", "开发文档", "miniprogramPre", "my-app", "newpp"}
ZIP_EXCLUDE_FILE_NAMES = {".DS_Store", "Thumbs.db"}
ZIP_EXCLUDE_FILE_SUFFIXES = (".log", ".map")
def _should_exclude_from_zip(arcname, is_file=True):
parts = arcname.replace("\\", "/").split("/")
for part in parts:
if part in ZIP_EXCLUDE_DIRS:
return True
if is_file and parts:
name = parts[-1]
if name in ZIP_EXCLUDE_FILE_NAMES or any(name.endswith(s) for s in ZIP_EXCLUDE_FILE_SUFFIXES):
return True
return False
def pack_standalone_zip(root):
"""打包 standalone 为 zip（devlop 模式用）"""
print("[2/7] 打包 standalone 为 zip ...")
standalone = os.path.join(root, ".next", "standalone")
static_src = os.path.join(root, ".next", "static")
public_src = os.path.join(root, "public")
ecosystem_src = os.path.join(root, "ecosystem.config.cjs")
if not os.path.isdir(standalone) or not os.path.isdir(static_src):
print(" [失败] 未找到 .next/standalone 或 .next/static")
return None
staging = tempfile.mkdtemp(prefix="soul_devlop_")
try:
for name in os.listdir(standalone):
_copy_with_dereference(os.path.join(standalone, name), os.path.join(staging, name))
node_modules_dst = os.path.join(staging, "node_modules")
pnpm_dir = os.path.join(node_modules_dst, ".pnpm")
if os.path.isdir(pnpm_dir):
for dep in ["styled-jsx"]:
dep_in_root = os.path.join(node_modules_dst, dep)
if not os.path.exists(dep_in_root):
for pnpm_pkg in os.listdir(pnpm_dir):
if pnpm_pkg.startswith(dep + "@"):
src_dep = os.path.join(pnpm_dir, pnpm_pkg, "node_modules", dep)
if os.path.isdir(src_dep):
shutil.copytree(src_dep, dep_in_root, symlinks=False, dirs_exist_ok=True)
break
os.makedirs(os.path.join(staging, ".next"), exist_ok=True)
shutil.copytree(static_src, os.path.join(staging, ".next", "static"), dirs_exist_ok=True)
if os.path.isdir(public_src):
shutil.copytree(public_src, os.path.join(staging, "public"), dirs_exist_ok=True)
if os.path.isfile(ecosystem_src):
shutil.copy2(ecosystem_src, os.path.join(staging, "ecosystem.config.cjs"))
pkg_json = os.path.join(staging, "package.json")
if os.path.isfile(pkg_json):
try:
with open(pkg_json, "r", encoding="utf-8") as f:
data = json.load(f)
data.setdefault("scripts", {})["start"] = "node server.js"
with open(pkg_json, "w", encoding="utf-8") as f:
json.dump(data, f, indent=2, ensure_ascii=False)
except Exception:
pass
server_js = os.path.join(staging, "server.js")
if os.path.isfile(server_js):
try:
with open(server_js, "r", encoding="utf-8") as f:
c = f.read()
if "|| 3000" in c:
with open(server_js, "w", encoding="utf-8") as f:
f.write(c.replace("|| 3000", "|| 30006"))
except Exception:
pass
zip_path = os.path.join(tempfile.gettempdir(), "soul_devlop.zip")
with zipfile.ZipFile(zip_path, "w", zipfile.ZIP_DEFLATED) as zf:
for name in os.listdir(staging):
path = os.path.join(staging, name)
if os.path.isfile(path):
if not _should_exclude_from_zip(name):
zf.write(path, name)
else:
for dirpath, dirs, filenames in os.walk(path):
dirs[:] = [d for d in dirs if not _should_exclude_from_zip(os.path.join(name, os.path.relpath(os.path.join(dirpath, d), path)), is_file=False)]
for f in filenames:
full = os.path.join(dirpath, f)
arcname = os.path.join(name, os.path.relpath(full, path))
if not _should_exclude_from_zip(arcname):
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
finally:
shutil.rmtree(staging, ignore_errors=True)
def upload_zip_and_extract_to_dist2(cfg, zip_path):
"""上传 zip 并解压到 dist2（devlop 模式）"""
print("[3/7] SSH 上传 zip 并解压到 dist2 ...")
sys.stdout.flush()
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)
client = paramiko.SSHClient()
client.set_missing_host_key_policy(paramiko.AutoAddPolicy())
try:
print(" 正在连接 %s@%s ..." % (cfg["user"], cfg["host"]))
sys.stdout.flush()
if cfg.get("ssh_key") and os.path.isfile(cfg["ssh_key"]):
client.connect(cfg["host"], username=cfg["user"], key_filename=cfg["ssh_key"], timeout=30, banner_timeout=30)
else:
client.connect(cfg["host"], username=cfg["user"], password=cfg["password"], timeout=30, banner_timeout=30)
print(" [OK] SSH 已连接，正在上传 zip（%.1f MB）..." % zip_size_mb)
sys.stdout.flush()
remote_zip = cfg["base_path"].rstrip("/") + "/soulTest_devlop.zip"
sftp = client.open_sftp()
# 上传进度：每 5MB 打印一次
chunk_mb = 5.0
last_reported = [0]
def _progress(transferred, total):
if total and total > 0:
now_mb = transferred / (1024 * 1024)
if now_mb - last_reported[0] >= chunk_mb or transferred >= total:
last_reported[0] = now_mb
print("\r 上传进度: %.1f / %.1f MB" % (now_mb, total / (1024 * 1024)), end="")
sys.stdout.flush()
sftp.put(zip_path, remote_zip, callback=_progress)
if zip_size_mb >= chunk_mb:
print("")
print(" [OK] zip 已上传，正在服务器解压（约 1–3 分钟）...")
sys.stdout.flush()
sftp.close()
dist2 = cfg["dist2_path"]
cmd = "rm -rf %s && mkdir -p %s && unzip -o -q %s -d %s && rm -f %s && echo OK" % (dist2, dist2, remote_zip, dist2, remote_zip)
stdin, stdout, stderr = client.exec_command(cmd, timeout=300)
out = stdout.read().decode("utf-8", errors="replace").strip()
err = stderr.read().decode("utf-8", errors="replace").strip()
if err:
print(" 服务器 stderr: %s" % err[:500])
exit_status = stdout.channel.recv_exit_status()
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))
import traceback
traceback.print_exc()
return False
finally:
client.close()
def run_pnpm_install_in_dist2(cfg):
"""服务器 dist2 内执行 pnpm install（devlop 模式）"""
print("[4/7] 服务器 dist2 内执行 pnpm install ...")
client = paramiko.SSHClient()
client.set_missing_host_key_policy(paramiko.AutoAddPolicy())
try:
if cfg.get("ssh_key") and os.path.isfile(cfg["ssh_key"]):
client.connect(cfg["host"], username=cfg["user"], key_filename=cfg["ssh_key"], timeout=15)
else:
client.connect(cfg["host"], username=cfg["user"], password=cfg["password"], timeout=15)
stdin, stdout, stderr = client.exec_command("bash -lc 'which pnpm'", timeout=10)
pnpm_path = stdout.read().decode("utf-8", errors="replace").strip()
if not pnpm_path:
return False, "未找到 pnpm，请服务器安装: npm install -g pnpm"
cmd = "bash -lc 'cd %s && %s install'" % (cfg["dist2_path"], pnpm_path)
stdin, stdout, stderr = client.exec_command(cmd, timeout=300)
out = stdout.read().decode("utf-8", errors="replace").strip()
err = stderr.read().decode("utf-8", errors="replace").strip()
if stdout.channel.recv_exit_status() != 0:
return False, "pnpm install 失败\n" + (err or out)
print(" [成功] pnpm install 完成")
return True, None
except Exception as e:
return False, str(e)
finally:
client.close()
def remote_swap_dist_and_restart(cfg):
"""暂停 → dist→dist1, dist2→dist → 删除 dist1 → 重启（devlop 模式）"""
print("[5/7] 宝塔 API 暂停 Node 项目 ...")
stop_node_project(cfg["panel_url"], cfg["api_key"], cfg["pm2_name"])
time.sleep(2)
print("[6/7] 服务器切换目录: dist→dist1, dist2→dist ...")
client = paramiko.SSHClient()
client.set_missing_host_key_policy(paramiko.AutoAddPolicy())
try:
if cfg.get("ssh_key") and os.path.isfile(cfg["ssh_key"]):
client.connect(cfg["host"], username=cfg["user"], key_filename=cfg["ssh_key"], timeout=15)
else:
client.connect(cfg["host"], username=cfg["user"], password=cfg["password"], timeout=15)
cmd = "cd %s && mv dist dist1 2>/dev/null; mv dist2 dist && rm -rf dist1 && echo OK" % cfg["base_path"]
stdin, stdout, stderr = client.exec_command(cmd, timeout=60)
out = stdout.read().decode("utf-8", errors="replace").strip()
if stdout.channel.recv_exit_status() != 0 or "OK" not in out:
print(" [失败] 切换失败")
return False
print(" [成功] 新版本位于 %s" % cfg["dist_path"])
finally:
client.close()
print("[7/7] 宝塔 API 重启 Node 项目 ...")
if not start_node_project(cfg["panel_url"], cfg["api_key"], cfg["pm2_name"]):
print(" [警告] 请到宝塔手动启动 %s" % cfg["pm2_name"])
return False
return True
# ==================== 主函数 ====================
def main():
parser = argparse.ArgumentParser(description="Soul 创业派对 - 统一部署脚本", formatter_class=argparse.RawDescriptionHelpFormatter, epilog=__doc__)
parser.add_argument("--mode", choices=["devlop", "deploy"], default="devlop", help="devlop=dist切换(默认), deploy=直接覆盖")
parser.add_argument("--no-build", action="store_true", help="跳过本地构建")
parser.add_argument("--no-upload", action="store_true", help="仅 deploy 模式：跳过 SSH 上传")
parser.add_argument("--no-api", action="store_true", help="仅 deploy 模式：上传后不调宝塔 API")
args = parser.parse_args()
script_dir = os.path.dirname(os.path.abspath(__file__))
root = os.path.dirname(script_dir)
if args.mode == "devlop":
cfg = get_cfg_devlop()
print("=" * 60)
print(" Soul 自动部署（dist 切换）")
print("=" * 60)
print(" 服务器: %s@%s 目录: %s Node: %s" % (cfg["user"], cfg["host"], cfg["base_path"], cfg["pm2_name"]))
print("=" * 60)
if not args.no_build:
print("[1/7] 本地构建 pnpm build ...")
if sys.platform == "win32" and not clean_standalone_before_build(root):
return 1
if not run_build(root):
return 1
elif not os.path.isfile(os.path.join(root, ".next", "standalone", "server.js")):
print("[错误] 未找到 .next/standalone/server.js")
return 1
else:
print("[1/7] 跳过本地构建")
zip_path = pack_standalone_zip(root)
if not zip_path:
return 1
if not upload_zip_and_extract_to_dist2(cfg, zip_path):
return 1
try:
os.remove(zip_path)
except Exception:
pass
ok, err = run_pnpm_install_in_dist2(cfg)
if not ok:
print(" [失败] %s" % (err or "pnpm install 失败"))
return 1
if not remote_swap_dist_and_restart(cfg):
return 1
print("")
print(" 部署完成！运行目录: %s" % cfg["dist_path"])
return 0
# deploy 模式
cfg = get_cfg()
print("=" * 60)
print(" Soul 一键部署（直接覆盖）")
print("=" * 60)
print(" 服务器: %s@%s 项目路径: %s PM2: %s" % (cfg["user"], cfg["host"], cfg["project_path"], cfg["pm2_name"]))
print("=" * 60)
if not args.no_upload:
check_node_environments(cfg)
if not args.no_build:
print("[1/4] 本地构建 ...")
if not run_build(root):
return 1
elif not os.path.isfile(os.path.join(root, ".next", "standalone", "server.js")):
print("[错误] 未找到 .next/standalone/server.js")
return 1
else:
print("[1/4] 跳过本地构建")
tarball = pack_standalone_tar(root)
if not tarball:
return 1
if not args.no_upload:
if not upload_and_extract(cfg, tarball):
return 1
try:
os.remove(tarball)
except Exception:
pass
else:
print(" 压缩包: %s" % tarball)
if not args.no_api and not args.no_upload:
deploy_via_baota_api(cfg)
print("")
print(" 部署完成！站点: %s" % cfg["site_url"])
return 0
if __name__ == "__main__":
sys.exit(main())

855
scripts/deploy_soul.py
View File

@@ -1,855 +0,0 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
Soul 创业派对 - 一键部署脚本
本地打包 + SSH 上传解压 + 宝塔 API 部署
使用方法:
python scripts/deploy_soul.py # 完整部署流程
python scripts/deploy_soul.py --no-build # 跳过本地构建
python scripts/deploy_soul.py --no-upload # 跳过 SSH 上传
python scripts/deploy_soul.py --no-api # 上传后不调宝塔 API 重启
环境变量（可选，覆盖默认配置）:
DEPLOY_HOST # SSH 服务器地址，默认 42.194.232.22
DEPLOY_USER # SSH 用户名，默认 root
DEPLOY_PASSWORD # SSH 密码，默认 Zhiqun1984
DEPLOY_SSH_KEY # SSH 密钥路径（优先于密码）
DEPLOY_PROJECT_PATH # 服务器项目路径，默认 /www/wwwroot/soul
BAOTA_PANEL_URL # 宝塔面板地址，默认 https://42.194.232.22:9988
BAOTA_API_KEY # 宝塔 API 密钥，默认 hsAWqFSi0GOCrunhmYdkxy92tBXfqYjd
DEPLOY_PM2_APP # PM2 项目名称，默认 soul
DEPLOY_PORT # Next.js 监听端口，默认 30006（与 package.json / ecosystem 一致）
DEPLOY_NODE_VERSION # Node 版本，默认 v22.14.0（用于显示）
DEPLOY_NODE_PATH # Node 可执行文件路径，默认 /www/server/nodejs/v22.14.0/bin
# 用于避免多 Node 环境冲突，确保使用指定的 Node 版本
"""
from __future__ import print_function
import os
import sys
import shutil
import tarfile
import tempfile
import subprocess
import argparse
import time
import hashlib
# 检查依赖
try:
import paramiko
except ImportError:
print("错误: 请先安装 paramiko")
print(" pip install paramiko")
sys.exit(1)
try:
import requests
import urllib3
urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)
except ImportError:
print("错误: 请先安装 requests")
print(" pip install requests")
sys.exit(1)
# ==================== 配置 ====================
def get_cfg():
"""获取部署配置"""
return {
# SSH 配置
"host": os.environ.get("DEPLOY_HOST", "42.194.232.22"),
"user": os.environ.get("DEPLOY_USER", "root"),
"password": os.environ.get("DEPLOY_PASSWORD", "Zhiqun1984"),
"ssh_key": os.environ.get("DEPLOY_SSH_KEY", ""),
"project_path": os.environ.get("DEPLOY_PROJECT_PATH", "/www/wwwroot/soul"),
# 宝塔 API 配置
"panel_url": os.environ.get("BAOTA_PANEL_URL", "https://42.194.232.22:9988"),
"api_key": os.environ.get("BAOTA_API_KEY", "hsAWqFSi0GOCrunhmYdkxy92tBXfqYjd"),
"pm2_name": os.environ.get("DEPLOY_PM2_APP", "soul"),
"site_url": os.environ.get("DEPLOY_SITE_URL", "https://soul.quwanzhi.com"),
"port": int(os.environ.get("DEPLOY_PORT", "30006")), # Next.js 监听端口，与 package.json / ecosystem 一致
# Node 环境配置
"node_version": os.environ.get("DEPLOY_NODE_VERSION", "v22.14.0"), # 指定 Node 版本
"node_path": os.environ.get("DEPLOY_NODE_PATH", "/www/server/nodejs/v22.14.0/bin"), # Node 可执行文件路径
}
# ==================== 宝塔 API ====================
def _get_sign(api_key):
"""宝塔鉴权签名：request_token = md5(request_time + md5(api_key))"""
now_time = int(time.time())
sign_str = str(now_time) + hashlib.md5(api_key.encode("utf-8")).hexdigest()
request_token = hashlib.md5(sign_str.encode("utf-8")).hexdigest()
return now_time, request_token
def _baota_request(panel_url, api_key, path, data=None):
"""发起宝塔 API 请求的通用函数"""
req_time, req_token = _get_sign(api_key)
payload = {
"request_time": req_time,
"request_token": req_token,
}
if data:
payload.update(data)
url = panel_url.rstrip("/") + "/" + path.lstrip("/")
try:
r = requests.post(url, data=payload, verify=False, timeout=30)
if r.text:
return r.json()
return {}
except Exception as e:
print(" API 请求失败: %s" % str(e))
return None
def get_node_project_list(panel_url, api_key):
"""获取 Node 项目列表"""
paths_to_try = [
"/project/nodejs/get_project_list",
"/plugin?action=a&name=nodejs&s=get_project_list",
]
for path in paths_to_try:
result = _baota_request(panel_url, api_key, path)
if result and (result.get("status") is True or "data" in result):
return result.get("data", [])
return None
def get_node_project_status(panel_url, api_key, pm2_name):
"""检查 Node 项目状态"""
projects = get_node_project_list(panel_url, api_key)
if projects:
for project in projects:
if project.get("name") == pm2_name:
return project
return None
def start_node_project(panel_url, api_key, pm2_name):
"""通过宝塔 API 启动 Node 项目"""
paths_to_try = [
"/project/nodejs/start_project",
"/plugin?action=a&name=nodejs&s=start_project",
]
for path in paths_to_try:
result = _baota_request(panel_url, api_key, path, {"project_name": pm2_name})
if result and (result.get("status") is True or result.get("msg") or "成功" in str(result)):
print(" [成功] 启动成功: %s" % pm2_name)
return True
return False
def stop_node_project(panel_url, api_key, pm2_name):
"""通过宝塔 API 停止 Node 项目"""
paths_to_try = [
"/project/nodejs/stop_project",
"/plugin?action=a&name=nodejs&s=stop_project",
]
for path in paths_to_try:
result = _baota_request(panel_url, api_key, path, {"project_name": pm2_name})
if result and (result.get("status") is True or result.get("msg") or "成功" in str(result)):
print(" [成功] 停止成功: %s" % pm2_name)
return True
return False
def restart_node_project(panel_url, api_key, pm2_name):
"""
通过宝塔 API 重启 Node 项目
返回 True 表示成功，False 表示失败
"""
# 先检查项目状态
project_status = get_node_project_status(panel_url, api_key, pm2_name)
if project_status:
print(" 项目状态: %s" % project_status.get("status", "未知"))
paths_to_try = [
"/project/nodejs/restart_project",
"/plugin?action=a&name=nodejs&s=restart_project",
]
for path in paths_to_try:
result = _baota_request(panel_url, api_key, path, {"project_name": pm2_name})
if result:
if result.get("status") is True or result.get("msg") or "成功" in str(result):
print(" [成功] 重启成功: %s" % pm2_name)
return True
if "msg" in result:
print(" API 返回: %s" % result.get("msg"))
print(" [警告] 重启失败，请检查宝塔 Node 插件是否安装、API 密钥是否正确")
return False
def add_or_update_node_project(panel_url, api_key, pm2_name, project_path, port=30006, node_path=None):
"""通过宝塔 API 添加或更新 Node 项目配置
Next.js standalone 的 server.js 通过 process.env.PORT 读端口（默认 3000），
这里在 run_cmd 中显式设置 PORT=port，与项目 package.json / ecosystem 的 30006 一致。
"""
paths_to_try = [
"/project/nodejs/add_project",
"/plugin?action=a&name=nodejs&s=add_project",
]
# Next.js standalone：显式传 PORT，避免宝塔未注入时用默认 3000
port_env = "PORT=%d " % port
if node_path:
run_cmd = port_env + "%s/node server.js" % node_path
else:
run_cmd = port_env + "node server.js"
payload = {
"name": pm2_name,
"path": project_path,
"run_cmd": run_cmd,
"port": str(port),
}
for path in paths_to_try:
result = _baota_request(panel_url, api_key, path, payload)
if result:
if result.get("status") is True:
print(" [成功] 项目配置已更新: %s" % pm2_name)
return True
if "msg" in result:
print(" API 返回: %s" % result.get("msg"))
return False
# ==================== 本地构建 ====================
def run_build(root):
"""执行本地构建"""
print("[1/4] 本地构建 pnpm build ...")
use_shell = sys.platform == "win32"
# 检查 standalone 目录是否已存在
standalone = os.path.join(root, ".next", "standalone")
server_js = os.path.join(standalone, "server.js")
try:
# 在 Windows 上处理编码问题：使用 UTF-8 和 errors='replace' 来避免解码错误
# errors='replace' 会在遇到无法解码的字符时用替换字符代替，避免崩溃
r = subprocess.run(
["pnpm", "build"],
cwd=root,
shell=use_shell,
timeout=600,
capture_output=True,
text=True,
encoding='utf-8',
errors='replace' # 遇到无法解码的字符时替换为占位符，避免 UnicodeDecodeError
)
# 安全地获取输出，处理可能的 None 值
stdout_text = r.stdout or ""
stderr_text = r.stderr or ""
# 检查是否是 Windows 符号链接权限错误
# 错误信息可能在 stdout 或 stderr 中
combined_output = stdout_text + stderr_text
is_windows_symlink_error = (
sys.platform == "win32" and
r.returncode != 0 and
("EPERM" in combined_output or
"symlink" in combined_output.lower() or
"operation not permitted" in combined_output.lower() or
"errno: -4048" in combined_output)
)
if r.returncode != 0:
if is_windows_symlink_error:
print(" [警告] Windows 符号链接权限错误（EPERM）")
print(" 这是 Windows 上 Next.js standalone 构建的常见问题")
print(" 解决方案（任选其一）：")
print(" 1. 开启 Windows 开发者模式：设置 → 隐私和安全性 → 针对开发人员 → 开发人员模式")
print(" 2. 以管理员身份运行终端再执行构建")
print(" 3. 使用 --no-build 跳过构建，使用已有的构建文件")
print("")
print(" 正在检查 standalone 输出是否可用...")
# 即使有错误，也检查 standalone 是否可用
if os.path.isdir(standalone) and os.path.isfile(server_js):
print(" [成功] 虽然构建有警告，但 standalone 输出可用，继续部署")
return True
else:
print(" [失败] standalone 输出不可用，无法继续")
return False
else:
print(" [失败] 构建失败，退出码:", r.returncode)
if stdout_text:
# 显示最后几行输出以便调试
lines = stdout_text.strip().split('\n')
if lines:
print(" 构建输出（最后10行）:")
for line in lines[-10:]:
try:
# 确保输出可以正常显示
print(" " + line)
except UnicodeEncodeError:
# 如果仍有编码问题，使用 ASCII 安全输出
print(" " + line.encode('ascii', 'replace').decode('ascii'))
if stderr_text:
print(" 错误输出（最后5行）:")
lines = stderr_text.strip().split('\n')
if lines:
for line in lines[-5:]:
try:
print(" " + line)
except UnicodeEncodeError:
print(" " + line.encode('ascii', 'replace').decode('ascii'))
return False
except subprocess.TimeoutExpired:
print(" [失败] 构建超时（超过10分钟）")
return False
except FileNotFoundError:
print(" [失败] 未找到 pnpm 命令，请先安装 pnpm")
print(" npm install -g pnpm")
return False
except UnicodeDecodeError as e:
print(" [失败] 构建输出编码错误:", str(e))
print(" 提示: 这可能是 Windows 编码问题，尝试设置环境变量 PYTHONIOENCODING=utf-8")
# 即使有编码错误，也检查 standalone 是否可用
if os.path.isdir(standalone) and os.path.isfile(server_js):
print(" [成功] 虽然构建有编码警告，但 standalone 输出可用，继续部署")
return True
return False
except Exception as e:
print(" [失败] 构建异常:", str(e))
import traceback
traceback.print_exc()
# 即使有异常，也检查 standalone 是否可用（可能是部分成功）
if os.path.isdir(standalone) and os.path.isfile(server_js):
print(" [提示] 检测到 standalone 输出，可能是部分构建成功")
print(" 如果确定要使用，可以使用 --no-build 跳过构建步骤")
return False
# 验证构建输出
if not os.path.isdir(standalone) or not os.path.isfile(server_js):
print(" [失败] 未找到 .next/standalone 或 server.js")
print(" 请确认 next.config.mjs 中设置了 output: 'standalone'")
return False
print(" [成功] 构建完成")
return True
# ==================== 打包 ====================
def pack_standalone(root):
"""打包 standalone 输出"""
print("[2/4] 打包 standalone ...")
standalone = os.path.join(root, ".next", "standalone")
static_src = os.path.join(root, ".next", "static")
public_src = os.path.join(root, "public")
ecosystem_src = os.path.join(root, "ecosystem.config.cjs")
# 检查必要文件
if not os.path.isdir(standalone):
print(" [失败] 未找到 .next/standalone 目录")
return None
if not os.path.isdir(static_src):
print(" [失败] 未找到 .next/static 目录")
return None
if not os.path.isdir(public_src):
print(" [警告] 未找到 public 目录，继续打包")
if not os.path.isfile(ecosystem_src):
print(" [警告] 未找到 ecosystem.config.cjs，继续打包")
staging = tempfile.mkdtemp(prefix="soul_deploy_")
try:
# 复制 standalone 内容
# standalone 目录应该包含：server.js, package.json, node_modules/ 等
print(" 正在复制 standalone 目录内容...")
# 使用更可靠的方法复制，特别是处理 pnpm 的符号链接结构
def copy_with_dereference(src, dst):
"""复制文件或目录，跟随符号链接"""
if os.path.islink(src):
# 如果是符号链接，复制目标文件
link_target = os.readlink(src)
if os.path.isabs(link_target):
real_path = link_target
else:
real_path = os.path.join(os.path.dirname(src), link_target)
if os.path.exists(real_path):
if os.path.isdir(real_path):
shutil.copytree(real_path, dst, symlinks=False, dirs_exist_ok=True)
else:
shutil.copy2(real_path, dst)
else:
# 如果链接目标不存在，直接复制链接本身
shutil.copy2(src, dst, follow_symlinks=False)
elif os.path.isdir(src):
# 对于目录，递归复制并处理符号链接
if os.path.exists(dst):
shutil.rmtree(dst)
shutil.copytree(src, dst, symlinks=False, dirs_exist_ok=True)
else:
shutil.copy2(src, dst)
for name in os.listdir(standalone):
src = os.path.join(standalone, name)
dst = os.path.join(staging, name)
if name == 'node_modules':
print(" 正在复制 node_modules（处理符号链接和 pnpm 结构）...")
copy_with_dereference(src, dst)
else:
copy_with_dereference(src, dst)
# 🔧 修复 pnpm 依赖：将 styled-jsx 从 .pnpm 提升到根 node_modules
print(" 正在修复 pnpm 依赖结构...")
node_modules_dst = os.path.join(staging, "node_modules")
pnpm_dir = os.path.join(node_modules_dst, ".pnpm")
if os.path.isdir(pnpm_dir):
# 需要提升的依赖列表（require-hook.js 需要）
required_deps = ["styled-jsx"]
for dep in required_deps:
dep_in_root = os.path.join(node_modules_dst, dep)
if not os.path.exists(dep_in_root):
# 在 .pnpm 中查找该依赖
for pnpm_pkg in os.listdir(pnpm_dir):
if pnpm_pkg.startswith(dep + "@"):
src_dep = os.path.join(pnpm_dir, pnpm_pkg, "node_modules", dep)
if os.path.isdir(src_dep):
print(" 提升依赖: %s -> node_modules/%s" % (pnpm_pkg, dep))
shutil.copytree(src_dep, dep_in_root, symlinks=False, dirs_exist_ok=True)
break
else:
print(" 依赖已存在: %s" % dep)
# 验证关键文件
server_js = os.path.join(staging, "server.js")
package_json = os.path.join(staging, "package.json")
node_modules = os.path.join(staging, "node_modules")
if not os.path.isfile(server_js):
print(" [警告] standalone 目录内未找到 server.js")
if not os.path.isfile(package_json):
print(" [警告] standalone 目录内未找到 package.json")
if not os.path.isdir(node_modules):
print(" [警告] standalone 目录内未找到 node_modules")
else:
# 检查 node_modules/next 是否存在
next_module = os.path.join(node_modules, "next")
if os.path.isdir(next_module):
print(" [成功] 已确认 node_modules/next 存在")
else:
print(" [警告] node_modules/next 不存在，可能导致运行时错误")
# 检查 styled-jsx 是否存在（require-hook.js 需要）
styled_jsx_module = os.path.join(node_modules, "styled-jsx")
if os.path.isdir(styled_jsx_module):
print(" [成功] 已确认 node_modules/styled-jsx 存在")
else:
print(" [警告] node_modules/styled-jsx 不存在，可能导致启动失败")
# 复制 .next/static
static_dst = os.path.join(staging, ".next", "static")
if os.path.exists(static_dst):
shutil.rmtree(static_dst)
os.makedirs(os.path.dirname(static_dst), exist_ok=True)
shutil.copytree(static_src, static_dst)
# 复制 public
if os.path.isdir(public_src):
public_dst = os.path.join(staging, "public")
if os.path.exists(public_dst):
shutil.rmtree(public_dst)
shutil.copytree(public_src, public_dst)
# 复制 ecosystem.config.cjs
if os.path.isfile(ecosystem_src):
shutil.copy2(ecosystem_src, os.path.join(staging, "ecosystem.config.cjs"))
# 确保 package.json 的 start 脚本正确（standalone 模式使用 node server.js）
package_json_path = os.path.join(staging, "package.json")
if os.path.isfile(package_json_path):
try:
import json
with open(package_json_path, 'r', encoding='utf-8') as f:
package_data = json.load(f)
# 确保 start 脚本使用 node server.js
if 'scripts' not in package_data:
package_data['scripts'] = {}
package_data['scripts']['start'] = 'node server.js'
with open(package_json_path, 'w', encoding='utf-8') as f:
json.dump(package_data, f, indent=2, ensure_ascii=False)
print(" [提示] 已修正 package.json 的 start 脚本为 'node server.js'")
except Exception as e:
print(" [警告] 无法修正 package.json:", str(e))
# 创建压缩包
tarball = os.path.join(tempfile.gettempdir(), "soul_deploy.tar.gz")
with tarfile.open(tarball, "w:gz") as tf:
for name in os.listdir(staging):
tf.add(os.path.join(staging, name), arcname=name)
size_mb = os.path.getsize(tarball) / 1024 / 1024
print(" [成功] 打包完成: %s (%.2f MB)" % (tarball, size_mb))
return tarball
except Exception as e:
print(" [失败] 打包异常:", str(e))
import traceback
traceback.print_exc()
return None
finally:
shutil.rmtree(staging, ignore_errors=True)
# ==================== Node 环境检查 ====================
def check_node_environments(cfg):
"""检查服务器上的 Node 环境"""
print("[检查] Node 环境检查 ...")
host = cfg["host"]
user = cfg["user"]
password = cfg["password"]
key_path = cfg["ssh_key"]
client = paramiko.SSHClient()
client.set_missing_host_key_policy(paramiko.AutoAddPolicy())
try:
if key_path:
client.connect(host, username=user, key_filename=key_path, timeout=15)
else:
client.connect(host, username=user, password=password, timeout=15)
# 检查系统默认 Node 版本
stdin, stdout, stderr = client.exec_command("which node && node -v", timeout=10)
default_node = stdout.read().decode("utf-8", errors="replace").strip()
if default_node:
print(" 系统默认 Node: %s" % default_node)
else:
print(" 警告: 未找到系统默认 Node")
# 检查宝塔安装的 Node 版本
stdin, stdout, stderr = client.exec_command("ls -d /www/server/nodejs/*/ 2>/dev/null | head -5", timeout=10)
node_versions = stdout.read().decode("utf-8", errors="replace").strip().split('\n')
node_versions = [v.strip().rstrip('/') for v in node_versions if v.strip()]
if node_versions:
print(" 宝塔 Node 版本列表:")
for version_path in node_versions:
version_name = version_path.split('/')[-1]
# 检查该版本的 Node 是否存在
stdin2, stdout2, stderr2 = client.exec_command("%s/node -v 2>/dev/null" % version_path, timeout=5)
node_ver = stdout2.read().decode("utf-8", errors="replace").strip()
if node_ver:
print(" - %s: %s" % (version_name, node_ver))
else:
print(" - %s: (不可用)" % version_name)
else:
print(" 警告: 未找到宝塔 Node 安装目录")
# 检查配置的 Node 版本
node_path = cfg.get("node_path", "/www/server/nodejs/v22.14.0/bin")
stdin, stdout, stderr = client.exec_command("%s/node -v 2>/dev/null" % node_path, timeout=5)
configured_node = stdout.read().decode("utf-8", errors="replace").strip()
if configured_node:
print(" 配置的 Node 版本: %s (%s)" % (configured_node, node_path))
else:
print(" 警告: 配置的 Node 路径不可用: %s" % node_path)
if node_versions:
# 自动使用第一个可用的版本
suggested_path = node_versions[0] + "/bin"
print(" 建议使用: %s" % suggested_path)
return True
except Exception as e:
print(" [警告] Node 环境检查失败: %s" % str(e))
return False
finally:
client.close()
# ==================== SSH 上传 ====================
def upload_and_extract(cfg, tarball_path):
"""SSH 上传并解压"""
print("[3/4] SSH 上传并解压 ...")
host = cfg["host"]
user = cfg["user"]
password = cfg["password"]
key_path = cfg["ssh_key"]
project_path = cfg["project_path"]
node_path = cfg.get("node_path", "/www/server/nodejs/v22.14.0/bin")
if not password and not key_path:
print(" [失败] 请设置 DEPLOY_PASSWORD 或 DEPLOY_SSH_KEY")
return False
client = paramiko.SSHClient()
client.set_missing_host_key_policy(paramiko.AutoAddPolicy())
try:
# 连接 SSH
print(" 正在连接 %s@%s ..." % (user, host))
if key_path:
if not os.path.isfile(key_path):
print(" [失败] SSH 密钥文件不存在: %s" % key_path)
return False
client.connect(host, username=user, key_filename=key_path, timeout=15)
else:
client.connect(host, username=user, password=password, timeout=15)
print(" [成功] SSH 连接成功")
# 上传文件和解压脚本
print(" 正在上传压缩包和脚本 ...")
sftp = client.open_sftp()
remote_tar = "/tmp/soul_deploy.tar.gz"
remote_script = "/tmp/soul_deploy_extract.sh"
try:
# 上传压缩包
sftp.put(tarball_path, remote_tar)
print(" [成功] 压缩包上传完成")
# 构建解压脚本，使用 bash 脚本文件避免语法错误
# 在脚本中指定使用特定的 Node 版本，避免多环境冲突
verify_script_content = """#!/bin/bash
# 设置 Node 环境路径，避免多环境冲突
export PATH=%s:$PATH
cd %s
rm -rf .next public ecosystem.config.cjs 2>/dev/null
rm -f server.js package.json 2>/dev/null
tar -xzf %s
rm -f %s
# 显示使用的 Node 版本
echo "使用 Node 版本: $(node -v)"
echo "Node 路径: $(which node)"
# 验证 node_modules/next 和 styled-jsx
echo "检查关键依赖..."
if [ ! -d 'node_modules/next' ] || [ ! -f 'node_modules/next/dist/server/require-hook.js' ]; then
echo '警告: node_modules/next 不完整'
fi
# 检查 styled-jsx（require-hook.js 需要）
if [ ! -d 'node_modules/styled-jsx' ]; then
echo '警告: styled-jsx 缺失，正在修复...'
# 尝试从 .pnpm 创建链接
if [ -d 'node_modules/.pnpm' ]; then
STYLED_JSX_DIR=$(find node_modules/.pnpm -maxdepth 1 -type d -name "styled-jsx@*" | head -1)
if [ -n "$STYLED_JSX_DIR" ]; then
echo "从 .pnpm 链接 styled-jsx: $STYLED_JSX_DIR"
ln -sf "$STYLED_JSX_DIR/node_modules/styled-jsx" node_modules/styled-jsx
fi
fi
fi
# 如果还是缺失，运行 npm install
if [ ! -d 'node_modules/styled-jsx' ]; then
if [ -f 'package.json' ] && command -v npm >/dev/null 2>&1; then
echo '运行 npm install --production 修复依赖...'
npm install --production --no-save 2>&1 | tail -10 || echo 'npm install 失败'
else
echo '无法自动修复: 缺少 package.json 或 npm 命令'
fi
fi
# 最终验证
echo "最终验证..."
if [ -d 'node_modules/next' ] && [ -f 'node_modules/next/dist/server/require-hook.js' ]; then
echo '✓ node_modules/next 存在'
else
echo '✗ node_modules/next 缺失'
fi
if [ -d 'node_modules/styled-jsx' ]; then
echo '✓ node_modules/styled-jsx 存在'
else
echo '✗ node_modules/styled-jsx 缺失（可能导致启动失败）'
fi
echo '解压完成'
""" % (node_path, project_path, remote_tar, remote_tar)
# 写入脚本文件
with sftp.open(remote_script, 'w') as f:
f.write(verify_script_content)
print(" [成功] 解压脚本上传完成")
finally:
sftp.close()
# 设置执行权限并执行脚本
print(" 正在解压并验证依赖...")
client.exec_command("chmod +x %s" % remote_script, timeout=10)
cmd = "bash %s" % remote_script
stdin, stdout, stderr = client.exec_command(cmd, timeout=120)
err = stderr.read().decode("utf-8", errors="replace").strip()
if err:
print(" 服务器 stderr:", err)
output = stdout.read().decode("utf-8", errors="replace").strip()
exit_status = stdout.channel.recv_exit_status()
if exit_status != 0:
print(" [失败] 解压失败，退出码:", exit_status)
return False
print(" [成功] 解压完成: %s" % project_path)
return True
except paramiko.AuthenticationException:
print(" [失败] SSH 认证失败，请检查用户名和密码")
return False
except paramiko.SSHException as e:
print(" [失败] SSH 连接异常:", str(e))
return False
except Exception as e:
print(" [失败] SSH 错误:", str(e))
import traceback
traceback.print_exc()
return False
finally:
client.close()
# ==================== 宝塔 API 部署 ====================
def deploy_via_baota_api(cfg):
"""通过宝塔 API 管理 Node 项目部署（针对 Next.js standalone：node server.js + PORT）"""
print("[4/4] 宝塔 API 管理 Node 项目 ...")
panel_url = cfg["panel_url"]
api_key = cfg["api_key"]
pm2_name = cfg["pm2_name"]
project_path = cfg["project_path"]
node_path = cfg.get("node_path", "/www/server/nodejs/v22.14.0/bin")
port = cfg.get("port", 30006) # 与 package.json dev/start -p 30006、ecosystem PORT 一致
# 1. 检查项目是否存在
print(" 检查项目状态...")
project_status = get_node_project_status(panel_url, api_key, pm2_name)
if not project_status:
print(" 项目不存在，尝试添加项目配置...")
# 尝试添加项目（如果项目不存在，这个操作可能会失败，但不影响后续重启）
# 使用指定的 Node 路径，避免多环境冲突
add_or_update_node_project(panel_url, api_key, pm2_name, project_path, port, node_path)
else:
print(" 项目已存在: %s" % pm2_name)
current_status = project_status.get("status", "未知")
print(" 当前状态: %s" % current_status)
# 检查启动命令是否使用了正确的 Node 路径
run_cmd = project_status.get("run_cmd", "")
if run_cmd and "node server.js" in run_cmd and node_path not in run_cmd:
print(" 警告: 项目启动命令可能未使用指定的 Node 版本")
print(" 当前命令: %s" % run_cmd)
print(" 建议命令: %s/node server.js" % node_path)
# 2. 停止项目（如果正在运行）
print(" 停止项目（如果正在运行）...")
stop_node_project(panel_url, api_key, pm2_name)
import time
time.sleep(2) # 等待停止完成
# 3. 重启项目
print(" 启动项目...")
ok = restart_node_project(panel_url, api_key, pm2_name)
if not ok:
# 如果重启失败，尝试直接启动
print(" 重启失败，尝试直接启动...")
ok = start_node_project(panel_url, api_key, pm2_name)
if not ok:
print(" 提示: 若 Node 接口不可用，请在宝塔面板【Node 项目】中手动重启 %s" % pm2_name)
print(" 项目路径: %s" % project_path)
print(" 启动命令: PORT=%d %s/node server.js" % (port, node_path))
print(" 端口: %d" % port)
print(" Node 版本: %s" % cfg.get("node_version", "v22.14.0"))
return ok
# ==================== 主函数 ====================
def main():
parser = argparse.ArgumentParser(
description="Soul 创业派对 - 本地打包 + SSH 上传 + 宝塔 API 部署",
formatter_class=argparse.RawDescriptionHelpFormatter,
epilog=__doc__
)
parser.add_argument("--no-build", action="store_true", help="跳过本地构建")
parser.add_argument("--no-upload", action="store_true", help="跳过 SSH 上传")
parser.add_argument("--no-api", action="store_true", help="上传后不调宝塔 API 重启")
args = parser.parse_args()
# 获取项目根目录
script_dir = os.path.dirname(os.path.abspath(__file__))
root = os.path.dirname(script_dir)
cfg = get_cfg()
print("=" * 60)
print(" Soul 创业派对 - 一键部署脚本")
print("=" * 60)
print(" 服务器: %s@%s" % (cfg["user"], cfg["host"]))
print(" 项目路径: %s" % cfg["project_path"])
print(" PM2 名称: %s" % cfg["pm2_name"])
print(" 站点地址: %s" % cfg["site_url"])
print(" 端口: %s" % cfg.get("port", 30006))
print(" Node 版本: %s" % cfg.get("node_version", "v22.14.0"))
print(" Node 路径: %s" % cfg.get("node_path", "/www/server/nodejs/v22.14.0/bin"))
print("=" * 60)
print("")
# 检查 Node 环境（可选，如果不需要可以跳过）
if not args.no_upload:
check_node_environments(cfg)
print("")
# 步骤 1: 本地构建
if not args.no_build:
if not run_build(root):
return 1
else:
standalone = os.path.join(root, ".next", "standalone", "server.js")
if not os.path.isfile(standalone):
print("[错误] 跳过构建但未找到 .next/standalone/server.js")
return 1
print("[跳过] 本地构建")
# 步骤 2: 打包
tarball_path = pack_standalone(root)
if not tarball_path:
return 1
# 步骤 3: SSH 上传并解压
if not args.no_upload:
if not upload_and_extract(cfg, tarball_path):
return 1
# 清理本地压缩包
try:
os.remove(tarball_path)
except Exception:
pass
else:
print("[跳过] SSH 上传")
print(" 压缩包位置: %s" % tarball_path)
# 步骤 4: 宝塔 API 重启
if not args.no_api and not args.no_upload:
deploy_via_baota_api(cfg)
elif args.no_api:
print("[跳过] 宝塔 API 重启")
print("")
print("=" * 60)
print(" 部署完成！")
print(" 前台: %s" % cfg["site_url"])
print(" 后台: %s/admin" % cfg["site_url"])
print("=" * 60)
return 0
if __name__ == "__main__":
sys.exit(main())

922
scripts/devlop.py
View File

File diff suppressed because it is too large Load Diff