happy-life-star/life-script/deploy.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
life-script 部署脚本
功能：项目构建、文件传输、原子切换、历史版本管理、回滚支持
使用系统自带的ssh/scp命令，无需额外依赖
"""

import os
import sys
import subprocess
import shutil
from datetime import datetime
from pathlib import Path

# 服务器配置
SERVER_IP = "101.200.208.45"
USERNAME = "root"

# 部署配置
APP_NAME = "life-script"
DEPLOY_BASE = "/data/www/course-web-deploy"
RELEASES_DIR = f"{DEPLOY_BASE}/releases"
LINK_PATH = "/data/www/course-of-life"
MAX_RELEASES = 5

# 本地路径
SCRIPT_DIR = Path(__file__).parent.absolute()
DIST_DIR = SCRIPT_DIR / "dist"


class Colors:
    """终端颜色"""
    GREEN = '\033[32m'
    RED = '\033[31m'
    YELLOW = '\033[33m'
    RESET = '\033[0m'


def log_info(msg):
    """打印信息日志"""
    print(f"{Colors.GREEN}[INFO]{Colors.RESET} {msg}")


def log_error(msg):
    """打印错误日志"""
    print(f"{Colors.RED}[ERROR]{Colors.RESET} {msg}")


def log_warn(msg):
    """打印警告日志"""
    print(f"{Colors.YELLOW}[WARN]{Colors.RESET} {msg}")


def run_command(cmd, cwd=None, shell=True, capture=True):
    """执行本地命令"""
    try:
        if capture:
            result = subprocess.run(
                cmd,
                cwd=cwd,
                shell=shell,
                capture_output=True,
                text=True
            )
            return result.returncode == 0, result.stdout, result.stderr
        else:
            result = subprocess.run(cmd, cwd=cwd, shell=shell)
            return result.returncode == 0, "", ""
    except Exception as e:
        return False, "", str(e)


def exec_ssh_cmd(cmd):
    """通过SSH执行远程命令"""
    ssh_cmd = f'ssh {USERNAME}@{SERVER_IP} "{cmd}"'
    return run_command(ssh_cmd)


def scp_upload(local_path, remote_path, recursive=False):
    """通过SCP上传文件或目录"""
    r_flag = "-r" if recursive else ""
    scp_cmd = f'scp {r_flag} "{local_path}" {USERNAME}@{SERVER_IP}:{remote_path}'
    success, stdout, stderr = run_command(scp_cmd)
    if not success:
        log_error(f"SCP上传失败: {stderr}")
    return success


def check_env():
    """检查本地环境"""
    log_info("检查本地环境...")
    
    # 检查npm
    success, _, _ = run_command("npm --version")
    if not success:
        log_error("未找到npm命令，请先安装Node.js")
        sys.exit(1)
    
    log_info("环境检查通过")


def build_project():
    """构建项目"""
    log_info("开始构建项目...")
    
    # 切换到项目目录
    os.chdir(SCRIPT_DIR)
    
    # 清理旧构建
    if DIST_DIR.exists():
        shutil.rmtree(DIST_DIR)
        log_info("已清理旧的dist目录")
    
    # 执行构建（不捕获输出，直接显示）
    log_info("执行: npm run build")
    success, _, _ = run_command("npm run build", capture=False)
    if not success:
        log_error("项目构建失败")
        sys.exit(1)
    
    log_info("项目构建成功")
    
    # 检查dist目录
    if not DIST_DIR.exists():
        log_error("dist目录不存在")
        sys.exit(1)


def deploy():
    """部署到服务器"""
    timestamp = datetime.now().strftime("%Y%m%d%H%M%S")
    release_path = f"{RELEASES_DIR}/{timestamp}"
    
    log_info(f"准备部署版本: {timestamp}")
    
    # 1. 创建远程目录
    log_info("创建远程目录...")
    exec_ssh_cmd(f"mkdir -p {release_path}")
    
    # 2. 上传文件
    log_info("上传文件到服务器...")
    for item in DIST_DIR.iterdir():
        if item.is_file():
            if not scp_upload(item, f"{release_path}/"):
                log_error("文件上传失败")
                sys.exit(1)
        else:
            if not scp_upload(item, f"{release_path}/", recursive=True):
                log_error("目录上传失败")
                sys.exit(1)
    log_info("文件上传成功")
    
    # 3. 设置权限
    log_info("设置文件权限...")
    exec_ssh_cmd(f"chmod -R 755 {release_path}")
    
    # 4. 原子切换软链接
    log_info("切换服务版本...")
    # 检查目标路径是否为普通目录
    exec_ssh_cmd(f"if [ -d '{LINK_PATH}' ] && [ ! -L '{LINK_PATH}' ]; then mv '{LINK_PATH}' '{LINK_PATH}_backup_$(date +%s)'; fi")
    # 创建软链接
    exec_ssh_cmd(f"ln -snf '{release_path}' '{LINK_PATH}'")
    
    log_info(f"部署完成！当前版本指向: {release_path}")
    
    # 5. 清理旧版本
    clean_old_releases()


def clean_old_releases():
    """清理旧版本，只保留最近的N个"""
    log_info(f"清理旧版本（保留最近{MAX_RELEASES}个）...")
    clean_cmd = f"cd {RELEASES_DIR} && ls -t | tail -n +{MAX_RELEASES + 1} | xargs -I {{}} rm -rf {{}}"
    exec_ssh_cmd(clean_cmd)


def rollback():
    """回滚到上一个版本"""
    log_info("开始回滚操作...")
    
    # 获取当前指向的版本
    success, current_link, _ = exec_ssh_cmd(f"readlink {LINK_PATH}")
    log_info(f"当前版本: {current_link}")
    
    # 获取上一个版本目录
    success, prev_version, _ = exec_ssh_cmd(
        f"ls -dt {RELEASES_DIR}/* | grep -v '{current_link}' | head -n 1"
    )
    
    if not prev_version:
        log_error("没有找到可回滚的历史版本")
        sys.exit(1)
    
    log_info(f"回滚目标版本: {prev_version}")
    exec_ssh_cmd(f"ln -snf {prev_version} {LINK_PATH}")
    log_info("回滚成功！")


def print_usage():
    """打印使用说明"""
    print("""
用法: python deploy.py [命令]

命令:
  deploy   - 构建并部署到服务器（默认）
  rollback - 回滚到上一个版本

示例:
  python deploy.py          # 部署
  python deploy.py rollback # 回滚
""")


def main():
    """主函数"""
    command = sys.argv[1] if len(sys.argv) > 1 else "deploy"
    
    if command == "rollback":
        rollback()
    elif command == "deploy":
        check_env()
        build_project()
        deploy()
    else:
        print_usage()
        sys.exit(1)


if __name__ == "__main__":
    main()