156 lines
5.0 KiB
Python
156 lines
5.0 KiB
Python
#!/usr/bin/env python3
|
||
# -*- coding: utf-8 -*-
|
||
"""
|
||
宝塔面板 API 模块 - Node 项目重启 / 计划任务触发
|
||
|
||
被 devlop.py 内部调用；也可单独使用：
|
||
python scripts/deploy_baota_pure_api.py # 重启 Node 项目
|
||
python scripts/deploy_baota_pure_api.py --create-dir # 并创建项目目录
|
||
python scripts/deploy_baota_pure_api.py --task-id 1 # 触发计划任务 ID=1
|
||
|
||
环境变量：
|
||
BAOTA_PANEL_URL # 宝塔面板地址，如 https://42.194.232.22:9988 或带安全入口
|
||
BAOTA_API_KEY # 宝塔 API 密钥（面板 → 设置 → API 接口）
|
||
DEPLOY_PM2_APP # PM2 项目名称，默认 soul
|
||
"""
|
||
|
||
from __future__ import print_function
|
||
|
||
import os
|
||
import sys
|
||
import time
|
||
import hashlib
|
||
|
||
try:
|
||
import requests
|
||
import urllib3
|
||
urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)
|
||
except ImportError:
|
||
print("请先安装: pip install requests")
|
||
sys.exit(1)
|
||
|
||
# 配置：可通过环境变量覆盖
|
||
CFG = {
|
||
"panel_url": os.environ.get("BAOTA_PANEL_URL", "https://42.194.232.22:9988"),
|
||
"api_key": os.environ.get("BAOTA_API_KEY", "hsAWqFSi0GOCrunhmYdkxy92tBXfqYjd"),
|
||
"pm2_name": os.environ.get("DEPLOY_PM2_APP", "soul"),
|
||
"project_path": os.environ.get("DEPLOY_PROJECT_PATH", "/www/wwwroot/soul"),
|
||
"site_url": os.environ.get("DEPLOY_SITE_URL", "https://soul.quwanzhi.com"),
|
||
}
|
||
|
||
|
||
def _get_sign(api_key):
|
||
"""宝塔鉴权签名：request_token = md5(request_time + md5(api_key))"""
|
||
now_time = int(time.time())
|
||
sign_str = str(now_time) + hashlib.md5(api_key.encode("utf-8")).hexdigest()
|
||
request_token = hashlib.md5(sign_str.encode("utf-8")).hexdigest()
|
||
return now_time, request_token
|
||
|
||
|
||
def _request(base_url, path, data=None, timeout=30):
|
||
"""发起宝塔 API 请求"""
|
||
url = base_url.rstrip("/") + "/" + path.lstrip("/")
|
||
api_key = CFG["api_key"]
|
||
if not api_key:
|
||
print("请设置 BAOTA_API_KEY（宝塔面板 → 设置 → API 接口）")
|
||
return None
|
||
req_time, req_token = _get_sign(api_key)
|
||
payload = {
|
||
"request_time": req_time,
|
||
"request_token": req_token,
|
||
}
|
||
if data:
|
||
payload.update(data)
|
||
try:
|
||
r = requests.post(
|
||
url,
|
||
data=payload,
|
||
verify=False,
|
||
timeout=timeout,
|
||
)
|
||
return r.json() if r.text else None
|
||
except Exception as e:
|
||
print("请求失败:", e)
|
||
return None
|
||
|
||
|
||
def restart_node_project(panel_url, api_key, pm2_name):
|
||
"""
|
||
通过宝塔 API 重启 Node 项目
|
||
返回 True 表示成功，False 表示失败
|
||
"""
|
||
# Node 项目管理为插件接口，路径可能因版本不同
|
||
paths_to_try = [
|
||
"/plugin?action=a&name=nodejs&s=restart_project",
|
||
"/project/nodejs/restart_project",
|
||
]
|
||
payload = {"project_name": pm2_name}
|
||
req_time, req_token = _get_sign(api_key)
|
||
payload["request_time"] = req_time
|
||
payload["request_token"] = req_token
|
||
|
||
url_base = panel_url.rstrip("/")
|
||
for path in paths_to_try:
|
||
url = url_base + path
|
||
try:
|
||
r = requests.post(url, data=payload, verify=False, timeout=30)
|
||
j = r.json() if r.text else {}
|
||
if j.get("status") is True or j.get("msg") or r.status_code == 200:
|
||
print(" 重启成功: %s" % pm2_name)
|
||
return True
|
||
# 某些版本返回不同结构
|
||
if "msg" in j:
|
||
print(" API 返回:", j.get("msg", j))
|
||
except Exception as e:
|
||
print(" 尝试 %s 失败: %s" % (path, e))
|
||
print(" 重启失败，请检查宝塔 Node 插件是否安装、API 密钥是否正确")
|
||
return False
|
||
|
||
|
||
def create_project_dir():
|
||
"""通过宝塔文件接口创建项目目录"""
|
||
path = "/files?action=CreateDir"
|
||
data = {"path": CFG["project_path"]}
|
||
j = _request(CFG["panel_url"], path, data)
|
||
if j and j.get("status") is True:
|
||
print(" 目录已创建: %s" % CFG["project_path"])
|
||
return True
|
||
print(" 创建目录失败:", j)
|
||
return False
|
||
|
||
|
||
def trigger_crontab_task(task_id):
|
||
"""触发计划任务"""
|
||
path = "/crontab?action=StartTask"
|
||
data = {"id": str(task_id)}
|
||
j = _request(CFG["panel_url"], path, data)
|
||
if j and j.get("status") is True:
|
||
print(" 计划任务 %s 已触发" % task_id)
|
||
return True
|
||
print(" 触发失败:", j)
|
||
return False
|
||
|
||
|
||
def main():
|
||
import argparse
|
||
parser = argparse.ArgumentParser(description="宝塔 API - 重启 Node / 触发计划任务")
|
||
parser.add_argument("--create-dir", action="store_true", help="创建项目目录")
|
||
parser.add_argument("--task-id", type=int, default=0, help="触发计划任务 ID")
|
||
args = parser.parse_args()
|
||
|
||
if args.create_dir:
|
||
create_project_dir()
|
||
if args.task_id:
|
||
ok = trigger_crontab_task(args.task_id)
|
||
sys.exit(0 if ok else 1)
|
||
ok = restart_node_project(
|
||
CFG["panel_url"],
|
||
CFG["api_key"],
|
||
CFG["pm2_name"],
|
||
)
|
||
sys.exit(0 if ok else 1)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|