189 lines
6.8 KiB
Python
189 lines
6.8 KiB
Python
#!/usr/bin/env python3
|
||
# -*- coding: utf-8 -*-
|
||
"""
|
||
kr 宝塔：正式环境发布 + 宝塔面板 API 重启 Go 项目 + 线上冒烟验证。
|
||
|
||
说明（与现状一致）：
|
||
- soul-api 默认 **全程宝塔 API**：`/files?action=upload` 上传、`UnZip` 解压、`DeleteFile` 删包、`SiteStop/SiteStart` 重启（见 soul-api/master.py `--deploy-method btapi`，无需 SSH）。
|
||
- 若需旧版 SSH 上传：设置环境变量 `DEPLOY_METHOD=ssh` 或 `python master.py --deploy-method ssh`（需服务器允许密码/密钥）。
|
||
- soul-admin 静态资源仍通过 soul-admin/master.py（SSH），宝塔无统一「站点文件 API」封装。
|
||
- 环境变量与 soul-api/master.py 相同：DEPLOY_HOST、DEPLOY_PASSWORD、BT_PANEL_URL、BT_API_KEY、BT_GO_PROJECT_NAME 等。
|
||
|
||
用法：
|
||
python3 scripts/deploy_kr_btapi_verify.py # 部署 API+管理端 + 验证
|
||
python3 scripts/deploy_kr_btapi_verify.py --verify-only # 仅公网冒烟（不部署）
|
||
python3 scripts/deploy_kr_btapi_verify.py --skip-bt-ping # 跳过宝塔 API 探活
|
||
"""
|
||
|
||
from __future__ import print_function
|
||
|
||
import argparse
|
||
import hashlib
|
||
import os
|
||
import subprocess
|
||
import sys
|
||
import time
|
||
|
||
ROOT = os.path.abspath(os.path.join(os.path.dirname(__file__), ".."))
|
||
SOUL_API = os.path.join(ROOT, "soul-api")
|
||
SOUL_ADMIN = os.path.join(ROOT, "soul-admin")
|
||
|
||
|
||
def _load_api_cfg():
|
||
import importlib.util
|
||
|
||
path = os.path.join(SOUL_API, "master.py")
|
||
spec = importlib.util.spec_from_file_location("soul_api_master_mod", path)
|
||
mod = importlib.util.module_from_spec(spec)
|
||
spec.loader.exec_module(mod)
|
||
return mod.get_cfg()
|
||
|
||
|
||
def bt_panel_ping(cfg):
|
||
"""调用宝塔 /system?action=GetDiskInfo 校验 API 密钥与面板可达性。"""
|
||
try:
|
||
import requests
|
||
|
||
try:
|
||
import urllib3
|
||
|
||
urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)
|
||
except Exception:
|
||
pass
|
||
except ImportError:
|
||
print("[失败] 需要 requests：pip install requests")
|
||
return False
|
||
|
||
url = (cfg.get("bt_panel_url") or "").rstrip("/")
|
||
key = cfg.get("bt_api_key") or ""
|
||
if not url or not key:
|
||
print("[失败] 缺少 BT_PANEL_URL 或 BT_API_KEY")
|
||
return False
|
||
req_time = int(time.time())
|
||
sk_md5 = hashlib.md5(key.encode()).hexdigest()
|
||
req_token = hashlib.md5(("%s%s" % (req_time, sk_md5)).encode()).hexdigest()
|
||
try:
|
||
r = requests.post(
|
||
url + "/system?action=GetDiskInfo",
|
||
data={"request_time": req_time, "request_token": req_token},
|
||
timeout=20,
|
||
verify=False,
|
||
)
|
||
if r.status_code != 200:
|
||
print("[宝塔API 探活] HTTP %s" % r.status_code)
|
||
return False
|
||
j = r.json() if "application/json" in (r.headers.get("content-type") or "") else {}
|
||
# 部分面板返回 data 为列表或非标准结构，仅当明确 status=False 视为失败
|
||
if isinstance(j, dict) and j.get("status") is False:
|
||
print("[宝塔API 探活] %s" % (j.get("msg") or j))
|
||
return False
|
||
print("[宝塔API 探活] 成功（GetDiskInfo）", flush=True)
|
||
return True
|
||
except Exception as e:
|
||
print("[宝塔API 探活] 异常: %s" % e)
|
||
return False
|
||
|
||
|
||
def run_cmd(cwd, args, env=None):
|
||
print("", flush=True)
|
||
print(">>> %s" % " ".join(args), flush=True)
|
||
e = os.environ.copy()
|
||
if env:
|
||
e.update(env)
|
||
r = subprocess.run(args, cwd=cwd, env=e)
|
||
return r.returncode == 0
|
||
|
||
|
||
def verify_public(base_api=None, base_admin=None):
|
||
"""公网冒烟：健康检查 + 只读业务接口 + 管理端首页。"""
|
||
try:
|
||
import requests
|
||
|
||
try:
|
||
import urllib3
|
||
|
||
urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)
|
||
except Exception:
|
||
pass
|
||
except ImportError:
|
||
print("[失败] 验证阶段需要 requests")
|
||
return False
|
||
|
||
base_api = (base_api or os.environ.get("VERIFY_API_BASE", "https://soulapi.quwanzhi.com")).rstrip("/")
|
||
base_admin = (base_admin or os.environ.get("VERIFY_ADMIN_BASE", "https://souladmin.quwanzhi.com")).rstrip("/")
|
||
|
||
checks = [
|
||
("GET %s/health" % base_api, "%s/health" % base_api, lambda r: r.status_code == 200 and '"status"' in r.text),
|
||
("GET %s/api/book/parts" % base_api, "%s/api/book/parts" % base_api, lambda r: r.status_code == 200),
|
||
("GET %s/api/config" % base_api, "%s/api/config" % base_api, lambda r: r.status_code == 200),
|
||
("GET %s/" % base_admin, "%s/" % base_admin, lambda r: r.status_code == 200 and "root" in r.text),
|
||
]
|
||
|
||
print("")
|
||
print("=" * 60)
|
||
print(" 线上冒烟验证")
|
||
print("=" * 60)
|
||
all_ok = True
|
||
for name, url, pred in checks:
|
||
try:
|
||
r = requests.get(url, timeout=20, verify=True)
|
||
ok = pred(r)
|
||
print(" [%s] %s" % ("OK" if ok else "FAIL", name))
|
||
if not ok:
|
||
print(" status=%s len=%s" % (r.status_code, len(r.text or "")))
|
||
all_ok = False
|
||
except Exception as ex:
|
||
print(" [FAIL] %s — %s" % (name, ex))
|
||
all_ok = False
|
||
return all_ok
|
||
|
||
|
||
def main():
|
||
parser = argparse.ArgumentParser(description="kr 宝塔：BT API 重启 + 部署 + 验证")
|
||
parser.add_argument("--verify-only", action="store_true", help="仅执行公网冒烟，不部署")
|
||
parser.add_argument("--skip-bt-ping", action="store_true", help="跳过宝塔面板 API 探活")
|
||
parser.add_argument(
|
||
"--restart-method",
|
||
choices=("btapi", "auto"),
|
||
default="auto",
|
||
help="soul-api master.py 的 --restart-method（默认先宝塔 Go 插件 API，失败再 SSH）",
|
||
)
|
||
args = parser.parse_args()
|
||
|
||
cfg = _load_api_cfg()
|
||
|
||
if not args.verify_only and not args.skip_bt_ping:
|
||
if not bt_panel_ping(cfg):
|
||
print("[失败] 宝塔 API 探活未通过，已中止（可用 --skip-bt-ping 跳过）")
|
||
return 1
|
||
|
||
if args.verify_only:
|
||
return 0 if verify_public() else 1
|
||
|
||
py = sys.executable
|
||
# master.py 默认也会做 GetDiskInfo 探活；本脚本已探活则跳过，避免重复请求
|
||
master_args = [py, "master.py", "--restart-method", args.restart_method]
|
||
if not args.skip_bt_ping:
|
||
master_args.append("--skip-bt-ping")
|
||
if not run_cmd(SOUL_API, master_args):
|
||
print("[失败] soul-api 部署失败")
|
||
return 1
|
||
|
||
if not run_cmd(SOUL_ADMIN, [py, "master.py"]):
|
||
print("[失败] soul-admin 部署失败")
|
||
return 1
|
||
|
||
time.sleep(2)
|
||
if not verify_public():
|
||
print("")
|
||
print("[失败] 线上冒烟存在未通过项（部署已执行，请排查 Nginx/证书/业务）")
|
||
return 1
|
||
|
||
print("")
|
||
print(" 全部完成：已部署且冒烟通过。")
|
||
return 0
|
||
|
||
|
||
if __name__ == "__main__":
|
||
sys.exit(main() or 0)
|