JSON(JavaScript Object Notation)是一種輕量級的數據交換格式,它以易於閱讀和編寫的文本形式表示數據。JSON 是一種獨立於編程語言的數據格式,因此在不同的編程語言中都有對應的解析器和生成器。JSON 格式的設計目標是易於理解、支持複雜數據結構和具有良好的可擴展性。 ...
JSON(JavaScript Object Notation)是一種輕量級的數據交換格式,它以易於閱讀和編寫的文本形式表示數據。JSON 是一種獨立於編程語言的數據格式,因此在不同的編程語言中都有對應的解析器和生成器。JSON 格式的設計目標是易於理解、支持複雜數據結構和具有良好的可擴展性。
JSON 數據是以鍵值對的形式存在的,而且易於閱讀和編寫。以下是一個簡單的 JSON 示例:
{
"name": "John Doe",
"age": 30,
"city": "New York",
"isStudent": false,
"grades": [95, 88, 75, 92],
"address": {
"street": "123 Main St",
"zipCode": "10001"
}
}
在這個例子中,JSON 對象包含了一些屬性,包括字元串、數字、布爾值、數組和嵌套的對象。
"name": "John Doe"
:字元串鍵值對。"age": 30
:數字鍵值對。"city": "New York"
:字元串鍵值對。"isStudent": false
:布爾鍵值對。"grades": [95, 88, 75, 92]
:數組鍵值對。"address": {...}
:嵌套對象。
在實際應用中,JSON 數據通常用於前後端之間的數據交換,或者配置文件的存儲。各種編程語言都提供了處理 JSON
數據的庫或模塊。
很早之前大概是兩年前,當時為了實現批量管理SSH
賬號密碼並實現自動巡檢功能,寫過一個簡單的命令行工具,通過使用JSON
實現對特定主機賬號密碼與組的管理,如下代碼,通過定義AdminDataBase()
類,傳如資料庫文件名database.json
實現對特定JSON
文件的增刪改查功能,在編寫該案例後我對JSON
的使用變得更加深刻了。
# -*- coding: utf-8 -*-
import os,json
class AdminDataBase(object):
def __init__(self, database_path):
self.database_path = database_path
# 判斷資料庫文件是否存在,不存則則創建一個.
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("[+] {} 結構已被初始化.".format(self.database_path))
# table 接收一個列表名,根據列表名輸出列表中的數據
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()
# 在原來的基礎上添加一臺新的主機,添加到HostList表中
def AddHost(self, uuid, address, username, password, port):
with open(self.database_path, "r", encoding="utf-8") as Read_Pointer:
# 讀取 database.json 中的數據到記憶體中
load_json = json.loads(Read_Pointer.read())
# 先讀入記憶體,修改後全部替換進去
host_list = load_json.get("HostList")
# 獲取到最後一次迴圈的元素列表
for each in range(len(host_list) - 1, len(host_list)):
# 判斷如果UUID不重覆則添加,否則跳過添加
if (host_list[each][0] != uuid):
host_list.append([uuid, address, username, password, port])
load_json["HostList"] = host_list
# 最後再次將改好的數據,回寫到文件保存
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
# 根據傳入UUID號修改特定主機數據
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 中的數據到記憶體中
load_json = json.loads(Read_Pointer.read())
host_list = load_json.get("HostList")
# 根據uuid尋找一維數組的下標位置
index = -1
for index, value in enumerate(host_list):
if value[0] == uuid:
print("[*] 已找到UUID {} 所在下標為 {}".format(uuid,index))
break
else:
index = -1
# 判斷是否找到了,找到了則修改
if index != -1:
# 修改指定下標數值,並將修改後的數據放入原始JSON文件中
host_list[index] = [uuid,modify_address,modify_username,modify_password,modify_port]
load_json["HostList"] = host_list
# 最後再次將改好的數據,回寫到文件保存
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
# 根據傳入的UUID號刪除主機數據,先刪除所對的組中的數據,然後在刪除主機數據
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中尋找並彈出相應UUID
for each in range(0,len(host_group)):
try:
# 迴圈每個字典
for k,v in host_group[each].items():
# 判斷UUID是否存在於每個字典中
if v.count(uuid) != 0:
# 移除並放入原始JSON中
v.remove(uuid)
load_json["HostGroup"] = v
# 最後再次將改好的數據,回寫到文件保存
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("[-] 當前組 {} 不能存在: {}".format(k,uuid))
break
except Exception:
print("[-] 當前組不能存在: {}".format(uuid))
return 0
# ----------------------------------------------
# 其次尋找HostList中的數據並彈出
host_list = load_json.get("HostList")
try:
# 根據uuid尋找一維數組的下標位置
index = -1
for index, value in enumerate(host_list):
if value[0] == uuid:
#print("[*] 已找到UUID {} 所在下標為 {}".format(uuid,index))
break
else:
index = -1
# 如果找到了,則直接根據下標彈出數據
if index != -1:
host_list.pop(index)
load_json["HostList"] = host_list
# 最後再次將改好的數據,回寫到文件保存
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
# 向資料庫中的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")
# 迴圈所有字典中的組
for each in range(0, len(group_obj)):
list_name = str(list(group_obj[each].keys())[0])
# print("組節點: {}".format(list_name))
# 迴圈判斷表中是否存在指定的組名稱
if (list_name == add_group_name):
print("[-] {} 存在於組中,無法繼續添加.".format(list_name))
return 0
# 讀取原始JSON文件,並構建回寫參數
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")
# 迴圈所有字典中的組
for each in range(0, len(group_obj)):
list_name = str(list(group_obj[each].keys())[0])
# 迴圈判斷表中是否存在指定的組名稱
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")
# 迴圈所有字典中的所有組
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]
# 判斷輸入的UUID號,是否在val列表中,不在則添加
if str(uuid) not in val:
val.append(str(uuid))
# 將原始數據賦值到新的表中
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")
# 迴圈所有字典中的組
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)
# 將彈出後的列表重新複製到主機組中
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
# 輸出所有主機組和全部節點信息
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")
# 迴圈解析所有組,並解析出UUID所對應的主機地址等信息
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]))
# 迴圈判斷,拿著組內UUID判斷是否存在,如果存在則列印出主機詳細信息
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")
# 只顯示所有的主機組,包括組中主機數
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("主機組名","主機數","主機列表"))
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
# 對指定的主機組進行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")
# 迴圈遍歷主機列表
for each in range(0, len(group_obj)):
for k, v in group_obj[each].items():
# 尋找主機組,找到後提取出所有的Value
if (k == group_name):
item_list = v
# 再迴圈,拿著v的值對base進行迴圈,提取出其中的用戶名密碼等
for val in range(0, len(item_list)):
# 得到迴圈列表:print(item_list[val])
base_obj = load_json.get("HostList")
# 迴圈base表中的計數器
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
使用案例
管理資料庫(AdminDataBase
)中的主機信息和主機分組信息。用戶通過輸入命令來執行不同的操作,如初始化資料庫、顯示主機列表、添加主機、修改主機信息、刪除主機等。
以下是代碼的主要功能和命令列表:
- 初始化資料庫:
Init
- 顯示主機列表:
ShowHostList
- 顯示主機分組:
ShowGroup
- 顯示所有主機分組:
ShowAllGroup
- 添加主機:
AddHost
- 修改主機信息:
ModifyHost
- 刪除主機:
DeleteHost
- 添加主機分組:
AddHostGroup
- 刪除主機分組:
DeleteHostGroup
- 將主機添加到指定分組:
AddHostGroupOnUUID
- 從指定分組刪除主機:
DelHostGroupOnUUID
- 按組執行 Ping 操作:
PingGroup
- 清屏:
clear
- 退出程式:
exit
- 幫助:
help
Init
初次使用需要執行本命令,對資料庫文件進行寫出,如下所示;
ShowHostList
用於輸出當前主機列表信息,如下圖所示;
ShowGroup
用於輸出當前主機組,如下圖所示;
ShowAllGroup
用於輸出所有的主機組以及組內的主機詳細信息,如下圖所示;
AddHost
添加一個新的主機記錄,如下圖所示;
ModifyHost
修改一個已有的主機記錄,此處以UUID作為修改條件,如下圖所示;
DeleteHost
根據一個UUID唯一標識,刪除一個已存在的主機記錄,如下圖所示;
AddHostGroup
新增一個組名,預設會攜帶1000為初始主機,如下圖所示;
DeleteHostGroup
刪除一整個主機組,如下圖所示;
AddHostGroupOnUUID
根據UUID號將特定主機添加到特定組內,如下圖所示;
DelHostGroupOnUUID
根據主機組名,刪除特定的UUID,如下圖所示;
PingGroup
對特定主機組執行Ping功能測試,此處可以擴展,如下圖所示;
總結部分
該案例只是用於學習如何靈活運用JSON實現數據的增刪改查,其實在實戰中意義不大,因為完全可以使用SQLite這類精簡資料庫,此案例只是本人為了熟悉JSON的增刪改查而寫的一個Demo工具。
文章出處:https://www.cnblogs.com/LyShark/p/17875693.html本博客所有文章除特別聲明外,均採用 BY-NC-SA 許可協議。轉載請註明出處!