搬瓦工高级教程:使用Python调用KiwiVM面板API

12/26/2019 11:37 上午 posted in  VPN

其实这些Python代码都很简单。就是调用一下了上面的API。官方给出的示例是PHP的(世界上最好的语言)。

当然,也有一些通过wget操作的,比如通过wget重启:

wget -qO- "https://api.64clouds.com/v1/restart?veid=706783&api_key=YOUR_API_KEY_HERE"
有了这些API有什么用呢?最简单的一个例子,如果你想每天定时给搬瓦工开关机,那么你就可以通过这些API来实现。比如你在另外一台VPS上写个脚本,每天定时把你的搬瓦工开机,每天定时再把它关机,是不是很有意思?(真的吗?)

或者你再有时间,可以自己搞个带GUI的Python的客户端。

代码在下面。稍后更新Github地址。

#/usr/bin/env python
# -*- coding: utf-8 -*-

import requests
import json
from utils import *
from apis import *

class Bandwagonhost(object):

    def __init__(self, veid, api_key):
        self.veid = veid
        self.api_key = api_key

    def start_vps(self):
        r = requests.get('https://api.64clouds.com/v1/start?veid=' + self.veid + '&api_key=' + self.api_key)
        print json.dumps(r.json(), indent=1)

    def stop_vps(self):
        r = requests.get('https://api.64clouds.com/v1/stop?veid=' + self.veid + '&api_key=' + self.api_key)
        print json.dumps(r.json(), indent=1)

    def restart_vps(self):
        r = requests.get('https://api.64clouds.com/v1/restart?veid=' + self.veid + '&api_key=' + self.api_key)
        print json.dumps(r.json(), indent=1)

    def kill_vps(self):
        r = requests.get('https://api.64clouds.com/v1/kill?veid=' + self.veid + '&api_key=' + self.api_key)
        print json.dumps(r.json(), indent=1)

    def get_server_info(self):
        r = requests.get('https://api.64clouds.com/v1/getServiceInfo?veid=' + self.veid + '&api_key=' + self.api_key)
        print json.dumps(r.json(), indent=1)
        return json.dumps(r.json(), indent=1)

    def get_live_service_info(self):
        r = requests.get('https://api.64clouds.com/v1/getLiveServiceInfo?veid=' + self.veid + '&api_key=' + self.api_key)
        print json.dumps(r.json(), indent=1)

    def get_available_os(self):
        r = requests.get('https://api.64clouds.com/v1/getAvailableOS?veid=' + self.veid + '&api_key=' + self.api_key)
        print json.dumps(r.json(), indent=1)

    def reinstall_os(self, os):
        r = requests.get(
            'https://api.64clouds.com/v1/reinstallOS?os=' + os + '&veid=' + self.veid + '&api_key=' + self.api_key)
        print json.dumps(r.json(), indent=1)

    def reset_root_password(self):
        r = requests.get('https://api.64clouds.com/v1/resetRootPassword?veid=' + self.veid + '&api_key=' + self.api_key)
        print json.dumps(r.json(), indent=1)

    # TODO draw a fig instead
    def get_raw_usage_stats(self):
        r = requests.get('https://api.64clouds.com/v1/getRawUsageStats?veid=' + self.veid + '&api_key=' + self.api_key)
        print json.dumps(r.json(), indent=1)

    def set_hostname(self, newHostname):
        server = self.get_server_info()
        if server.json()['vm_type'] != 'ovz':
            print 'only for ovz servers'
            return False
        else:
            r = requests.get(
                'https://api.64clouds.com/v1/setHostname?newHostname=' + newHostname
                + '&veid=' + self.veid + '&api_key=' + self.api_key)
            print json.dumps(r.json(), indent=1)

    def set_ptr_record(self, ip, ptr):
        r = requests.get(
            'https://api.64clouds.com/v1/setPTR?ip=' + ip + '&ptr=' + ptr + '&veid='
            + self.veid + '&api_key=' + self.api_key)
        print json.dumps(r.json(), indent=1)

    # no need
    def basic_shell_cd(self):
        pass

    def basic_shell_exec(self):
        pass

    def shell_script_exec(self):
        pass

    def snapshot_create(self, description='Automatic_Snapshot'):
        r = requests.get(
            'https://api.64clouds.com/v1/snapshot/create?description=' + description + '&veid='
            + self.veid + '&api_key=' + self.api_key)
        print json.dumps(r.json(), indent=1)

    def snapshot_list(self):
        r = requests.get(
            'https://api.64clouds.com/v1/snapshot/list?veid=' + self.veid + '&api_key=' + self.api_key)
        print json.dumps(r.json(), indent=1)

    def snapshot_delete(self, snapshot):
        r = requests.get(
            'https://api.64clouds.com/v1/snapshot/delete?snapshot=' + snapshot
            + '&veid=' + self.veid + '&api_key=' + self.api_key)
        print json.dumps(r.json(), indent=1)

    def snapshot_restore(self, snapshot):
        r = requests.get(
            'https://api.64clouds.com/v1/snapshot/restore?snapshot=' + snapshot
            + '&veid=' + self.veid + '&api_key=' + self.api_key)
        print json.dumps(r.json(), indent=1)

    def snapshot_toggle_sticky(self, snapshot, sticky):
        r = requests.get(
            'https://api.64clouds.com/v1/snapshot/toggleSticky?snapshot=' + snapshot
            + '&sticky=' + sticky + '&veid=' + self.veid + '&api_key=' + self.api_key)
        print json.dumps(r.json(), indent=1)

    def snapshot_export(self, snapshot):
        r = requests.get(
            'https://api.64clouds.com/v1/snapshot/export?snapshot=' + snapshot
            + '&veid=' + self.veid + '&api_key=' + self.api_key)
        print json.dumps(r.json(), indent=1)

    def snapshot_import(self, source_veid, source_token):
        r = requests.get(
            'https://api.64clouds.com/v1/snapshot/import?sourceVeid=' + source_veid + '&sourceToken=' + source_token
            + '&veid=' + self.veid + '&api_key=' + self.api_key)
        print json.dumps(r.json(), indent=1)

    # TODO need test if the ip is ipv4 or ipv6
    def ipv6_add(self, ip):
        r = requests.get(
            'https://api.64clouds.com/v1/ipv6/add?ip=' + ip + '&veid=' + self.veid + '&api_key=' + self.api_key)
        print json.dumps(r.json(), indent=1)

    def ipv6_delete(self, ip):
        r = requests.get(
            'https://api.64clouds.com/v1/ipv6/delete?ip=' + ip + '&veid=' + self.veid + '&api_key=' + self.api_key)
        print json.dumps(r.json(), indent=1)

    def migrate_get_locations(self):
        r = requests.get(
            'https://api.64clouds.com/v1/migrate/getLocations?veid=' + self.veid + '&api_key=' + self.api_key)
        print json.dumps(r.json(), indent=1)

    def migrate_start(self, location):
        r = requests.get(
            'https://api.64clouds.com/v1/migrate/start?location=' + location
            + '&veid=' + self.veid + '&api_key=' + self.api_key)
        print json.dumps(r.json(), indent=1)

    def clone_from_external_server(self, external_server_ip, external_server_ssh_port, external_server_root_password):
        server = self.get_server_info()
        if server.json()['vm_type'] != 'ovz':
            print 'only for ovz servers'
            return False
        else:
            r = requests.get(
                'https://api.64clouds.com/v1/cloneFromExternalServer?externalServerIP=' + external_server_ip
                + '&externalServerSSHport=' + external_server_ssh_port + '&externalServerRootPassword'
                + external_server_root_password + '&veid=' + self.veid + '&api_key=' + self.api_key)
            print json.dumps(r.json(), indent=1)

    def get_suspension_details(self):
        r = requests.get(
            'https://api.64clouds.com/v1/getSuspensionDetail?&veid=' + self.veid + '&api_key=' + self.api_key)
        print json.dumps(r.json(), indent=1)

    def unsuspend(self, record_id):
        r = requests.get(
            'https://api.64clouds.com/v1/unsuspend?record_id=' + record_id
            + '&veid=' + self.veid + '&api_key=' + self.api_key)
        print json.dumps(r.json(), indent=1)

    def get_rate_limit_status(self):
        r = requests.get(
            'https://api.64clouds.com/v1/getRateLimitStatus?veid=' + self.veid + '&api_key=' + self.api_key)
        print json.dumps(r.json(), indent=1)

if __name__ == '__main__':
    # do some test
    bwh = Bandwagonhost(VEID_BWH,API_KEY_BWH)
    bwh_info = bwh.get_server_info()
    # bwh.create_snapshot()
    # bwh.get_raw_usage_stats()
    for keys in json.loads(bwh_info).keys():
       print json.loads(bwh_info)[keys]