soul-yongping/soul-api/devlop.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
soulApp (soul-api) Go 项目一键部署到宝塔
- 本地交叉编译 Linux 二进制
- 上传到 /www/wwwroot/自营/soul-api
- 重启：优先宝塔 API（需配置），否则 SSH 下 setsid nohup 启动

宝塔 API 重启（可选）：在环境变量或 .env 中设置
  BT_PANEL_URL   = https://你的面板地址:9988
  BT_API_KEY     = 面板 设置 -> API 接口 中的密钥
  BT_GO_PROJECT_NAME = soulApi  （与宝塔 Go 项目列表里名称一致）
并安装 requests: pip install requests
"""

from __future__ import print_function

import hashlib
import os
import sys
import tempfile
import argparse
import subprocess
import shutil
import tarfile
import time

try:
    import paramiko
except ImportError:
    print("错误: 请先安装 paramiko")
    print("  pip install paramiko")
    sys.exit(1)

try:
    import requests
    try:
        import urllib3
        urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)
    except Exception:
        pass
except ImportError:
    requests = None

# ==================== 配置 ====================

DEPLOY_PROJECT_PATH = "/www/wwwroot/自营/soul-api"
DEFAULT_SSH_PORT = int(os.environ.get("DEPLOY_SSH_PORT", "22022"))


# 宝塔 API 密钥（写死，用于部署后重启 Go 项目）
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 {
        "host": host,
        "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", DEPLOY_PROJECT_PATH),
        "bt_panel_url": bt_url,
        "bt_api_key": os.environ.get("BT_API_KEY", BT_API_KEY_DEFAULT),
        "bt_go_project_name": os.environ.get("BT_GO_PROJECT_NAME", "soulApi"),
    }


# ==================== 本地构建 ====================


def run_build(root):
    """交叉编译 Go 二进制（Linux amd64）"""
    print("[1/4] 本地交叉编译 Go 二进制 ...")
    env = os.environ.copy()
    env["GOOS"] = "linux"
    env["GOARCH"] = "amd64"
    env["CGO_ENABLED"] = "0"
    # 必须 shell=False，否则 Windows 下 -ldflags 等参数会被当成包路径导致 "malformed import path"
    cmd = ["go", "build", "-o", "soul-api", "./cmd/server"]
    try:
        r = subprocess.run(
            cmd,
            cwd=root,
            env=env,
            shell=False,
            timeout=120,
            capture_output=True,
            text=True,
            encoding="utf-8",
            errors="replace",
        )
        if r.returncode != 0:
            print("  [失败] go build 失败，退出码:", r.returncode)
            if r.stderr:
                for line in (r.stderr or "").strip().split("\n")[-10:]:
                    print("    " + line)
            return None
        out_path = os.path.join(root, "soul-api")
        if not os.path.isfile(out_path):
            print("  [失败] 未找到编译产物 soul-api")
            return None
        print("  [成功] 编译完成: %s (%.2f MB)" % (out_path, os.path.getsize(out_path) / 1024 / 1024))
        return out_path
    except subprocess.TimeoutExpired:
        print("  [失败] 编译超时")
        return None
    except FileNotFoundError:
        print("  [失败] 未找到 go 命令，请安装 Go")
        return None
    except Exception as e:
        print("  [失败] 编译异常:", str(e))
        return None


# ==================== 打包 ====================

DEPLOY_PORT = 8080


def set_env_port(env_path, port=DEPLOY_PORT):
    """将 .env 文件中的 PORT 设为指定值（用于部署包）"""
    if not os.path.isfile(env_path):
        return
    with open(env_path, "r", encoding="utf-8", errors="replace") as f:
        lines = f.readlines()
    found = False
    new_lines = []
    for line in lines:
        s = line.strip()
        if "=" in s and s.split("=", 1)[0].strip() == "PORT":
            new_lines.append("PORT=%s\n" % port)
            found = True
        else:
            new_lines.append(line)
    if not found:
        new_lines.append("PORT=%s\n" % port)
    with open(env_path, "w", encoding="utf-8", newline="\n") as f:
        f.writelines(new_lines)


def set_env_mini_program_state(env_path, state):
    """将 .env 中的 WECHAT_MINI_PROGRAM_STATE 设为 developer/formal（打包前按环境覆盖）"""
    if not os.path.isfile(env_path):
        return
    key = "WECHAT_MINI_PROGRAM_STATE"
    with open(env_path, "r", encoding="utf-8", errors="replace") as f:
        lines = f.readlines()
    found = False
    new_lines = []
    for line in lines:
        s = line.strip()
        if "=" in s and s.split("=", 1)[0].strip() == key:
            new_lines.append("%s=%s\n" % (key, state))
            found = True
        else:
            new_lines.append(line)
    if not found:
        new_lines.append("%s=%s\n" % (key, state))
    with open(env_path, "w", encoding="utf-8", newline="\n") as f:
        f.writelines(new_lines)


def pack_deploy(root, binary_path, include_env=True):
    """打包二进制和 .env 为 tar.gz"""
    print("[2/4] 打包部署文件 ...")
    staging = tempfile.mkdtemp(prefix="soul_api_deploy_")
    try:
        shutil.copy2(binary_path, os.path.join(staging, "soul-api"))
        env_src = os.path.join(root, ".env")
        staging_env = os.path.join(staging, ".env")
        if include_env and os.path.isfile(env_src):
            shutil.copy2(env_src, staging_env)
            print("  [已包含] .env")
        else:
            env_example = os.path.join(root, ".env.example")
            if os.path.isfile(env_example):
                shutil.copy2(env_example, staging_env)
                print("  [已包含] .env.example -> .env (请服务器上检查配置)")
        if os.path.isfile(staging_env):
            set_env_port(staging_env, DEPLOY_PORT)
            set_env_mini_program_state(staging_env, "formal")
            print("  [已设置] PORT=%s（部署用）, WECHAT_MINI_PROGRAM_STATE=formal（正式环境）" % DEPLOY_PORT)
        tarball = os.path.join(tempfile.gettempdir(), "soul_api_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)


# ==================== 宝塔 API 重启 ====================


def restart_via_bt_api(cfg):
    """通过宝塔 API 重启 Go 项目（需配置 BT_PANEL_URL、BT_API_KEY、BT_GO_PROJECT_NAME）"""
    url = cfg.get("bt_panel_url") or ""
    key = cfg.get("bt_api_key") or ""
    name = cfg.get("bt_go_project_name", "soulApi")
    if not url or not key:
        return False
    if not requests:
        print("  [提示] 未安装 requests，无法使用宝塔 API，将用 SSH 重启。pip install requests")
        return False
    try:
        req_time = int(time.time())
        sk_md5 = hashlib.md5(key.encode()).hexdigest()
        req_token = hashlib.md5(("%s%s" % (req_time, sk_md5)).encode()).hexdigest()
        # 宝塔 Go 项目插件：先停止再启动，接口以实际面板版本为准
        base = url.rstrip("/")
        params = {"request_time": req_time, "request_token": req_token}
        # 常见形式：/plugin?name=go_project，POST 带 action、project_name
        for action in ("stop_go_project", "start_go_project"):
            data = dict(params)
            data["action"] = action
            data["project_name"] = name
            r = requests.post(
                base + "/plugin?name=go_project",
                data=data,
                timeout=15,
                verify=False,
            )
            if r.status_code != 200:
                continue
            j = r.json() if r.headers.get("content-type", "").startswith("application/json") else {}
            if action == "stop_go_project":
                time.sleep(2)
            if j.get("status") is False and j.get("msg"):
                print("  [宝塔API] %s: %s" % (action, j.get("msg", "")))
        # 再调一次 start 确保启动
        data = dict(params)
        data["action"] = "start_go_project"
        data["project_name"] = name
        r = requests.post(base + "/plugin?name=go_project", data=data, timeout=15, verify=False)
        if r.status_code == 200:
            j = r.json() if r.headers.get("content-type", "").startswith("application/json") else {}
            if j.get("status") is True:
                print("  [成功] 已通过宝塔 API 重启 Go 项目: %s" % name)
                return True
        return False
    except Exception as e:
        print("  [宝塔API 失败] %s" % str(e))
        return False


# ==================== SSH 上传 ====================


def upload_and_extract(cfg, tarball_path, no_restart=False, restart_method="auto"):
    """上传 tar.gz 到服务器并解压、重启"""
    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"], port=DEFAULT_SSH_PORT,
                username=cfg["user"], key_filename=cfg["ssh_key"],
                timeout=15,
            )
        else:
            client.connect(
                cfg["host"], port=DEFAULT_SSH_PORT,
                username=cfg["user"], password=cfg["password"],
                timeout=15,
            )
        sftp = client.open_sftp()
        remote_tar = "/tmp/soul_api_deploy.tar.gz"
        project_path = cfg["project_path"]
        sftp.put(tarball_path, remote_tar)
        sftp.close()

        cmd = (
            "mkdir -p %s && cd %s && tar -xzf %s && "
            "chmod +x soul-api && rm -f %s && echo OK"
        ) % (project_path, project_path, remote_tar, remote_tar)
        stdin, stdout, stderr = client.exec_command(cmd, timeout=60)
        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)

        if not no_restart:
            print("[4/4] 重启 soulApp 服务 ...")
            ok = False
            if restart_method in ("auto", "btapi") and (cfg.get("bt_panel_url") and cfg.get("bt_api_key")):
                ok = restart_via_bt_api(cfg)
            if not ok and restart_method in ("auto", "ssh"):
                # SSH：只杀「工作目录为本项目」的 soul-api，避免误杀其他 Go 项目
                restart_cmd = (
                    "cd %s && T=$(readlink -f .) && for p in $(pgrep -f soul-api 2>/dev/null); do "
                    "[ \"$(readlink -f /proc/$p/cwd 2>/dev/null)\" = \"$T\" ] && kill $p 2>/dev/null; done; "
                    "sleep 2; setsid nohup ./soul-api >> soul-api.log 2>&1 </dev/null & "
                    "sleep 3; T=$(readlink -f .) && for p in $(pgrep -f soul-api 2>/dev/null); do "
                    "[ \"$(readlink -f /proc/$p/cwd 2>/dev/null)\" = \"$T\" ] && echo RESTART_OK && exit 0; done; echo RESTART_FAIL"
                ) % project_path
                stdin, stdout, stderr = client.exec_command(restart_cmd, timeout=20)
                out = stdout.read().decode("utf-8", errors="replace").strip()
                err = (stderr.read().decode("utf-8", errors="replace") or "").strip()
                if err:
                    print("  [stderr] %s" % err[:200])
                ok = "RESTART_OK" in out
                if ok:
                    print("  [成功] soulApp 已通过 SSH 重启")
                else:
                    print("  [警告] SSH 重启状态未知，请到宝塔 Go 项目里手动点击启动，或执行: cd %s && ./soul-api" % project_path)
        else:
            print("[4/4] 跳过重启 (--no-restart)")

        return True
    except Exception as e:
        print("  [失败] SSH 错误:", str(e))
        return False
    finally:
        client.close()


# ==================== 主函数 ====================


def main():
    parser = argparse.ArgumentParser(
        description="soulApp (soul-api) Go 项目一键部署到宝塔",
        formatter_class=argparse.RawDescriptionHelpFormatter,
    )
    parser.add_argument("--no-build", action="store_true", help="跳过本地编译（使用已有 soul-api 二进制）")
    parser.add_argument("--no-env", action="store_true", help="不打包 .env（保留服务器现有 .env）")
    parser.add_argument("--no-restart", action="store_true", help="上传后不重启服务")
    parser.add_argument(
        "--restart-method",
        choices=("auto", "btapi", "ssh"),
        default="auto",
        help="重启方式: auto=先试宝塔API再SSH, btapi=仅宝塔API, ssh=仅SSH (默认 auto)",
    )
    args = parser.parse_args()

    script_dir = os.path.dirname(os.path.abspath(__file__))
    root = script_dir

    cfg = get_cfg()
    print("=" * 60)
    print("  soulApp 一键部署到宝塔")
    print("=" * 60)
    print("  服务器: %s@%s:%s" % (cfg["user"], cfg["host"], DEFAULT_SSH_PORT))
    print("  目标目录: %s" % cfg["project_path"])
    print("=" * 60)

    binary_path = os.path.join(root, "soul-api")
    if not args.no_build:
        p = run_build(root)
        if not p:
            return 1
    else:
        if not os.path.isfile(binary_path):
            print("[错误] 未找到 soul-api 二进制，请先编译或去掉 --no-build")
            return 1
        print("[1/4] 跳过编译，使用现有 soul-api")

    tarball = pack_deploy(root, binary_path, include_env=not args.no_env)
    if not tarball:
        return 1

    if not upload_and_extract(cfg, tarball, no_restart=args.no_restart, restart_method=args.restart_method):
        return 1

    try:
        os.remove(tarball)
    except Exception:
        pass

    print("")
    print("  部署完成！目录: %s" % cfg["project_path"])
    return 0


if __name__ == "__main__":
    sys.exit(main())