阿里云数据库ClickHouse添加当前公网ip至白名单

公司路由器重拨公网ip就会改变，每次改变就要手动去设置阿里云数据库ClickHouse 的访问白名单，于是想写个脚本自动设置。
查看阿里的Clickhouse 文档发现是有提供几个api，但未整合进阿里的Openapi, 也就没办法在线调试了，只能自己写。
ck api文档地址： https://help.aliyun.com/document_detail/170869.html

#1. 安装SDK

pip install aliyun-python-sdk-clickhouse

#2. python3 代码

from aliyunsdkcore.client import AcsClient
from aliyunsdkcore.acs_exception.exceptions import ClientException
from aliyunsdkcore.acs_exception.exceptions import ServerException
from aliyunsdkclickhouse.request.v20191111 import DescribeDBClusterAccessWhiteListRequest
from aliyunsdkclickhouse.request.v20191111 import ModifyDBClusterAccessWhiteListRequest
from aliyunsdkclickhouse.request.v20191111 import DescribeDBClustersRequest
import json
import requests


client = AcsClient(
    "LTAI4GAKxxxxxxxxxxxxxxxxx",
    "7pm5iPYegxxxxxxxxxxxxxxxxxxxxx",
    "cn-hangzhou"
)


def postdingding(ip, cluster_id):
    HEADERS = {"Content-Type": "application/json ;charset=utf-8"}
    data = {
        "msgtype": "markdown",
        "markdown": {
            "title": f'阿里云Clickhouse 添加公网ip 成功',
            "text": f'已添加 {ip} 至dev ck集群 {cluster_id}'
        },
        "at": {
            "atMobiles": [
                ""
            ],
            "isAtAll": "true"
        }
    }
    String_textMsg = json.dumps(data)
    # 钉钉地址
    dingdingurl = 'https://oapi.dingtalk.com/robot/send?access_token=91d1946f77712d8256axxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx'
    res = requests.post(dingdingurl, data=String_textMsg, headers=HEADERS)


def get_cluster_id():
    request = DescribeDBClustersRequest.DescribeDBClustersRequest()
    request.set_accept_format(json)
    response = client.do_action_with_exception(request)
    res = str(response, encoding='utf-8')
    js = json.loads(res)
    cids = []
    for c in js['DBClusters']['DBCluster']:
        cids.append(c['DBClusterId'])
    return cids


def get_public_whitelist(cluster_id):
    request = DescribeDBClusterAccessWhiteListRequest.DescribeDBClusterAccessWhiteListRequest()
    request.set_DBClusterId(cluster_id)
    request.set_accept_format(json)
    response = client.do_action_with_exception(request)
    res = str(response, encoding='utf-8')
    js = json.loads(res)
    ip_list = []
    for i in js['DBClusterAccessWhiteList']['IPArray']:
        if i["DBClusterIPArrayName"] == 'public':
            ip_list = i["SecurityIPList"].split(',')
    return ip_list


def add_ip_to_public_whitelist(cluster_id, ip):
    request = ModifyDBClusterAccessWhiteListRequest.ModifyDBClusterAccessWhiteListRequest()
    request.set_DBClusterId(cluster_id)
    request.set_DBClusterIPArrayName('public')
    request.set_SecurityIps(ip)
    request.set_ModifyMode('Append')
    response = client.do_action_with_exception(request)
    res = str(response, encoding='utf-8')
    print(res)


def get_public_ip():
    return requests.get('https://checkip.amazonaws.com').text.strip()


def main():
    ck_cluster_list = get_cluster_id()
    current_ip = get_public_ip()
    print(current_ip)
    for cid in ck_cluster_list:
        whitelist = get_public_whitelist(cid)
        print(whitelist)
        if current_ip not in whitelist:
            add_ip_to_public_whitelist(cid, current_ip)
            postdingding(current_ip, cid)


if __name__ == '__main__':
    main()

#3. crontab 定时执行

# crontab -e
* * * * * /root/daily_monitor/bin/python3 /root/daily_monitor/ali_ck_whitelist.py