Python利用解析JSON實現(xiàn)主機管理
更新時間：2023年12月05日 08:05:00 作者：微軟技術(shù)分享
JSON 是一種獨立于編程語言的數(shù)據(jù)格式,因此在不同的編程語言中都有對應(yīng)的解析器和生成器,本文主要介紹了Python如何通過解析JSON實現(xiàn)主機管理,感興趣的小伙伴可以了解一下
前言
JSON（JavaScript Object Notation）是一種輕量級的數(shù)據(jù)交換格式，它以易于閱讀和編寫的文本形式表示數(shù)據(jù)。JSON 是一種獨立于編程語言的數(shù)據(jù)格式，因此在不同的編程語言中都有對應(yīng)的解析器和生成器。JSON 格式的設(shè)計目標(biāo)是易于理解、支持復(fù)雜數(shù)據(jù)結(jié)構(gòu)和具有良好的可擴展性。
JSON 數(shù)據(jù)是以鍵值對的形式存在的，而且易于閱讀和編寫。以下是一個簡單的 JSON 示例：
{
"name": "John Doe",
"age": 30,
"city": "New York",
"isStudent": false,
"grades": [95, 88, 75, 92],
"address": {
"street": "123 Main St",
"zipCode": "10001"
}
}
在這個例子中，JSON 對象包含了一些屬性，包括字符串、數(shù)字、布爾值、數(shù)組和嵌套的對象。
"name": "John Doe"：字符串鍵值對。"age": 30：數(shù)字鍵值對。"city": "New York"：字符串鍵值對。"isStudent": false：布爾鍵值對。"grades": [95, 88, 75, 92]：數(shù)組鍵值對。"address": {...}：嵌套對象。
在實際應(yīng)用中，JSON 數(shù)據(jù)通常用于前后端之間的數(shù)據(jù)交換，或者配置文件的存儲。各種編程語言都提供了處理 JSON數(shù)據(jù)的庫或模塊。
很早之前大概是兩年前，當(dāng)時為了實現(xiàn)批量管理SSH賬號密碼并實現(xiàn)自動巡檢功能，寫過一個簡單的命令行工具，通過使用JSON實現(xiàn)對特定主機賬號密碼與組的管理，如下代碼，通過定義AdminDataBase()類，傳如數(shù)據(jù)庫文件名database.json實現(xiàn)對特定JSON文件的增刪改查功能，在編寫該案例后我對JSON的使用變得更加深刻了。
# -*- coding: utf-8 -*-
import os,json
class AdminDataBase(object):
def __init__(self, database_path):
self.database_path = database_path
# 判斷數(shù)據(jù)庫文件是否存在,不存則則創(chuàng)建一個.
def InitDatabase(self):
if os.path.exists(self.database_path) != None :
init_database = \
{
"HostList": [["1000", "127.0.0.1", "username", "password", "22"]],
"HostGroup": [{"DefaultGroup": ["1000"]}, ],
}
with open(self.database_path, "w", encoding="utf-8") as fp:
fp.write(json.dumps(init_database))
print("[+] {} 結(jié)構(gòu)已被初始化.".format(self.database_path))
# table 接收一個列表名,根據(jù)列表名輸出列表中的數(shù)據(jù)
def ShowHostList(self):
with open(self.database_path, "r", encoding="utf-8") as Read_Pointer:
load_json = json.loads(Read_Pointer.read())
base = load_json.get("HostList")
print("-" * 80)
print("UUID \t 主機地址 \t\t\t 登錄用戶 \t\t 登錄密碼 \t 端口")
print("-" * 80)
for each in range(0, len(base)):
print("{0:4} \t {1:15} \t {2:10} \t {3:10} \t {4:10}"
.format(base[each][0], base[each][1], base[each][2], base[each][3], base[each][4]))
print()
# 在原來的基礎(chǔ)上添加一臺新的主機,添加到HostList表中
def AddHost(self, uuid, address, username, password, port):
with open(self.database_path, "r", encoding="utf-8") as Read_Pointer:
# 讀取 database.json 中的數(shù)據(jù)到內(nèi)存中
load_json = json.loads(Read_Pointer.read())
# 先讀入內(nèi)存,修改后全部替換進(jìn)去
host_list = load_json.get("HostList")
# 獲取到最后一次循環(huán)的元素列表
for each in range(len(host_list) - 1, len(host_list)):
# 判斷如果UUID不重復(fù)則添加,否則跳過添加
if (host_list[each][0] != uuid):
host_list.append([uuid, address, username, password, port])
load_json["HostList"] = host_list
# 最后再次將改好的數(shù)據(jù),回寫到文件保存
with open(self.database_path, "w", encoding="utf-8") as Write_Pointer:
dump_json = json.dumps(load_json)
Write_Pointer.write(dump_json)
print("[+] UUID {} 添加成功.".format(uuid))
else:
print("[-] UUID {} 列表中存在,添加失敗.".format(uuid))
return 0
# 根據(jù)傳入UUID號修改特定主機數(shù)據(jù)
def ModifyHost(self,uuid,modify_address,modify_username,modify_password,modify_port):
with open(self.database_path, "r", encoding="utf-8") as Read_Pointer:
# 讀取 database.json 中的數(shù)據(jù)到內(nèi)存中
load_json = json.loads(Read_Pointer.read())
host_list = load_json.get("HostList")
# 根據(jù)uuid尋找一維數(shù)組的下標(biāo)位置
index = -1
for index, value in enumerate(host_list):
if value[0] == uuid:
print("[*] 已找到UUID {} 所在下標(biāo)為 {}".format(uuid,index))
break
else:
index = -1
# 判斷是否找到了,找到了則修改
if index != -1:
# 修改指定下標(biāo)數(shù)值,并將修改后的數(shù)據(jù)放入原始JSON文件中
host_list[index] = [uuid,modify_address,modify_username,modify_password,modify_port]
load_json["HostList"] = host_list
# 最后再次將改好的數(shù)據(jù),回寫到文件保存
with open(self.database_path, "w", encoding="utf-8") as Write_Pointer:
dump_json = json.dumps(load_json)
Write_Pointer.write(dump_json)
print("[+] UUID {} 修改完成.".format(uuid))
return 0
else:
print("[-] UUID {} 不存在.".format(uuid))
return 0
# 根據(jù)傳入的UUID號刪除主機數(shù)據(jù),先刪除所對的組中的數(shù)據(jù),然后在刪除主機數(shù)據(jù)
def DeleteHost(self,uuid):
with open(self.database_path, "r", encoding="utf-8") as Read_Pointer:
load_json = json.loads(Read_Pointer.read())
host_group = load_json.get("HostGroup")
# 首先在HostGroup中尋找并彈出相應(yīng)UUID
for each in range(0,len(host_group)):
try:
# 循環(huán)每個字典
for k,v in host_group[each].items():
# 判斷UUID是否存在于每個字典中
if v.count(uuid) != 0:
# 移除并放入原始JSON中
v.remove(uuid)
load_json["HostGroup"] = v
# 最后再次將改好的數(shù)據(jù),回寫到文件保存
with open(self.database_path, "w", encoding="utf-8") as Write_Pointer:
dump_json = json.dumps(load_json)
Write_Pointer.write(dump_json)
#print("[+] UUID {} 修改完成.".format(uuid))
else:
#print("[-] 當(dāng)前組 {} 不能存在: {}".format(k,uuid))
break
except Exception:
print("[-] 當(dāng)前組不能存在: {}".format(uuid))
return 0
# ----------------------------------------------
# 其次尋找HostList中的數(shù)據(jù)并彈出
host_list = load_json.get("HostList")
try:
# 根據(jù)uuid尋找一維數(shù)組的下標(biāo)位置
index = -1
for index, value in enumerate(host_list):
if value[0] == uuid:
#print("[*] 已找到UUID {} 所在下標(biāo)為 {}".format(uuid,index))
break
else:
index = -1
# 如果找到了,則直接根據(jù)下標(biāo)彈出數(shù)據(jù)
if index != -1:
host_list.pop(index)
load_json["HostList"] = host_list
# 最后再次將改好的數(shù)據(jù),回寫到文件保存
with open(self.database_path, "w", encoding="utf-8") as Write_Pointer:
dump_json = json.dumps(load_json)
Write_Pointer.write(dump_json)
print("[+] UUID {} 主機: {} 已被移除.".format(uuid,host_list[index][1]))
return 0
except Exception:
return 0
# 向數(shù)據(jù)庫中的HostGroup字段中添加一個主機組
def AddHostGroup(self,add_group_name):
with open(self.database_path, "r", encoding="utf-8") as Read_Pointer:
load_json = json.loads( Read_Pointer.read() )
group_obj = load_json.get("HostGroup")
# 循環(huán)所有字典中的組
for each in range(0, len(group_obj)):
list_name = str(list(group_obj[each].keys())[0])
# print("組節(jié)點: {}".format(list_name))
# 循環(huán)判斷表中是否存在指定的組名稱
if (list_name == add_group_name):
print("[-] {} 存在于組中,無法繼續(xù)添加.".format(list_name))
return 0
# 讀取原始JSON文件,并構(gòu)建回寫參數(shù)
tmp = {}
tmp[add_group_name] = ["1000"]
group_obj.append(tmp)
# 回寫添加主機組
with open(self.database_path, "w", encoding="utf-8") as Write_Pointer:
dump_json = json.dumps(load_json)
Write_Pointer.write(dump_json)
print("[+] 主機組 {} 已添加".format(add_group_name))
# 在主機組中刪除一個主機組
def DeleteHostGroup(self,delete_group_name):
with open(self.database_path, "r", encoding="utf-8") as Read_Pointer:
load_json = json.loads( Read_Pointer.read() )
group_obj = load_json.get("HostGroup")
# 循環(huán)所有字典中的組
for each in range(0, len(group_obj)):
list_name = str(list(group_obj[each].keys())[0])
# 循環(huán)判斷表中是否存在指定的組名稱
if (list_name == delete_group_name):
# 如果存在于組中,那我們就把他的each彈出列表
group_obj.pop(each)
# 將彈出后的列表再次寫回JSON序列中
with open(self.database_path, "w", encoding="utf-8") as Write_Pointer:
dump_json = json.dumps(load_json)
Write_Pointer.write(dump_json)
print("[-] 主機組 {} 已移除.".format(delete_group_name))
return 0
print("[-] 主機組 {} 不存在.".format(delete_group_name))
return 0
# 向指定主機組中添加一個主機UUID
def AddHostGroupOnUUID(self,group_name, uuid):
with open(self.database_path, "r", encoding="utf-8") as Read_Pointer:
load_json = json.loads( Read_Pointer.read() )
group_obj = load_json.get("HostGroup")
# 循環(huán)所有字典中的所有組
for each in range(0, len(group_obj)):
list_name = str(list(group_obj[each].keys())[0])
# 如果找到了需要添加的組
if (list_name == group_name):
tmp = group_obj[each]
# 獲取到原始列表中的數(shù)據(jù)
val = list(tmp.values())[0]
# 判斷輸入的UUID號,是否在val列表中,不在則添加
if str(uuid) not in val:
val.append(str(uuid))
# 將原始數(shù)據(jù)賦值到新的表中
group_obj[each][list_name] = val
else:
print("[-] UUID {} 已存在于 {} 主機組,添加失敗.".format(uuid,group_name))
return 0
with open(self.database_path, "w", encoding="utf-8") as Write_Pointer:
dump_json = json.dumps(load_json)
Write_Pointer.write(dump_json)
print("[+] 向主機組 {} 增加UUID {} 完成".format(group_name, uuid))
return 0
# 從指定主機組中刪除一個指定的UUID號
def DeleteHostGroupOnUUID(self,group_name, uuid):
with open(self.database_path, "r", encoding="utf-8") as Read_Pointer:
load_json =json.loads( Read_Pointer.read() )
group_obj = load_json.get("HostGroup")
# 循環(huán)所有字典中的組
for each in range(0, len(group_obj)):
list_name = str(list(group_obj[each].keys())[0])
# 先來尋找到需要刪除的主機組
if (list_name == group_name):
tmp = group_obj[each]
# 尋找指定組中的指定列表
val = list(tmp.values())[0]
for x in range(0, len(val)):
if (val[x] == uuid):
print("[*] 搜索UUID: {} 索引值: {}".format(val[x], x))
# 將要刪除的UUID彈出列表
val.pop(x)
# 將彈出后的列表重新復(fù)制到主機組中
group_obj[each][list_name] = val
# 保存彈出后的列表到序列中
with open(self.database_path, "w", encoding="utf-8") as write_fp:
dump_json = json.dumps(load_json)
write_fp.write(dump_json)
print("[+] 從主機組 {} 彈出UUID {} 完成".format(group_name, uuid))
return 0
return 0
# 輸出所有主機組和全部節(jié)點信息
def ShowAllGroup(self):
with open(self.database_path, "r", encoding="utf-8") as Read_Pointer:
load_json = json.loads( Read_Pointer.read() )
group_obj = load_json.get("HostGroup")
# 循環(huán)解析所有組,并解析出UUID所對應(yīng)的主機地址等信息
for each in range(0, len(group_obj)):
for k, v in group_obj[each].items():
print("-" * 140)
print("主機組: {}".format(k))
print("-" * 140)
for each in range(0, len(v)):
# print("--> UUID: {}".format(v[each]))
# 循環(huán)判斷,拿著組內(nèi)UUID判斷是否存在,如果存在則打印出主機詳細(xì)信息
base_obj = load_json.get("HostList")
for x in range(0, len(base_obj)):
if (v[each] == base_obj[x][0]):
print("UUID: {0:6} \t 主機地址: {1:15} \t 主機賬號: {2:15} \t 主機密碼: {3:15} \t 端口: {4:5} \t".
format(base_obj[x][0], base_obj[x][1], base_obj[x][2], base_obj[x][3],
base_obj[x][4]))
print("\n")
# 只顯示所有的主機組,包括組中主機數(shù)
def ShowGroup(self):
with open(self.database_path, "r", encoding="utf-8") as Read_Pointer:
load_json = json.loads( Read_Pointer.read() )
group_obj = load_json.get("HostGroup")
print("-" * 80)
print("{0:20} \t {1:5} \t {2:100}".format("主機組名","主機數(shù)","主機列表"))
print("-" * 80)
for each in range(0,len(group_obj)):
for k,v in group_obj[each].items():
print("{0:20} \t {1:5} \t {2:100}".format(k,str(len(v)), str(v)))
print()
return 0
# 對指定的主機組進(jìn)行Ping測試
def PingGroup(self,group_name):
with open(self.database_path, "r", encoding="utf-8") as Read_Pointer:
load_json = json.loads( Read_Pointer.read() )
group_obj = load_json.get("HostGroup")
# 循環(huán)遍歷主機列表
for each in range(0, len(group_obj)):
for k, v in group_obj[each].items():
# 尋找主機組,找到后提取出所有的Value
if (k == group_name):
item_list = v
# 再循環(huán),拿著v的值對base進(jìn)行循環(huán),提取出其中的用戶名密碼等
for val in range(0, len(item_list)):
# 得到循環(huán)列表：print(item_list[val])
base_obj = load_json.get("HostList")
# 循環(huán)base表中的計數(shù)器
for base_count in range(0, len(base_obj)):
# 判斷base表中是否存在指定UUID,如果存在則輸出其值
if (base_obj[base_count][0] == item_list[val]):
print("地址: {} \t 用戶名: {} [ok]".format(base_obj[base_count][1],base_obj[base_count][2]))
if __name__ == "__main__":
db = AdminDataBase("database.json")
while True:
try:
cmd = str(input("[LyShell] # ")).split()
if (cmd == ""):
continue
elif (cmd[0] == "exit"):
exit(1)
elif (cmd[0] == "clear"):
os.system("cls")
elif(cmd[0] == "Init"):
db.InitDatabase()
elif(cmd[0] == "ShowHostList"):
db.ShowHostList()
elif(cmd[0] == "ShowGroup"):
db.ShowGroup()
elif(cmd[0] == "ShowAllGroup"):
db.ShowAllGroup()
elif(cmd[0] == "AddHost" or cmd[0] == "ModifyHost"):
if (len(cmd) - 1 >= 5):
uuid = str(cmd[1]).split("=")[1]
address = str(cmd[2]).split("=")[1]
username = str(cmd[3]).split("=")[1]
password = str(cmd[4]).split("=")[1]
port = str(cmd[5]).split("=")[1]
if cmd[0] == "AddHost":
db.AddHost(uuid,address,username,password,port)
elif cmd[0] == "ModifyHost":
db.ModifyHost(uuid,address,username,password,port)
elif(cmd[0] == "DeleteHost"):
if(len(cmd)-1 >= 1):
uuid = str(cmd[1]).split("=")[1]
db.DeleteHost(uuid)
elif(cmd[0] == "AddHostGroup"):
group_name = str(cmd[1]).split("=")[1]
db.AddHostGroup(group_name)
elif(cmd[0] == "DeleteHostGroup"):
group_name = str(cmd[1]).split("=")[1]
db.DeleteHostGroup(group_name)
elif(cmd[0] == "AddHostGroupOnUUID"):
group_name = str(cmd[1]).split("=")[1]
uuid = str(cmd[2]).split("=")[1]
db.AddHostGroupOnUUID(group_name,uuid)
elif(cmd[0] == "DelHostGroupOnUUID"):
group_name = str(cmd[1]).split("=")[1]
uuid = str(cmd[2]).split("=")[1]
db.DeleteHostGroupOnUUID(group_name,uuid)
elif(cmd[0] == "PingGroup"):
group_name = str(cmd[1]).split("=")[1]
db.PingGroup(group_name)
elif (cmd[0] == "help"):
print("By: LyShark")
else:
print("Not Command")
except Exception:
continue
使用案例
管理數(shù)據(jù)庫（AdminDataBase）中的主機信息和主機分組信息。用戶通過輸入命令來執(zhí)行不同的操作，如初始化數(shù)據(jù)庫、顯示主機列表、添加主機、修改主機信息、刪除主機等。
以下是代碼的主要功能和命令列表：
- 初始化數(shù)據(jù)庫：
Init - 顯示主機列表：
ShowHostList - 顯示主機分組：
ShowGroup - 顯示所有主機分組：
ShowAllGroup - 添加主機：
AddHost - 修改主機信息：
ModifyHost - 刪除主機：
DeleteHost - 添加主機分組：
AddHostGroup - 刪除主機分組：
DeleteHostGroup - 將主機添加到指定分組：
AddHostGroupOnUUID - 從指定分組刪除主機：
DelHostGroupOnUUID - 按組執(zhí)行 Ping 操作：
PingGroup - 清屏：
clear - 退出程序：
exit - 幫助：
help
Init
初次使用需要執(zhí)行本命令，對數(shù)據(jù)庫文件進(jìn)行寫出，如下所示；
ShowHostList
用于輸出當(dāng)前主機列表信息，如下圖所示；
ShowGroup
用于輸出當(dāng)前主機組，如下圖所示；
ShowAllGroup
用于輸出所有的主機組以及組內(nèi)的主機詳細(xì)信息，如下圖所示；
AddHost
添加一個新的主機記錄，如下圖所示；
ModifyHost
修改一個已有的主機記錄，此處以UUID作為修改條件，如下圖所示；
DeleteHost
根據(jù)一個UUID唯一標(biāo)識，刪除一個已存在的主機記錄，如下圖所示；
AddHostGroup
新增一個組名，默認(rèn)會攜帶1000為初始主機，如下圖所示；
DeleteHostGroup
刪除一整個主機組，如下圖所示；
AddHostGroupOnUUID
根據(jù)UUID號將特定主機添加到特定組內(nèi)，如下圖所示；
DelHostGroupOnUUID
根據(jù)主機組名，刪除特定的UUID，如下圖所示；
PingGroup
對特定主機組執(zhí)行Ping功能測試，此處可以擴展，如下圖所示；
總結(jié)
該案例只是用于學(xué)習(xí)如何靈活運用JSON實現(xiàn)數(shù)據(jù)的增刪改查，其實在實戰(zhàn)中意義不大，因為完全可以使用SQLite這類精簡數(shù)據(jù)庫，此案例只是本人為了熟悉JSON的增刪改查而寫的一個Demo工具。
以上就是Python利用解析JSON實現(xiàn)主機管理的詳細(xì)內(nèi)容，更多關(guān)于Python解析JSON的資料請關(guān)注腳本之家其它相關(guān)文章！
相關(guān)文章
PyQt 5 設(shè)置Logo圖標(biāo)和Title標(biāo)題的操作
這篇文章主要介紹了PyQt 5 設(shè)置Logo圖標(biāo)和Title標(biāo)題的操作，具有很好的參考價值，希望對大家有所幫助。一起跟隨小編過來看看吧2021-03-03
在Python中使用filter去除列表中值為假及空字符串的例子
今天小編就為大家分享一篇在Python中使用filter去除列表中值為假及空字符串的例子，具有很好的參考價值，希望對大家有所幫助。一起跟隨小編過來看看吧2019-11-11
Python實現(xiàn)檢索指定網(wǎng)段內(nèi)所有的數(shù)據(jù)庫服務(wù)器
這篇文章主要為大家詳細(xì)介紹了如何使用Python實現(xiàn)檢索指定網(wǎng)段內(nèi)所有的數(shù)據(jù)庫服務(wù)器,文中的示例代碼講解詳細(xì),感興趣的小伙伴可以了解下2025-02-02
Python中日志模塊logging的使用技巧和應(yīng)用詳解
在Python開發(fā)中,日志記錄是一個非常重要的環(huán)節(jié),它不僅有助于開發(fā)者追蹤程序的執(zhí)行流程,還能在出現(xiàn)問題時提供關(guān)鍵信息,幫助快速定位并解決問題,本文將結(jié)合實際案例,詳細(xì)介紹logging模塊的基礎(chǔ)用法和高級特性,需要的朋友可以參考下2024-08-08
這篇文章主要介紹了Python對象的list如何快速按照屬性查找問題,具有很好的參考價值,希望對大家有所幫助,如有錯誤或未考慮完全的地方,望不吝賜教2023-11-11
Python利用psutil實現(xiàn)獲取硬件,網(wǎng)絡(luò)和進(jìn)程信息
Python?有一個第三方模塊叫?psutil，專門用來獲取操作系統(tǒng)以及硬件相關(guān)的信息，比如：CPU、磁盤、網(wǎng)絡(luò)、內(nèi)存等等。下面來看一下它的用法2022-07-07
一小時學(xué)會TensorFlow2之基本操作2實例代碼
這篇文章主要介紹了TensorFlow2的基本操作和實例代碼，本文給大家介紹的非常詳細(xì)，對大家的學(xué)習(xí)或工作具有一定的參考借鑒價值，需要的朋友可以參考下2021-09-09
這篇文章主要介紹了關(guān)于Python如何安裝requests庫,requests庫自稱“HTTP for Humans”,直譯過來的意思是專門為人類設(shè)計的HTTP庫,能夠被開發(fā)人員安全地使用,需要的朋友可以參考下2023-04-04
今天小編就為大家分享一篇Python刪除n行后的其他行方法，具有很好的參考價值，希望對大家有所幫助。一起跟隨小編過來看看吧2019-01-01
PyTorch提供了一個非常方便的函數(shù)flatten()來完成這個任務(wù),本文將介紹Pytorch閱讀文檔中的flatten函數(shù),并提供一些示例代碼,感興趣的朋友一起看看吧2023-11-11