马春杰杰 Exit Reader Mode

宝塔API汇总以及Python使用示例

API:

/// 获取系统基础统计
  static const apiGetSystemTotal = "/system?action=GetSystemTotal";

  /// 获取磁盘分区信息
  static const apiGetDiskInfo = "/system?action=GetDiskInfo";

  /// 获取实时状态信息(CPU、内存、网络、负载)
  static const apiGetNetWork = "/system?action=GetNetWork";

  /// 检查是否有安装任务
  static const apiGetTaskCount = "/ajax?action=GetTaskCount";

  /// 检查面板更新
  static const apiUpdatePanel = "/ajax?action=UpdatePanel";

  /// 获取网站列表
  static const apiGetSites = "/data?action=getData&table=sites";

  /// 获取网站分类
  static const apiGetSiteTypes = "/site?action=get_site_types";

  /// 获取已安装的 PHP 版本列表
  static const apiGetPHPVersion = "/site?action=GetPHPVersion";

  /// 创建网站
  static const apiAddSite = "/site?action=AddSite";

  /// 删除网站
  static const apiDeleteSite = "/site?action=DeleteSite";

  /// 停用网站
  static const apiSiteStop = "/site?action=SiteStop";

  /// 启用网站
  static const apiSiteStart = "/site?action=SiteStart";

  /// 网站到期时间
  static const apiSetEdate = "/site?action=SetEdate";

  /// 修改网站备注
  static const apiSetSitePs = "/data?action=setPs&table=sites";

  /// 获取网站备份列表
  static const apiGetBackup = "/data?action=getData&table=backup";

  /// 创建网站备份
  static const apiToBackup = "/site?action=ToBackup";

  /// 删网站备份
  static const apiDelBackup = "/site?action=DelBackup";

  /// 获取网站的域名列表
  static const apiGetDataDomain = "/data?action=getData&table=domain";

  /// 获取FTP
  static const apiGetDataFtps = "/data?action=getData&table=ftps";

  /// 获取数据库
  static const apiGetDataDatabases = "/data?action=getData&table=databases";

  /// 添加数据库
  static const apiAddDataDatabase = "/database?action=AddDatabase";

  /// 添加域名
  static const apiAddDomain = "/site?action=AddDomain";

  /// 删除域名
  static const apiDelDomain = "/site?action=DelDomain";

  /// 获取可选的预定义伪静态列表
  static const apiGetRewriteList = "/site?action=GetRewriteList";

  /// 获取指定预定义伪静态规则内容(获取文件内容)
  static const apiGetFileBody = "/files?action=GetFileBody";

  /// 保存伪静态规则内容(保存文件内容)
  static const apiSaveFileBody = "/files?action=SaveFileBody";

  /// 取回指定网站的根目录
  static const apiGetSiteKeyPath = "/data?action=getKey&table=sites&key=path";

  /// 取回防跨站配置/运行目录/日志开关状态/可设置的运行目录列表/密码访问状态
  static const apiGetDirUserINI = "/site?action=GetDirUserINI";

  /// 设置防跨站状态(自动取反)
  static const apiSetDirUserINI = "/site?action=SetDirUserINI";

  /// 设置是否写访问日志
  static const apiSetLogsOpen = "/site?action=logsOpen";

  /// 修改网站根目录
  static const apiSetPath = "/site?action=SetPath";

  /// 设置是否写访问日志
  static const apiSetSiteRunPath = "/site?action=SetSiteRunPath";

  /// 设置密码访问
  static const apiSetHasPwd = "/site?action=SetHasPwd";

  /// 关闭密码访问
  static const apiCloseHasPwd = "/site?action=CloseHasPwd";

  /// 获取流量限制相关配置(仅支持 nginx)
  static const apiGetLimitNet = "/site?action=GetLimitNet";

  /// 开启或保存流量限制配置(仅支持 nginx)
  static const apiSetLimitNet = "/site?action=SetLimitNet";

  /// 关闭流量限制(仅支持 nginx)
  static const apiCloseLimitNet = "/site?action=CloseLimitNet";

  /// 取默认文档信息
  static const apiGetIndex = "/site?action=GetIndex";

  /// 设置默认文档
  static const apiSetIndex = "/site?action=SetIndex";

  /// 获取文件
  static const apiGetDir = "/files?action=GetDir";

  /// 创建数据库备份
  static const apiDatabaseToBackup = "/database?action=ToBackup";

  /// 重置密码
  static const apiResDatabasePassword = "/database?action=ResDatabasePassword";

  /// 删除数据库
  static const apiDeleteDatabase = "/database?action=DeleteDatabase";

  /// 修改数据库备注
  static const apiSetDatabasePs = "/data?action=setPs&table=databases";

  /// 删除文件
  static const apiDeleteFile = "/files?action=DeleteFile";

  /// 恢复数据库
  static const apiInputSql = "/database?action=InputSql";

  /// 添加FTP
  static const apiAddFtpUser = "/ftp?action=AddUser";

  /// 停用启用FTP
  static const apiFtpStatus = "/ftp?action=SetStatus";

  /// 设置FTP密码
  static const apiFtpSetUserPassword = "/ftp?action=SetUserPassword";

  /// 删除FTP用户
  static const apiFtpDeleteUser = "/ftp?action=DeleteUser";

  /// 新建目录
  static const apiFileCreateDir = "/files?action=CreateDir";

  /// 新建文件夹
  static const apiFileCreateFile = "/files?action=CreateFile";

  /// 移动文件
  static const apiFileMvFile = "/files?action=MvFile";

  /// 复制文件
  static const apiFileCopyFile = "/files?action=CopyFile";

  /// 压缩文件
  static const apiFileZip = "/files?action=Zip";

  /// 解压文件
  static const apiFileUnZip = "/files?action=UnZip";

  /// 面板操作日志
  static const apiGetLogs = "/data?action=getData&table=logs&tojs=getLogs";

  /// 面板运行日志
  static const apiGetRunLogs = "/config?action=get_panel_error_logs";

  /// 计划任务日志
  static const apiGetTaskLogs = "/crontab?action=GetLogs";

  /// 获取计划任务
  static const apiGetCronTab = "/data?action=getData&table=crontab";

  /// 网站操作日志
  static const apiGetSiteLogs = "/logs/panel/get_logs_bytype";

  /// 网站运行日志
  static const apiGetSiteRunLogs = "/site?action=GetSiteLogs";

  /// 网站错误日志
  static const apiGetSiteErrLogs = "/site?action=get_site_errlog";

  /// 端口规则
  static const apiGetPortRulesList = "/safe/firewall/get_rules_list";

  /// 端口转发
  static const apiGetPortForwardsList = "/safe/firewall/get_forward_list";

  /// IP规则
  static const apiGetIPRulesList = "/safe/firewall/get_ip_rules_list";

  /// 地区规则
  static const apiGetCountryRulesList = "/safe/firewall/get_country_list";

  /// 创建端口规则
  static const apiCreatePortRuleList = "/safe/firewall/create_rules";

  /// 删除端口规则
  static const apiDeleteortRuleList = "/safe/firewall/remove_rules";

  /// 删除端口规则
  static const apiModifyPortRule = "/safe/firewall/modify_rules";

  /// 创建IP规则
  static const apiCreateIpRule = "/safe/firewall/create_ip_rules";

  /// 删除IP规则
  static const apiDeleteIpRule = "/safe/firewall/remove_ip_rules";

  /// 删除IP规则
  static const apiModifyIpRule = "/safe/firewall/modify_ip_rules";

  /// 创建端口转发规则
  static const apiCreateForward = "/safe/firewall/create_forward";

  /// 删除端口转发规则
  static const apiDeleteForward = "/safe/firewall/remove_forward";

  /// 删除端口转发规则
  static const apiModifyForward = "/safe/firewall/modify_forward";

  /// 创建地区规则
  static const apiCreateCountry = "/safe/firewall/create_country";

  /// 删除地区规则
  static const apiDeleteCountry = "/safe/firewall/remove_country";

  /// 修改地区规则
  static const apiModifyCountry = "/safe/firewall/modify_country";

  /// 获取国家列表
  static const apiGetCountrys = "/safe/firewall/get_countrys";

  /// 获取防火墙信息
  static const apiGetFirewallInfo = "/safe/firewall/get_firewall_info";

  /// 获取SSH信息
  static const apiGetSshInfo = "/safe/ssh/GetSshInfo";

  /// 禁ping设置
  static const apiSetPing = "/firewall?action=SetPing";

  /// 防火墙开关设置
  static const apiFirewallAdmin = "/safe/firewall/firewall_admin";

  /// SSH开关设置
  static const apiSetSSH = "/firewall?action=SetSshStatus";

  /// 启动计划任务
  static const apiStartTask = "/crontab?action=StartTask";

  /// 设置计划任务状态
  static const apiSetTaskStatus = "/crontab?action=set_cron_status";

  /// 删除计划任务状态
  static const apiDelCrontab = "/crontab?action=DelCrontab";

  /// 计划任务
  static const apiGetCrontabList = "/crontab?action=GetCrontab";

  /// 释放内存
  static const apiReMemory = "/system?action=ReMemory";

  /// 获取配置
  static const apiGetConfig = "/config?action=get_config";

  /// 配置面板
  static const apiSetPanel = "/config?action=setPanel";

  /// 负载
  static const apiGetLoadAverage = "/ajax?action=get_load_average";

  /// CPU 内存
  static const apiGetCpuIo = "/ajax?action=GetCpuIo";

  /// 磁盘IO
  static const apiGetDiskIo = "/ajax?action=GetDiskIo";

  /// 网络IO
  static const apiGetNetWorkIo = "/ajax?action=GetNetWorkIo";

  /// 获取监控状态
  static const apiSetControl = "/config?action=SetControl";

  /// 获取软件列表
  static const apiGetSoftList = "/plugin?action=get_soft_list";

  /// 卸载软件
  static const apiUnstallPlugin = "/plugin?action=uninstall_plugin";

  /// 服务管理
  static const apiServiceAdmin = "/system?action=ServiceAdmin";

  /// 网站数据获取
  static const apiGetDataList = "/crontab?action=GetDataList";

  /// 数据库数据获取
  static const apiGetDataBaseList = "/crontab?action=GetDatabases";

  /// 添加计划任务
  static const apiAddCrontab = "/crontab?action=AddCrontab";

  /// 查询计划任务
  static const apiGetCrondFind = "/crontab?action=get_crond_find";

  /// 修改计划任务
  static const apiModifyCrond = "/crontab?action=modify_crond";

  /// 远程下载
  static const apiDownloadFile = "/files?action=DownloadFile";

