xuanyuanaosheng
commented
1 day ago lib库
#!/usr/bin/env python
# encoding: utf-8
'''
@license: (C) Copyright 2018-2020, Paul
@file: BlueKingTools.py
@created_time: 12/28/2021 9:09 AM
@updated_time:
@desc: Just for fun :)
'''
import json
import requests
class BlueKingCMDBAgent(object):
def __init__(self, url, bk_username, bk_app_code, bk_app_secret, index_start=0, index_limit=50):
self.base_url = url
self.bk_app_secret = bk_app_secret
self.bk_app_code = bk_app_code
self.bk_username = bk_username
self.index_start = index_start
self.index_limit = index_limit
self.common_fields = self.initCommonFields()
def initCommonFields(self):
return {
"bk_app_code": self.bk_app_code,
"bk_app_secret": self.bk_app_secret,
"bk_username": self.bk_username,
"page": {
"start": self.index_start,
"limit": self.index_limit,
"sort": ""
}
}
def newClient(self):
session = requests.Session()
session.headers = {
"Accept": "application/json, text/plain, */*",
"Content-Type": "application/json;charset=UTF-8",
}
session.verify = False
return session
def sendRequest(self, client: requests.Session, uri, data: dict):
payload = {}
payload.update(data)
payload.update(self.common_fields)
response = client.request(
method="POST",
url=f"{self.base_url}{uri}",
data=json.dumps(payload)
)
try:
content = response.json()
return 1, content
except Exception as e:
content = response.content.decode("utf-8")
return 0, content
def getAllInstance(self, bk_obj_id):
payload = {
"bk_obj_id": bk_obj_id,
"fields": [
],
"condition": {
},
}
client = self.newClient()
return self.sendRequest(
client,
"/api/c/compapi/v2/cc/search_inst_by_object/",
payload)
def getInstanceByInstanceName(self, bk_inst_name, bk_obj_id):
payload = {
"bk_obj_id": bk_obj_id,
"fields": [
],
"condition": {
"bk_inst_name": f"{bk_inst_name}"
},
"page": {
"start": 0,
"limit": 10,
"sort": ""
}
}
client = self.newClient()
code, content = self.sendRequest(
client,
"/api/c/compapi/v2/cc/search_inst_by_object/",
payload)
#print(code, content)
if code:
return self.dataClean(content.get("data").get("info"), "bk_inst_name", bk_inst_name)
def getInstanceByIP(self, ip, bk_obj_id, method="$eq"):
payload = {
"bk_obj_id": bk_obj_id,
"condition": {
"physical_device": [
{
"field": "ip_address",
"operator": method,
"value": ip
}
],
"virtual_server": [
{
"field": "ip_address",
"operator": method,
"value": ip
}
],
"ali_cloud_server": [
{
"field": "ip_address",
"operator": method,
"value": ip
}
]
},
"page": {
"start": 0,
"limit": 20,
"sort": "bk_obj_id"
}
}
client = self.newClient()
code, content = self.sendRequest(
client,
"/api/c/compapi/v2/cc/search_inst/",
payload)
if code:
return self.dataClean(content.get("data").get("info"), "ip_address", ip)
def getInstanceBySN(self, sn, bk_obj_id, device_type, method="$eq"):
if device_type == 'Physical Server':
payload = {
"bk_obj_id": bk_obj_id,
"condition": {
"physical_device": [
{
"field": "device_sn",
"operator": method,
"value": sn
}
]
},
"page": {
"start": 0,
"limit": 20,
"sort": "bk_obj_id"
}
}
client = self.newClient()
code, content = self.sendRequest(
client,
"/api/c/compapi/v2/cc/search_inst/",
payload)
if code:
return self.dataClean(content.get("data").get("info"), "device_sn", sn)
elif device_type == 'Switch':
payload = {
"bk_obj_id": bk_obj_id,
"condition": {
"Switch": [
{
"field": "switch_sn",
"operator": method,
"value": sn
}
]
},
"page": {
"start": 0,
"limit": 20,
"sort": "bk_obj_id"
}
}
client = self.newClient()
code, content = self.sendRequest(
client,
"/api/c/compapi/v2/cc/search_inst/",
payload)
if code:
return self.dataClean(content.get("data").get("info"), "switch_sn", sn)
def createInst(self, bk_obj_id, payload):
payload.update({
"bk_obj_id": bk_obj_id
})
client = self.newClient()
code, content = self.sendRequest(
client,
"/api/c/compapi/v2/cc/create_inst/",
payload)
if code:
return content
def updateInst(self, inst_name, bk_obj_id, payload):
inst = self.getInstanceByInstanceName(inst_name, bk_obj_id)
if inst != None:
inst_id = inst.get("bk_inst_id")
payload.update({
"bk_inst_id": inst_id,
"bk_obj_id": bk_obj_id
})
client = self.newClient()
code, content = self.sendRequest(
client,
"/api/c/compapi/v2/cc/update_inst/",
payload)
if code:
return content
def updatePhysical(self, sn, bk_obj_id, device_type, payload):
inst = self.getInstanceBySN(sn, bk_obj_id, device_type)
if inst != None:
inst_id = inst.get("bk_inst_id")
payload.update({
"bk_inst_id": inst_id,
"bk_obj_id": bk_obj_id
})
client = self.newClient()
code, content = self.sendRequest(
client,
"/api/c/compapi/v2/cc/update_inst/",
payload)
if code:
return content
def deleteInstByInstName(self, inst_name, bk_obj_id):
inst = self.getInstanceByInstanceName(inst_name, bk_obj_id)
if inst != None:
inst_id = inst.get("bk_inst_id")
client = self.newClient()
code, content = self.sendRequest(
client,
"/api/c/compapi/v2/cc/delete_inst/",
{
"bk_supplier_account": "0",
"bk_obj_id": bk_obj_id,
"bk_inst_id": inst_id
})
if code:
return content
def dataClean(self, dataArray: list, field, field_value):
for inst in dataArray:
fieldValue = inst.get(field)
if fieldValue == field_value:
return inst 调用示例
- 批量字段检查脚本，可以根据实际情况，对对应的字段进行处理
# -*- coding: utf-8 -*- # This file is auto-generated, don't edit it. Thanks. from ast import IsNot import requests import json import pandas as pd import socket from BlueKingTools import BlueKingCMDBAgent
CMDB Info
cmdb_agent = BlueKingCMDBAgent( url=f"url", bk_username="automation", bk_app_code="bk_cmdb", index_limit=10000, bk_app_secret="XXXXX47-XXXX-417a-XXXX-7XXXX83078a" )
payload={} headers = {}
response=cmdb_agent.getAllInstance('virtual_server')
print(response)
all_instacnes = response[1].get("data").get("info")
print(all_instacnes)
for i,instance in enumerate(all_instacnes, start=1): i_platform_support = instance.get("platform_support") i_host_name = instance.get("bk_inst_name") i_os_type = instance.get("os_type") i_ip = instance.get("ip_address") i_environment = instance.get("Environment")
print(i,i_host_name, i_platform_support, i_os_type)
# 把主机名里面有大写的，全部转换为小写
if i_host_name.find("EN-XX") != -1 and i_host_name.find("bruce.me") == -1:
pay_load = {'bk_inst_name': i_host_name.lower() + ".bruce.me"}
response = cmdb_agent.updateInst(i_host_name,'virtual_server',pay_load)
print("将所有服务器包含大写字母的转换为小写:",i,i_host_name,response)
# 把系统为Linux, platform_support 为 "IT Linux" 换成 "Linux Engineering"
if i_os_type == "Linux" and i_platform_support == "IT Linux":
pay_load = {'platform_support': 'Linux Engineering'}
response = cmdb_agent.updateInst(i_host_name,'virtual_server',pay_load)
print("批量更新服务器所属的组:",i,i_host_name,response)
# 把 IP 为空 且能 获得IP地址的项更新
if i_ip == "":
#print(i_ip, i_host_name)
try:
i_ip = socket.gethostbyname(i_host_name)
pay_load = {'ip_address': i_ip}
response = cmdb_agent.updateInst(i_host_name,'virtual_server',pay_load)
print("IP为空且能解析到对应的IP地址:",i,i_host_name,i_ip,response)
except Exception as e:
response = cmdb_agent.deleteInstByInstName(i_host_name,'virtual_server')
print("删除IP为空且通过主机名查不到IP地址的:",i,i_host_name,i_environment, response)
# 把 IP 为空 且 环境为 Retired 的实例删掉
if i_ip == "" and (i_environment == "Retired" or i_environment == "Development" or i_environment == "Unknown"):
response = cmdb_agent.deleteInstByInstName(i_host_name,'virtual_server')
print("删除IP为空且环境为退役/开发/Unknown状态的服务器实例:",i,i_host_name,i_environment, response)
2. 通过Excel批量更新 CMDB
```batch_update.py
# -*- coding: utf-8 -*-
# This file is auto-generated, don't edit it. Thanks.
import requests
import json
import pandas as pd
from BlueKingTools import BlueKingCMDBAgent
# CMDB Info
cmdb_agent = BlueKingCMDBAgent(
url=f"url",
bk_username="automation",
bk_app_code="bk_cmdb",
bk_app_secret="XXXXXX7-975c-XXXa-XXXXX-7f8a9XXXXXX8a"
)
payload={}
headers = {}
wb = pd.read_excel("D:\blueking-api\hosts_vmware.xlsx")
for n,row in wb.iterrows():
row_dict = row.to_dict()
i_host_name = row_dict.get("hostname")
i_hostip = row_dict.get("host_ip")
i_device_site = row_dict.get("device_site")
i_environment = row_dict.get("Environment")
i_platform_support = row_dict.get("platform_support")
i_application_support = row_dict.get("app_support")
i_application_notes = row_dict.get("app_note")
#print(i_host_name,i_hostip)
pay_load = {
"bk_inst_name": i_host_name,
"virtual_subtype" : "VMWare",
"device_site": i_device_site,
"ip_address": i_hostip,
"Environment": i_environment,
"platform_support": i_platform_support,
"app_support": i_application_support,
"app_note": i_application_notes
}
## 根据实际情况，批量插入或者批量更新，同时这里也可以多考虑考虑，将两个情况考虑为一种情况。
#response = cmdb_agent.createInst("virtual_server",pay_load)
response = cmdb_agent.updateInst(i_host_name,'virtual_server',pay_load)
print(i_host_name,response)
项目
对官方SDK的简单测试
使用项目：https://github.com/TencentBlueKing/bkpaas-python-sdk , 其简单的API测试脚本如下：