Files
happy-life-star/deploy.py
T
peanut e313086d9b fix(deploy): Windows 平台 curl 验证使用 NUL 替代 /dev/null
Windows cmd.exe 不支持 /dev/null(那是 Unix 概念),
导致 curl 写入失败返回 exit code 23,验证误报为'访问异常'。
改为根据 sys.platform 动态选择 null 设备。
2026-07-19 13:25:01 +08:00

617 lines
22 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
情绪博物馆 - 统一部署脚本
合并 deploy-all.sh / deploy-to-prod.sh / deploy-domain.sh
使用方法:
python deploy.py # 部署所有服务
python deploy.py backend # 仅部署后端
python deploy.py frontend # 仅部署前端
python deploy.py admin # 仅部署管理后台
python deploy.py life-script # 仅部署 Life-Script
python deploy.py ssl # 仅申请 SSL 证书
python deploy.py nginx # 仅部署 Nginx 配置
python deploy.py verify # 仅验证部署结果
python deploy.py all # 部署所有服务(同无参数)
Author: Peanut
Created: 2026-05-17
Purpose: 统一部署所有服务到生产服务器,替代原有的多个 shell 脚本
"""
import io
import os
import re
import sys
import time
import subprocess
from pathlib import Path
# 强制 stdout/stderr 使用 UTF-8 编码,避免 Windows GBK 编码错误
if hasattr(sys.stdout, 'buffer'):
sys.stdout = io.TextIOWrapper(sys.stdout.buffer, encoding='utf-8', errors='replace', line_buffering=True)
sys.stderr = io.TextIOWrapper(sys.stderr.buffer, encoding='utf-8', errors='replace', line_buffering=True)
# 当前 Python 可执行文件(Windows 用 python.exeLinux/Mac 用 python3
PYTHON = sys.executable
# ============================================================================
# 环境变量修复(Windows 专用)
# ============================================================================
def _build_full_path():
"""构造包含所有关键目录的完整 PATH 字符串。
Windows cmd.exe 启动时会用自己的"标准 PATH"(只有 2 段用户 PATH),
完全忽略父进程传入的 PATH 变量。解决办法是在每个 cmd.exe 命令前
用 `set "PATH=..."` 显式设置。这个函数返回要设置的完整 PATH。
"""
if sys.platform != 'win32':
return None
parts = [
r'C:\Windows\system32',
r'C:\Windows',
r'C:\Windows\System32\Wbem',
r'C:\Windows\System32\OpenSSH',
]
# 从已知环境变量自动推断关键目录(不硬编码)
if os.environ.get('MAVEN_HOME'):
parts.append(os.environ['MAVEN_HOME'] + r'\bin')
if os.environ.get('JAVA_HOME'):
parts.append(os.environ['JAVA_HOME'] + r'\bin')
# Node.js 常见安装位置自动检测
for d in [r'C:\Program Files\nodejs', r'F:\Program Files\nodejs',
r'C:\Program Files (x86)\nodejs', r'F:\Programs\nvm']:
if Path(d).exists():
parts.append(d)
# 保留用户已有的 PATH(不丢失用户自定义工具)
user_path = os.environ.get('PATH', '')
for p in user_path.split(';'):
if p and p not in parts:
parts.append(p)
return ';'.join(parts)
def _wrap_cmd_for_windows(cmd):
"""为 Windows cmd.exe 命令包装 set PATH 头部,解决 PATH 截断问题。
原因:cmd.exe 启动后 PATH 只有 2 段(pnpm/npm),找不到 mvn/node/npm。
修复:在每个命令前用 `set "PATH=完整路径" &&` 显式设置 PATH。
"""
if sys.platform != 'win32':
return cmd
full_path = _build_full_path()
return f'set "PATH={full_path}" && {cmd}'
def _ensure_env():
"""修复 Windows 子进程编码问题 + 准备 PATH 工具。
Windows cmd.exe 启动时强制用 GBK 编码输出(即使父进程是 UTF-8),
设置 PYTHONIOENCODING/PYTHONUTF8 强制 Python 子进程使用 UTF-8。
实际 PATH 修复在 _wrap_cmd_for_windows() 中按需包装每个命令。
"""
if sys.platform != 'win32':
return
# 强制 Python 子进程使用 UTF-8 编码
os.environ.setdefault('PYTHONIOENCODING', 'utf-8')
os.environ.setdefault('PYTHONUTF8', '1')
os.environ.setdefault('PYTHONLEGACYWINDOWSSTDIO', '0')
# ============================================================================
# 配置
# ============================================================================
SERVER_IP = "101.200.208.45"
USERNAME = "root"
DOMAIN = "lifescript.happylifeos.com"
# 项目根目录(脚本所在目录)
PROJECT_DIR = Path(__file__).parent.absolute()
SSH_OPTS = f"-o BatchMode=yes -o ConnectTimeout=10 -o StrictHostKeyChecking=no"
# ============================================================================
# 日志
# ============================================================================
class Colors:
RED = '\033[0;31m'
GREEN = '\033[0;32m'
YELLOW = '\033[1;33m'
BLUE = '\033[0;34m'
NC = '\033[0m'
def log_info(msg):
print(f"{Colors.GREEN}[INFO]{Colors.NC} {msg}")
def log_warn(msg):
print(f"{Colors.YELLOW}[WARN]{Colors.NC} {msg}")
def log_error(msg):
print(f"{Colors.RED}[ERROR]{Colors.NC} {msg}")
def log_section(msg):
print(f"\n{Colors.BLUE}{'=' * 60}{Colors.NC}")
print(f"{Colors.BLUE}{msg}{Colors.NC}")
print(f"{Colors.BLUE}{'=' * 60}{Colors.NC}\n")
# ============================================================================
# 工具函数
# ============================================================================
def run_command(cmd, cwd=None, timeout=120, capture=True):
"""执行本地命令(自动处理 Windows cmd.exe PATH 截断)"""
# 设置 PYTHONUNBUFFERED=1 确保 Python 子进程不缓冲 stdout
env = os.environ.copy()
env['PYTHONUNBUFFERED'] = '1'
# Windows 上包装 set PATH 前缀,解决 cmd.exe PATH 截断问题
effective_cmd = _wrap_cmd_for_windows(cmd)
try:
if capture:
result = subprocess.run(
effective_cmd, shell=True, cwd=cwd, env=env,
capture_output=True, text=True, encoding='utf-8', errors='replace',
timeout=timeout
)
return result.returncode == 0, result.stdout.strip(), result.stderr.strip()
else:
result = subprocess.run(
effective_cmd, shell=True, cwd=cwd, env=env, timeout=timeout
)
return result.returncode == 0, "", ""
except subprocess.TimeoutExpired:
return False, "", f"命令执行超时 ({timeout}s): {cmd}"
except Exception as e:
return False, "", str(e)
def run_ssh_args(args, timeout=30, capture=True):
"""执行 ssh/scp 命令(使用列表参数,避免 Windows cmd.exe 找不到 ssh"""
try:
if capture:
result = subprocess.run(
args, shell=False, cwd=PROJECT_DIR,
capture_output=True, text=True, encoding='utf-8', errors='replace',
timeout=timeout
)
return result.returncode == 0, result.stdout.strip(), result.stderr.strip()
else:
result = subprocess.run(
args, shell=False, cwd=PROJECT_DIR, timeout=timeout
)
return result.returncode == 0, "", ""
except subprocess.TimeoutExpired:
return False, "", f"命令执行超时 ({timeout}s): {' '.join(args)}"
except Exception as e:
return False, "", str(e)
def ssh_command(cmd, timeout=30):
"""在远程服务器执行命令"""
return run_ssh_args([
"ssh", "-o", "BatchMode=yes", "-o", "ConnectTimeout=10",
"-o", "StrictHostKeyChecking=no",
f"{USERNAME}@{SERVER_IP}", cmd
], timeout=timeout)
def scp_file(local, remote, timeout=120):
"""上传文件到远程服务器"""
return run_ssh_args([
"scp", "-o", "BatchMode=yes", "-o", "ConnectTimeout=10",
"-o", "StrictHostKeyChecking=no",
local, f"{USERNAME}@{SERVER_IP}:{remote}"
], timeout=timeout)
def check_ssh():
"""检查 SSH 连接"""
log_info("检查 SSH 连接...")
success, stdout, stderr = ssh_command("echo ok", timeout=15)
if not success:
log_error(f"SSH 连接失败,请先配置免密登录: ssh {USERNAME}@{SERVER_IP}")
if stderr:
log_error(f"错误详情: {stderr}")
sys.exit(1)
log_info("SSH 连接正常")
# ============================================================================
# Nginx 站点启用(防御性诊断 + 清理)
# ============================================================================
# 变量安全校验正则
_SAFE_DOMAIN = re.compile(r'^[a-zA-Z0-9.-]+$')
_SAFE_REMOTE_CONF = re.compile(r'^[a-zA-Z0-9./_-]+$')
def _validate_nginx_inputs(remote_conf: str, domain: str) -> bool:
"""校验 domain 和 remote_conf 防止 shell 注入和路径遍历。
Returns: True 合法;False 非法(已 log_error
"""
if not domain or not _SAFE_DOMAIN.match(domain):
log_error(f"非法 domain 名称: {domain!r}")
return False
if domain.startswith('-') or domain.endswith('-') \
or domain.startswith('.') or domain.endswith('.') \
or '..' in domain:
log_error(f"非法 domain 结构: {domain!r} (首尾连字符/点 或 连续点非法)")
return False
if not remote_conf or not _SAFE_REMOTE_CONF.match(remote_conf):
log_error(f"非法 remote_conf 路径: {remote_conf!r}")
return False
if '//' in remote_conf or '..' in remote_conf:
log_error(f"remote_conf 包含可疑路径片段: {remote_conf!r}")
return False
return True
def enable_nginx_site(remote_conf: str, domain: str) -> bool:
"""幂等地启用 nginx 站点(sites-enabled symlink)。
流程:
1. 诊断目标路径当前状态(symlink/dir/file/missing
2. 清理异常状态(目录/文件/断链);失败短路
3. 二次检查 + 创建 symlink
4. 移除 default 站点(失败仅 warning
5. 验证 symlink + default 状态
Returns: True 成功;False 失败(stderr 已打印)
"""
if not _validate_nginx_inputs(remote_conf, domain):
return False
target = f"/etc/nginx/sites-enabled/{domain}.conf"
log_info(f"启用 nginx 站点: {target} -> {remote_conf}")
# ===== 步骤 1: 诊断目标路径状态 =====
log_info("诊断步骤: 检查目标路径...")
diagnose_cmd = (
f'test -e {target} && '
f'(test -L {target} && echo "symlink" || '
f'(test -d {target} && echo "dir" || echo "file")) || '
f'echo "missing"'
)
ok, status, err = ssh_command(diagnose_cmd, timeout=30)
if not ok:
log_error(f"诊断步骤失败: {err}")
return False
status = status.strip()
log_info(f"诊断结果: {status}")
# ===== 步骤 2: 清理异常状态 =====
log_info("清理步骤: 处理异常状态...")
cleanup_cmd = (
f'case "{status}" in '
f'symlink) unlink {target} ;; '
f'dir) rm -rf {target} ;; '
f'file) rm -f {target} ;; '
f'missing) ;; '
f'esac'
)
ok, _, err = ssh_command(cleanup_cmd, timeout=30)
if not ok:
log_error(f"清理步骤失败: {err}")
log_error("上下文: 目标 = " + target)
log_error("提示: 权限可能不足(检查 /etc/nginx/sites-enabled 写权限)")
return False
log_info("清理步骤完成")
# ===== 步骤 3: 建链(两段式:先二次检查,再 ln -snf) =====
log_info("建链步骤: 创建 symlink...")
# 第一段:二次检查 target 不存在(防并发进程干扰)
guard_cmd = f'if [ -e {target} ]; then echo "二次检查失败:target 仍存在" >&2; exit 1; fi'
ok, _, err = ssh_command(guard_cmd, timeout=30)
if not ok:
log_error(f"建链步骤失败 - 二次检查: {err}")
log_error("提示: 并发进程干扰(其他进程在步骤 2 后、步骤 3 前创建了同名项)")
log_error("上下文: 目标 = " + target)
return False
# 第二段:执行 ln -snf(独立 SSH 调用,错误信息不会被二次检查误报)
link_cmd = f'ln -snf {remote_conf} {target}'
ok, _, err = ssh_command(link_cmd, timeout=30)
if not ok:
log_error(f"建链步骤失败 - ln 执行: {err}")
log_error("提示: 确认 scp_file 是否成功上传了 sites-available 下的 conf 文件")
log_error("上下文: 目标 = " + target)
return False
log_info("建链步骤完成")
# ===== 步骤 4: 清理 default 站点(失败仅 warning =====
log_info("清默认步骤: 移除 default 站点...")
ok, _, err = ssh_command("rm -f /etc/nginx/sites-enabled/default", timeout=30)
if not ok:
log_warn(f"清默认步骤失败(不阻塞): {err}")
log_warn("提示: default 文件可能仍存在,nginx 可能有端口冲突;步骤 5 验证会确认")
else:
log_info("清默认步骤完成")
# ===== 步骤 5: 验证结果 =====
log_info("验证步骤: 检查 symlink 和 default 状态...")
verify_cmd = (
f'readlink {target} && '
f'(test ! -e /etc/nginx/sites-enabled/default && echo "default:OK" || echo "default:STILL_EXISTS") && '
f'ls -la /etc/nginx/sites-enabled/'
)
ok, stdout, err = ssh_command(verify_cmd, timeout=30)
if not ok:
log_error(f"验证步骤失败: {err}")
log_error(f"实际输出: {stdout}")
return False
log_info("验证结果:")
for line in stdout.split('\n'):
if line.strip():
log_info(f" {line}")
log_info("启用站点配置完成")
return True
# ============================================================================
# Nginx 配置
# ============================================================================
def deploy_nginx():
"""部署 Nginx 配置"""
log_section("部署 Nginx 配置")
nginx_conf = PROJECT_DIR / "conf" / "emotion-museum.conf"
if not nginx_conf.exists():
log_error(f"Nginx 配置文件不存在: {nginx_conf}")
return False
log_info("上传 Nginx 配置文件...")
remote_conf = f"/etc/nginx/sites-available/{DOMAIN}.conf"
# 先创建远程目录
ssh_command("mkdir -p /etc/nginx/sites-available /etc/nginx/sites-enabled", timeout=15)
ok, _, err = scp_file(str(nginx_conf), remote_conf, timeout=60)
if not ok:
log_error(f"上传 Nginx 配置失败: {err}")
return False
log_info("启用站点配置...")
if not enable_nginx_site(remote_conf, DOMAIN):
log_error("启用站点失败(详见 enable_nginx_site 内部日志)")
return False
log_info("验证 Nginx 配置...")
ok, _, err = ssh_command("nginx -t", timeout=15)
if ok:
ssh_command("systemctl reload nginx", timeout=30)
log_info("Nginx 配置重载完成")
return True
else:
log_error(f"Nginx 配置验证失败: {err}")
return False
# ============================================================================
# 后端 - 调用 server/deploy.sh remote
# ============================================================================
def deploy_backend():
"""部署后端服务"""
log_section("部署后端服务")
deploy_script = PROJECT_DIR / "server" / "deploy.py"
if not deploy_script.exists():
log_error(f"后端部署脚本不存在: {deploy_script}")
return False
log_info("执行后端部署...")
ok, _, err = run_command(
f"{PYTHON} deploy.py remote",
cwd=str(PROJECT_DIR / "server"),
timeout=600,
capture=False
)
if ok:
log_info("后端部署完成")
return True
else:
log_error(f"后端部署失败: {err}")
return False
# ============================================================================
# 前端 - 调用 web/deploy.sh
# ============================================================================
def deploy_frontend():
"""部署用户前端"""
log_section("部署用户前端")
deploy_script = PROJECT_DIR / "web" / "deploy.py"
if not deploy_script.exists():
log_error(f"前端部署脚本不存在: {deploy_script}")
return False
log_info("执行前端部署...")
ok, _, err = run_command(
f"{PYTHON} deploy.py",
cwd=str(PROJECT_DIR / "web"),
timeout=600,
capture=False
)
if ok:
log_info("前端部署完成")
return True
else:
log_error(f"前端部署失败: {err}")
return False
# ============================================================================
# 管理后台 - 调用 web-admin/deploy.py
# ============================================================================
def deploy_admin():
"""部署管理后台"""
log_section("部署管理后台")
deploy_script = PROJECT_DIR / "web-admin" / "deploy.py"
if not deploy_script.exists():
log_error(f"管理后台部署脚本不存在: {deploy_script}")
return False
log_info("执行管理后台部署...")
ok, _, err = run_command(
f"{PYTHON} deploy.py",
cwd=str(PROJECT_DIR / "web-admin"),
timeout=600,
capture=False
)
if ok:
log_info("管理后台部署完成")
return True
else:
log_error(f"管理后台部署失败: {err}")
return False
# ============================================================================
# Life-Script - 调用 life-script/deploy.sh
# ============================================================================
def deploy_life_script():
"""部署 Life-Script"""
log_section("部署 Life-Script")
deploy_script = PROJECT_DIR / "life-script" / "deploy.py"
if not deploy_script.exists():
log_error(f"Life-Script 部署脚本不存在: {deploy_script}")
return False
log_info("执行 Life-Script 部署...")
ok, _, err = run_command(
f"{PYTHON} deploy.py",
cwd=str(PROJECT_DIR / "life-script"),
timeout=600,
capture=False
)
if ok:
log_info("Life-Script 部署完成")
return True
else:
log_error(f"Life-Script 部署失败: {err}")
return False
# ============================================================================
# 验证部署
# ============================================================================
def verify_deploy():
"""验证部署结果"""
log_section("验证部署")
log_info("检查 HTTPS 访问...")
endpoints = [
(f"https://{DOMAIN}/", "前端页面"),
(f"https://{DOMAIN}/emotion-museum-admin/", "管理后台"),
(f"https://{DOMAIN}/api/", "API 代理"),
(f"http://{DOMAIN}/", "HTTP 跳转"),
]
for url, label in endpoints:
null_device = 'NUL' if sys.platform == 'win32' else '/dev/null'
ok, stdout, _ = run_command(
f'curl -k -s -o {null_device} -w "HTTP %{{http_code}}" {url}',
timeout=30
)
if ok and stdout:
log_info(f" {label}: {stdout}")
else:
log_warn(f" {label}: 访问异常")
log_info("验证完成")
return True
# ============================================================================
# 主程序
# ============================================================================
def print_usage():
"""打印使用说明"""
print("情绪博物馆 - 统一部署脚本")
print()
print("使用方法:")
print(" python deploy.py # 部署所有服务")
print(" python deploy.py backend # 仅部署后端")
print(" python deploy.py frontend # 仅部署前端")
print(" python deploy.py admin # 仅部署管理后台")
print(" python deploy.py life-script # 仅部署 Life-Script")
print(" python deploy.py ssl # 仅申请 SSL 证书")
print(" python deploy.py nginx # 仅部署 Nginx 配置")
print(" python deploy.py verify # 仅验证部署结果")
print(" python deploy.py all # 部署所有服务(同无参数)")
def main():
# 修复 Windows cmd.exe 子进程环境块截断问题
_ensure_env()
deploy_type = sys.argv[1] if len(sys.argv) > 1 else "all"
if deploy_type in ("--help", "-h", "help"):
print_usage()
return
start_time = time.time()
log_section("情绪博物馆 - 统一部署")
log_info(f"部署类型: {deploy_type}")
log_info(f"部署时间: {time.strftime('%Y-%m-%d %H:%M:%S')}")
actions = {
"backend": lambda: check_ssh() or True and deploy_backend(),
"frontend": lambda: check_ssh() or True and deploy_frontend(),
"admin": lambda: check_ssh() or True and deploy_admin(),
"life-script": lambda: check_ssh() or True and deploy_life_script(),
"nginx": lambda: check_ssh() or True and deploy_nginx(),
"verify": lambda: verify_deploy(),
"all": None, # handled separately
}
if deploy_type == "all":
check_ssh()
deploy_nginx() or True
deploy_backend()
deploy_frontend()
deploy_admin()
deploy_life_script()
verify_deploy()
elif deploy_type in actions:
actions[deploy_type]()
else:
log_error(f"无效的部署类型: {deploy_type}")
print("使用方法: python deploy.py [backend|frontend|admin|life-script|nginx|verify|all]")
sys.exit(1)
duration = int(time.time() - start_time)
log_section(f"部署完成 (耗时 {duration}s)")
log_info(f"用户前端: https://{DOMAIN}/")
log_info(f"管理后台: https://{DOMAIN}/emotion-museum-admin/")
log_info(f"Life-Script: https://{DOMAIN}/life-script/")
log_info(f"API 地址: https://{DOMAIN}/api")
log_info(f"WebSocket: wss://{DOMAIN}/ws")
log_info(f"API 文档: https://{DOMAIN}/api/doc.html")
log_info(f"Swagger UI: https://{DOMAIN}/api/swagger-ui/index.html")
log_info(f"API Specs: https://{DOMAIN}/api/v3/api-docs")
if __name__ == "__main__":
main()