Python示例:

# -*- coding: utf-8 -*-
import hashlib
import time
import requests
import prettytable


# 宝塔面板操作类
class BtPanel:
    __BTURL = ''
    __APIKEY = ''
    __REQ = requests.session()

    # 初始化宝塔面板
    def __init__(self, host: str, apisk: str):
        '''
        :param host: 宝塔面板地址(末尾不加/)
        :param apisk: 宝塔面板API密钥
        '''
        self.__BTURL = host
        self.__APIKEY = apisk

    # 计算MD5
    def __GetMD5(self, s: str):
        '''
        计算字符串的MD5值
        :param s: 待计算的字符串
        :return: MD5值
        '''
        m = hashlib.md5()
        m.update(s.encode('utf-8'))
        return m.hexdigest()

    # 签名计算
    # 签名计算
    def __GetToken(self):
        request_time = int(time.time())  # 获取请求时间戳
        # 注意以下行中的改动:去除了逗号
        request_token = self.__GetMD5(str(request_time) + '' + self.__GetMD5(self.__APIKEY))
        return {'request_time': request_time, 'request_token': request_token}

    # 获取站点列表
    def GetSites(self, showlog=True):
        '''
        获取宝塔面板站点列表
        :showlog: 是否输出结果
        :return: 站点列表数据
        '''
        if showlog:
            print('\n### 获取站点列表...')
        tk = self.__GetToken()    # 获取签名
        playload = {
            'request_time': tk['request_time'],
            'request_token': tk['request_token'],
            'p': 1,
            'limit': 100,
            'order': 'id'
        }
        print(playload)
        res = self.__REQ.post(url=self.__BTURL + '/data?action=getData&table=sites', data=playload).json()
        # 判断请求是否成功
        if 'status' in res and 'msg' in res and not res['status']:
            print('>>> 获取站点列表失败:', res['msg'])
            return False
        # 使用prettytable输出站点列表
        if showlog:
            tb = prettytable.PrettyTable()
            tb.field_names = ['网站名', '站点类型', '备注', 'SSL域名', 'SSL到期时间', 'SSL剩余天数', '证书品牌', '状态']
            tb.align['网站名'] = 'l'
            tb.align['备注'] = 'l'
            tb.align['SSL域名'] = 'l'
            tb.align['证书品牌'] = 'l'
            for site in res['data']:
                site_status = '运行' if site['status'] == '1' else '停止'
                if site['ssl'] == -1:
                    sslinfo = {'notAfter': '-', 'endtime': '-', 'subject': '-', 'issuer': '-'}
                else:
                    sslinfo = {'notAfter': site['ssl']['notAfter'], 'endtime': site['ssl']['endtime'],
                               'subject': site['ssl']['subject'], 'issuer': site['ssl']['issuer']}
                # 安全访问字典键,使用get方法提供默认值
                project_type = site.get('project_type', '未知类型')
                tb.add_row(
                    [site['name'], project_type, site.get('ps', '无备注'), sslinfo['subject'], sslinfo['notAfter'],
                     sslinfo['endtime'], sslinfo['issuer'], site_status])

            print(tb)
        return res['data']

    # 设置站点SSL证书
    def SetSSL(self, site_name: str, ssl_cert_content: str, ssl_key_content: str):
        '''
        设置站点SSL证书
        :param site_name: 站点名称(域名)
        :param ssl_cert_content: ssl证书内容
        :param ssl_key_content: ssl私钥内容
        :return: 设置结果
        '''
        print('\n### 设置站点SSL证书...')
        tk = self.__GetToken()    # 获取签名
        playload = {
            'request_time': tk['request_time'],
            'request_token': tk['request_token'],
            'type': 0,
            'siteName': site_name,
            'key': ssl_key_content,
            'csr': ssl_cert_content
        }
        res = self.__REQ.post(url=self.__BTURL + '/site?action=SetSSL', data=playload).json()
        # 判断请求是否成功
        if 'status' in res and 'msg' in res and not res['status']:
            print('>>> 设置站点SSL证书失败:', res['msg'])
            return False
        print('>>>', res['msg'])    # 输出结果
        return res['status']

    # 获取证书夹列表
    def GetCertList(self, showlog=True):
        if showlog:
            print('\n### 获取证书夹列表...')
        tk = self.__GetToken()  # 获取签名
        playload = {
            'request_time': tk['request_time'],
            'request_token': tk['request_token'],
            'force_refresh': 1  # 0:获取本地证书 1:获取云端证书
        }
        print("请求参数:", playload)  # 打印请求参数用于调试
        res = self.__REQ.post(url=self.__BTURL + '/ssl?action=getcertlist', data=playload)
        print("响应内容:", res.text)  # 打印响应内容用于调试

        if res.status_code == 200:
            data = res.json()
            if 'status' in data and 'msg' in data and not data['status']:
                print('>>> 获取证书夹列表失败:', data['msg'])
                return False
            # 使用prettytable输出证书夹列表
            if showlog and 'data' in data:
                tb = prettytable.PrettyTable()
                tb.field_names = ['域名', 'SSL到期时间', 'SSL剩余天数', '证书品牌', '可选域名']
                tb.align['域名'] = 'l'
                tb.align['证书品牌'] = 'l'
                tb.align['可选域名'] = 'l'
                for cert in data['data']:
                    tb.add_row([cert['subject'], cert['info']['notAfter'], cert['endtime'], cert['info']['issuer'],
                                cert['dns']])
                print(tb)
            return data
        else:
            print('>>> API请求失败,状态码:', res.status_code)
            return False

    # 删除过期SSL证书
    def DelExpiredSSL(self):
        '''
        删除过期SSL证书
        :showlog: 是否输出结果
        :return: 删除结果
        '''
        print('\n### 删除过期SSL证书...')
        certs = self.GetCertList(showlog=False)    # 获取证书列表
        if not certs:
            return False
        expiredcerts = []    # 过期证书列表
        for cert in certs:
            if cert['endtime'] < 0:    # 证书已过期
                expiredcerts.append(cert['subject'])    # 加入过期证书列表
                tk = self.__GetToken()    # 获取签名
                playload = {
                    'request_time': tk['request_time'],
                    'request_token': tk['request_token'],
                    'local': 1,
                    'ssl_hash': cert['hash']
                }
                res = self.__REQ.post(url=self.__BTURL + '/ssl?action=remove_cloud_cert', data=playload).json()
                # 判断请求是否成功
                if 'status' in res and 'msg' in res and not res['status']:
                    print('>>> 删除过期SSL证书失败:', res['msg'])
                    return False
                print('>>> 过期证书', cert['subject'], res['msg'])
                return res['status']
        if len(expiredcerts) == 0:
            print('>>> 证书夹内未发现过期证书')
        return expiredcerts

    # 重启Nginx
    def RestartNginx(self):
        '''
        重启Nginx
        :return: 重启结果
        '''
        tk = self.__GetToken()    # 获取签名
        playload = {
            'request_time': tk['request_time'],
            'request_token': tk['request_token'],
            'name': 'nginx',
            'type': 'restart'
        }
        res = self.__REQ.post(url=self.__BTURL + '/system?action=ServiceAdmin', data=playload).json()
        # 判断请求是否成功
        if 'status' in res and 'msg' in res and not res['status']:
            print('>>> 重启Nginx失败:', res['msg'])
            return False
        return res.json()['status']

    # 获取系统基本信息
    def GetSystemInfo(self, showlog=True):
        '''
        获取系统基本信息
        :showlog: 是否输出结果
        :return: 系统基本信息
        '''
        if showlog:
            print('\n### 获取系统基本信息...')
        tk = self.__GetToken()    # 获取签名
        playload = {
            'request_time': tk['request_time'],
            'request_token': tk['request_token']
        }
        res = self.__REQ.post(url=self.__BTURL + '/system?action=GetSystemTotal', data=playload).json()
        # 判断请求是否成功
        # 判断res是否存在status字段,若不存在则说明请求失败
        if 'status' in res and 'msg' in res and not res['status']:
            print('>>> 获取系统基本信息失败:', res['msg'])
            return False
        res['memUtilization'] = round(res['memRealUsed'] / res['memTotal'], 2) * 100    # 内存使用率
        if showlog:
            print('>>> 面板地址:', self.__BTURL)
            print('>>> 操作系统:', res['system'])
            print('>>> 面板版本:', res['version'])
            print('>>> 运行时间:', res['time'])
            print('>>> CPU使用率:', res['cpuRealUsed'], '%')
            print('>>> 内存使用率:', res['memUtilization'], '%')
            print('>>> 内存总量:', round(res['memTotal'] / 1024, 2), 'GB')
        return res
if __name__ == '__main__':

    bt = BtPanel('http://xxxx:xxx', 'xxxxx')
    bt.GetSystemInfo()
    bt.GetSites()