@@ -0,0 +1,338 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
soul-api Go 项目一键部署到宝塔，重启的是宝塔里的 soulDev 项目
- 本地交叉编译 Linux 二进制
- 上传到 /www/wwwroot/自营/soul-dev
- 重启 soulDev： （ ） ，
宝塔 API 重启（可选）：在环境变量或 .env 中设置
BT_PANEL_URL = https://你的面板地址:9988
BT_API_KEY = 面板 设置 -> API 接口 中的密钥
BT_GO_PROJECT_NAME = soulDev （与宝塔 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-dev "
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 " , " soulDev " ) ,
}
# ==================== 本地构建 ====================
def run_build ( root ) :
""" 交叉编译 Go 二进制（ ） """
print ( " [1/4] 本地交叉编译 Go 二进制 ... " )
env = os . environ . copy ( )
env [ " GOOS " ] = " linux "
env [ " GOARCH " ] = " amd64 "
env [ " CGO_ENABLED " ] = " 0 "
# 必须 shell=False，
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
# ==================== 打包 ====================
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 " )
if include_env and os . path . isfile ( env_src ) :
shutil . copy2 ( env_src , os . path . join ( staging , " .env " ) )
print ( " [已包含] .env " )
else :
env_example = os . path . join ( root , " .env.example " )
if os . path . isfile ( env_example ) :
shutil . copy2 ( env_example , os . path . join ( staging , " .env " ) )
print ( " [已包含] .env.example -> .env (请服务器上检查配置) " )
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 " , " soulDev " )
if not url or not key :
return False
if not requests :
print ( " [提示] 未安装 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，
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] 重启 soulDev 服务 ... " )
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：
restart_cmd = (
" cd %s && pkill -f ' ./soul-api ' 2>/dev/null; sleep 2; "
" setsid nohup ./soul-api >> soul-api.log 2>&1 </dev/null & "
" sleep 3; pgrep -f ' ./soul-api ' >/dev/null && echo RESTART_OK || 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 ( " [成功] soulDev 已通过 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 = " soul-api 一键部署到宝塔，重启 soulDev 项目 " ,
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（ ） " )
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 ( " soul-api 部署到宝塔，重启 soulDev " )
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 ( ) )
乘风