一、grafana 批量视图备份
import requests
import json
import urllib3
import osfrom requests.auth import HTTPBasicAuthfilename_folders_map = "folders_map.json"
type_folder = "dash-folder"
type_dashboard = "dash-db"# Grafana服务器地址及API密钥
grafana_url = "http://127.0.0.1:30301/api"
api_key = "YOUR_API_KEY"
username = "admin"
passwd = "admin123"
requests.packages.urllib3.disable_warnings()
new_grafana_url = "https://127.0.0.1:30301/api"
new_es_url = "https://quickstart-es-data-nodes.es8:9200"
new_es_passwd = "new_es_passwd "headers = {'Content-Type': 'application/json'
}# 获取数据源列表
def get_datasources():response = requests.get(f'{grafana_url}/datasources', headers=headers, auth=HTTPBasicAuth(username, passwd),verify=False)if response.status_code == 200:print(json.loads(response.text))return json.loads(response.text)else:print("Failed to retrieve dashboards.")def save_datasource_detail(datasource_id):response = requests.get(f'{grafana_url}/datasources/{datasource_id}', headers=headers,auth=HTTPBasicAuth(username, passwd), verify=False)if response.status_code == 200:ret_json = json.loads(response.text)print(ret_json)name = ret_json["name"]# 打开文件（若不存在则创建）并写入JSON数据with open("datasource_" + name + ".json", "w") as file:# 使用json.dump()函数将数据转换为JSON格式后写入文件json.dump(ret_json, file)return ret_jsonelse:print("Failed to retrieve dashboards.")def set_datasource_detail(datasource_json, password="123"):# del datasource_json["uid"]# datasource_json["name"] = datasource_json["name"] + "03"datasource_json["secureJsonData"] = {'basicAuthPassword': password}new_json_obj = json.dumps(datasource_json)print(new_json_obj)response = requests.post(f'{new_grafana_url}/datasources', data=new_json_obj, headers=headers,auth=HTTPBasicAuth(username, passwd),verify=False)if response.status_code == 200:print(json.loads(response.text))return json.loads(response.text)else:print("Failed to retrieve dashboards.")def set_datasource_detail(datasource_json, es_url=new_es_url,password=new_es_passwd):# del datasource_json["uid"]# datasource_json["name"] = datasource_json["name"] + "03"if "elasticsearch" == datasource_json["type"]:datasource_json["url"] = new_es_urldatasource_json["secureJsonData"] = {'basicAuthPassword': password}new_json_obj = json.dumps(datasource_json)print(new_json_obj)response = requests.post(f'{new_grafana_url}/datasources', data=new_json_obj, headers=headers,auth=HTTPBasicAuth(username, passwd),verify=False)if response.status_code == 200:print(json.loads(response.text))return json.loads(response.text)else:print("Failed to retrieve dashboards.")def del_datasource_detail(datasource_id):response = requests.delete(f'{new_grafana_url}/datasources/{datasource_id}', headers=headers,auth=HTTPBasicAuth(username, passwd), verify=False)if response.status_code == 200:ret_json = json.loads(response.text)return ret_jsonelse:print("Failed to retrieve dashboards.")# 获文件夹列表
def get_folders(url):response = requests.get(f'{url}/folders', headers=headers, auth=HTTPBasicAuth(username, passwd),verify=False)if response.status_code == 200:return json.loads(response.text)else:print("Failed to retrieve dashboards.")def save_folder_detail(folder_uid):response = requests.get(f'{grafana_url}/folders/{folder_uid}', headers=headers,auth=HTTPBasicAuth(username, passwd), verify=False)if response.status_code == 200:ret_json = json.loads(response.text)print(ret_json)name = ret_json["title"]# 打开文件（若不存在则创建）并写入JSON数据with open("folder_" + name + ".json", "w") as file:# 使用json.dump()函数将数据转换为JSON格式后写入文件json.dump(ret_json, file)return ret_jsonelse:print("Failed to retrieve dashboards.")def set_folder_detail(folder_json):# datasource_json["name"] = datasource_json["name"] + "03"new_json_obj = {"title": folder_json["title"], "uid": folder_json["uid"]}json_obj = json.dumps(new_json_obj)print(new_json_obj)response = requests.post(f'{new_grafana_url}/folders', data=json_obj, headers=headers,auth=HTTPBasicAuth(username, passwd),verify=False)if response.status_code == 200:print(json.loads(response.text))return json.loads(response.text)else:print("Failed to retrieve dashboards.")# 获取所有仪表盘列表
def get_dashboards():response = requests.get(f'{grafana_url}/search', headers=headers, auth=HTTPBasicAuth(username, passwd),verify=False)if response.status_code == 200:print(json.loads(response.text))return json.loads(response.text)else:print("Failed to retrieve dashboards.")def save_dashboard_detail(dashboard_id , folder_id):response = requests.get(f'{grafana_url}/dashboards/uid/{dashboard_id}', headers=headers,auth=HTTPBasicAuth(username, passwd), verify=False)if response.status_code == 200:ret_json = json.loads(response.text)print(ret_json)dashboard = ret_json["dashboard"]name = dashboard["title"]if folder_id == None:# 打开文件（若不存在则创建）并写入JSON数据with open("dashboard_" + name + ".json", "w") as file:# 使用json.dump()函数将数据转换为JSON格式后写入文件json.dump(ret_json, file)else:with open("dashboard_folder" + str(folder_id) + "_" + name + ".json", "w") as file:# 使用json.dump()函数将数据转换为JSON格式后写入文件json.dump(ret_json, file)return ret_jsonelse:print("Failed to retrieve dashboards.")def set_dashboard_detail(dashboard_json):folderUid = dashboard_json["meta"]["folderUid"]del dashboard_json["meta"]#dashboard = file_json["dashboard"]dashboard_json["folderUid"] = folderUiddashboard_json["overwrite"] = Truedashboard_json["dashboard"]["id"] = NonedashboardJson = json.dumps(dashboard_json)print(f"{dashboardJson}")response = requests.post(f'{new_grafana_url}/dashboards/import', data=dashboardJson, headers=headers,auth=HTTPBasicAuth(username, passwd),verify=False)if response.status_code == 200:print(json.loads(response.text))return json.loads(response.text)else:print("Failed to retrieve dashboards.")# 创建新的仪表盘
def create_new_dashboard(title):data = {"overwrite": False,"dashboard": {"id": None,"uid": "","title": title,# ...其他配置项...},"folderId": 0,"message": ""}headers = {'Authorization': f'Bearer {api_key}','Content-Type': 'application/json'}response = requests.post(f'{grafana_url}/dashboards/db', json=data, headers=headers)if response.status_code == 200:return response.json().get('slug')else:print("Failed to create new dashboard.")# 更新现有仪表盘
def update_existing_dashboard(dashboard_slug, title):data = {"overwrite": True,"dashboard": {"id": None,"uid": "","title": title,# ...其他配置项...},"folderId": 0,"message": ""}headers = {'Authorization': f'Bearer {api_key}','Content-Type': 'application/json'}response = requests.put(f'{grafana_url}/dashboards/{dashboard_slug}', json=data, headers=headers)if response.status_code == 200:return response.json().get('slug')else:print("Failed to update existing dashboard.")# 删除指定仪表盘
def delete_dashboard(dashboard_slug):headers = {'Authorization': f'Bearer {api_key}'}response = requests.delete(f'{grafana_url}/dashboards/{dashboard_slug}', headers=headers)if response.status_code == 200:print("Dashboard deleted successfully.")else:print("Failed to delete the dashboard.")# 测试函数
if __name__ == "__main__":# # 清空所有数据源# datasources = get_datasources()# for datasource in datasources:# # print(datasource["id"])# datasource_detail = del_datasource_detail(datasource["id"])# print(datasource_detail)current_dir = os.getcwd()file_names = os.listdir(current_dir)# 获取所有数据源并保存到磁盘datasources = get_datasources()for datasource in datasources:# print(datasource["id"])datasource_detail = save_datasource_detail(datasource["id"])print(datasource_detail)# 数据源导入# for file_name in file_names:# if file_name.startswith("datasource") & file_name.endswith(".json"):# print(file_name)## with open(file_name, 'r') as f:# file_json = json.load(f)# ret = set_datasource_detail(file_json, new_es_url,new_es_passwd)# 获取所有文件夹并保存到磁盘folders = get_folders(grafana_url)for folder in folders:print(folder["title"])folder_detail = save_folder_detail(folder["uid"])print(folder_detail)# 文件夹导入# for file_name in file_names:# if file_name.startswith("folder") & file_name.endswith(".json"):# print(file_name)# with open(file_name, 'r') as f:# file_json = json.load(f)# ret = set_folder_detail(file_json)# 获取所有仪表盘列表dashboards = get_dashboards()for board in dashboards:# if type_folder==board["type"]:# print(board["title"])if type_dashboard == board["type"]:print(board["title"])save_dashboard_detail(board["uid"],board["folderId"])# 仪表盘导入# for file_name in file_names:# if file_name.startswith("dashboard") & file_name.endswith(".json"):# print(file_name)# with open(file_name, 'r') as f:# file_json = json.load(f)# ret = set_dashboard_detail(file_json)# # 创建新的仪表盘# new_dashboard_slug = create_new_dashboard("New Dashboard")# print(f"Created a new dashboard with slug: {new_dashboard_slug}")## # 更新现有仪表盘# updated_dashboard_slug = update_existing_dashboard(new_dashboard_slug, "Updated Dashboard Title")# print(f"Updated an existing dashboard with slug: {updated_dashboard_slug}")## # 删除指定仪表盘# delete_dashboard(updated_dashboard_slug)
二、grafana 批量视图恢复
import requests
import json
import urllib3
import osfrom requests.auth import HTTPBasicAuthfilename_folders_map = "folders_map.json"
type_folder = "dash-folder"
type_dashboard = "dash-db"# Grafana服务器地址及API密钥
grafana_url = "https://127.0.0.1:30301/api"
api_key = "YOUR_API_KEY"
username = "admin"
passwd = "admin123"
requests.packages.urllib3.disable_warnings()
# new_grafana_url = "https://127.0.0.2:30301/api"
# new_es_url = "https://quickstart-es-data-nodes.es8:9200"
# new_es_passwd = "new_es_passwd "
new_grafana_url = "https://127.0.0.1:30301/api"
new_es_url = "https://quickstart-es-data-nodes.es8:9200"
new_es_passwd = "new_es_passwd "old_domain = 'traefik.kidsi4.cn'
new_domain = 'web.bcs2sz.com'headers = {'Content-Type': 'application/json'
}# 获取数据源列表
def get_datasources():response = requests.get(f'{new_grafana_url}/datasources', headers=headers, auth=HTTPBasicAuth(username, passwd),verify=False)if response.status_code == 200:print(json.loads(response.text))return json.loads(response.text)else:print("Failed to retrieve dashboards.")def save_datasource_detail(datasource_id):response = requests.get(f'{grafana_url}/datasources/{datasource_id}', headers=headers,auth=HTTPBasicAuth(username, passwd), verify=False)if response.status_code == 200:ret_json = json.loads(response.text)print(ret_json)name = ret_json["name"]# 打开文件（若不存在则创建）并写入JSON数据with open("datasource_" + name + ".json", "w") as file:# 使用json.dump()函数将数据转换为JSON格式后写入文件json.dump(ret_json, file)return ret_jsonelse:print("Failed to retrieve dashboards.")def set_datasource_detail(datasource_json, password="123"):# del datasource_json["uid"]# datasource_json["name"] = datasource_json["name"] + "03"datasource_json["secureJsonData"] = {'basicAuthPassword': password}new_json_obj = json.dumps(datasource_json)print(new_json_obj)response = requests.post(f'{new_grafana_url}/datasources', data=new_json_obj, headers=headers,auth=HTTPBasicAuth(username, passwd),verify=False)if response.status_code == 200:print(json.loads(response.text))return json.loads(response.text)else:print("Failed to retrieve dashboards.")def set_datasource_detail(datasource_json, es_url=new_es_url,password=new_es_passwd):# del datasource_json["uid"]# datasource_json["name"] = datasource_json["name"] + "03"if "elasticsearch" == datasource_json["type"]:datasource_json["url"] = new_es_urldatasource_json["secureJsonData"] = {'basicAuthPassword': password}new_json_obj = json.dumps(datasource_json)print(new_json_obj)response = requests.post(f'{new_grafana_url}/datasources', data=new_json_obj, headers=headers,auth=HTTPBasicAuth(username, passwd),verify=False)if response.status_code == 200:print(json.loads(response.text))return json.loads(response.text)else:print("Failed to retrieve dashboards.")def del_datasource_detail(datasource_id):response = requests.delete(f'{new_grafana_url}/datasources/{datasource_id}', headers=headers,auth=HTTPBasicAuth(username, passwd), verify=False)if response.status_code == 200:ret_json = json.loads(response.text)return ret_jsonelse:print("Failed to retrieve dashboards.")# 获文件夹列表
def get_folders(url):response = requests.get(f'{url}/folders', headers=headers, auth=HTTPBasicAuth(username, passwd),verify=False)if response.status_code == 200:return json.loads(response.text)else:print("Failed to retrieve dashboards.")def save_folder_detail(folder_uid):response = requests.get(f'{grafana_url}/folders/{folder_uid}', headers=headers,auth=HTTPBasicAuth(username, passwd), verify=False)if response.status_code == 200:ret_json = json.loads(response.text)print(ret_json)name = ret_json["title"]# 打开文件（若不存在则创建）并写入JSON数据with open("folder_" + name + ".json", "w") as file:# 使用json.dump()函数将数据转换为JSON格式后写入文件json.dump(ret_json, file)return ret_jsonelse:print("Failed to retrieve dashboards.")def set_folder_detail(folder_json):# datasource_json["name"] = datasource_json["name"] + "03"new_json_obj = {"title": folder_json["title"], "uid": folder_json["uid"]}json_obj = json.dumps(new_json_obj)print(new_json_obj)response = requests.post(f'{new_grafana_url}/folders', data=json_obj, headers=headers,auth=HTTPBasicAuth(username, passwd),verify=False)if response.status_code == 200:print(json.loads(response.text))return json.loads(response.text)else:print("Failed to retrieve dashboards.")# 获取所有仪表盘列表
def get_dashboards():response = requests.get(f'{grafana_url}/search', headers=headers, auth=HTTPBasicAuth(username, passwd),verify=False)if response.status_code == 200:print(json.loads(response.text))return json.loads(response.text)else:print("Failed to retrieve dashboards.")def save_dashboard_detail(dashboard_id):response = requests.get(f'{grafana_url}/dashboards/uid/{dashboard_id}', headers=headers,auth=HTTPBasicAuth(username, passwd), verify=False)if response.status_code == 200:ret_json = json.loads(response.text)print(ret_json)dashboard = ret_json["dashboard"]name = dashboard["title"]# 打开文件（若不存在则创建）并写入JSON数据with open("dashboard_" + name + ".json", "w") as file:# 使用json.dump()函数将数据转换为JSON格式后写入文件json.dump(ret_json, file)return ret_jsonelse:print("Failed to retrieve dashboards.")def set_dashboard_detail(dashboard_json):folderUid = dashboard_json["meta"]["folderUid"]del dashboard_json["meta"]#dashboard = file_json["dashboard"]dashboard_json["folderUid"] = folderUiddashboard_json["overwrite"] = Truedashboard_json["dashboard"]["id"] = NonedashboardJson = json.dumps(dashboard_json)print(f"{dashboardJson}")response = requests.post(f'{new_grafana_url}/dashboards/import', data=dashboardJson, headers=headers,auth=HTTPBasicAuth(username, passwd),verify=False)if response.status_code == 200:print(json.loads(response.text))return json.loads(response.text)else:print("Failed to retrieve dashboards.")# 创建新的仪表盘
def create_new_dashboard(title):data = {"overwrite": False,"dashboard": {"id": None,"uid": "","title": title,# ...其他配置项...},"folderId": 0,"message": ""}headers = {'Authorization': f'Bearer {api_key}','Content-Type': 'application/json'}response = requests.post(f'{grafana_url}/dashboards/db', json=data, headers=headers)if response.status_code == 200:return response.json().get('slug')else:print("Failed to create new dashboard.")# 更新现有仪表盘
def update_existing_dashboard(dashboard_slug, title):data = {"overwrite": True,"dashboard": {"id": None,"uid": "","title": title,# ...其他配置项...},"folderId": 0,"message": ""}headers = {'Authorization': f'Bearer {api_key}','Content-Type': 'application/json'}response = requests.put(f'{grafana_url}/dashboards/{dashboard_slug}', json=data, headers=headers)if response.status_code == 200:return response.json().get('slug')else:print("Failed to update existing dashboard.")# 删除指定仪表盘
def delete_dashboard(dashboard_slug):headers = {'Authorization': f'Bearer {api_key}'}response = requests.delete(f'{grafana_url}/dashboards/{dashboard_slug}', headers=headers)if response.status_code == 200:print("Dashboard deleted successfully.")else:print("Failed to delete the dashboard.")# 测试函数
if __name__ == "__main__":# # 清空所有数据源datasources = get_datasources()for datasource in datasources:# print(datasource["id"])datasource_detail = del_datasource_detail(datasource["id"])print(datasource_detail)current_dir = os.getcwd()file_names = os.listdir(current_dir)# 数据源导入for file_name in file_names:if file_name.startswith("datasource") & file_name.endswith(".json"):print(file_name)with open(file_name, 'r') as f:file_json = json.load(f)ret = set_datasource_detail(file_json, new_es_url,new_es_passwd)# 获取所有文件夹并保存到磁盘# folders = get_folders(grafana_url)# for folder in folders:# print(folder["title"])# folder_detail = save_folder_detail(folder["uid"])# print(folder_detail)# 文件夹导入# current_dir = os.getcwd()# file_names = os.listdir(current_dir)for file_name in file_names:if file_name.startswith("folder") & file_name.endswith(".json"):print(file_name)with open(file_name, 'r') as f:file_json = json.load(f)ret = set_folder_detail(file_json)# 获取所有仪表盘列表# dashboards = get_dashboards()# for board in dashboards:# # if type_folder==board["type"]:# # print(board["title"])# if type_dashboard == board["type"]:# print(board["title"])# save_dashboard_detail(board["uid"])# 仪表盘导入# current_dir = os.getcwd()# file_names = os.listdir(current_dir)for file_name in file_names:if file_name.startswith("dashboard") & file_name.endswith(".json"):print(file_name)with open(file_name, 'r') as f:file_json = json.load(f)# 域名处理Json_str2 = json.dumps(file_json)Json_str3= Json_str2.replace(old_domain,new_domain)Json_str4 = Json_str3.replace(old_domain, new_domain)json5 = json.loads( Json_str4 )ret = set_dashboard_detail(json5)# # 创建新的仪表盘# new_dashboard_slug = create_new_dashboard("New Dashboard")# print(f"Created a new dashboard with slug: {new_dashboard_slug}")## # 更新现有仪表盘# updated_dashboard_slug = update_existing_dashboard(new_dashboard_slug, "Updated Dashboard Title")# print(f"Updated an existing dashboard with slug: {updated_dashboard_slug}")## # 删除指定仪表盘# delete_dashboard(updated_dashboard_slug)
