mirror of
https://github.com/daiyanan1992/qinglongtest
synced 2025-12-18 07:44:51 +08:00
213 lines
6.6 KiB
Python
213 lines
6.6 KiB
Python
import json, os
|
||
import time
|
||
from sys import stdout
|
||
|
||
import requests,re
|
||
|
||
ql_auth_path = '/ql/data/config/auth.json'
|
||
ql_config_path = '/ql/data/config/config.sh'
|
||
#判断环境变量
|
||
flag = 'new'
|
||
if not os.path.exists(ql_auth_path):
|
||
ql_auth_path = '/ql/config/auth.json'
|
||
ql_config_path = '/ql/config/config.sh'
|
||
if not os.path.exists(ql_config_path):
|
||
ql_config_path = '/ql/config/config.js'
|
||
flag = 'old'
|
||
# ql_auth_path = r'D:\Docker\ql\config\auth.json'
|
||
ql_url = 'http://localhost:5600'
|
||
|
||
|
||
def __get_token() -> str or None:
|
||
with open(ql_auth_path, 'r', encoding='utf-8') as f:
|
||
j_data = json.load(f)
|
||
return j_data.get('token')
|
||
|
||
|
||
def __get__headers() -> dict:
|
||
headers = {
|
||
'Accept': 'application/json',
|
||
'Content-Type': 'application/json;charset=UTF-8',
|
||
'Authorization': 'Bearer ' + __get_token()
|
||
}
|
||
return headers
|
||
|
||
# 封装读取环境变量的方法
|
||
def get_cookie(key, default="", output=True):
|
||
def no_read():
|
||
if output:
|
||
print_now(f"未填写环境变量 {key} 请添加")
|
||
return default
|
||
return get_cookie_data(key) if get_cookie_data(key) else no_read()
|
||
|
||
#获取ck
|
||
def get_cookie_data(name):
|
||
ck_list = []
|
||
cookie = None
|
||
cookies = get_config_and_envs(name)
|
||
for ck in cookies:
|
||
if ck.get('status') == 0:
|
||
ck_list.append(ck.get('value'))
|
||
if len(ck_list) < 1:
|
||
print('变量{}共配置{}条CK,请添加环境变量,或查看环境变量状态'.format(name,len(ck_list)))
|
||
return ck_list
|
||
|
||
# 修改print方法 避免某些环境下python执行print 不会去刷新缓存区导致信息第一时间不及时输出
|
||
def print_now(content):
|
||
print(content)
|
||
stdout.flush()
|
||
|
||
|
||
# 查询环境变量
|
||
def get_envs(name: str = None) -> list:
|
||
params = {
|
||
't': int(time.time() * 1000)
|
||
}
|
||
if name is not None:
|
||
params['searchValue'] = name
|
||
res = requests.get(ql_url + '/api/envs', headers=__get__headers(), params=params)
|
||
j_data = res.json()
|
||
if j_data['code'] == 200:
|
||
return j_data['data']
|
||
return []
|
||
|
||
|
||
# 查询环境变量+config.sh变量
|
||
def get_config_and_envs(name: str = None) -> list:
|
||
params = {
|
||
't': int(time.time() * 1000)
|
||
}
|
||
#返回的数据data
|
||
data = []
|
||
if name is not None:
|
||
params['searchValue'] = name
|
||
res = requests.get(ql_url + '/api/envs', headers=__get__headers(), params=params)
|
||
j_data = res.json()
|
||
if j_data['code'] == 200:
|
||
data = j_data['data']
|
||
with open(ql_config_path, 'r', encoding='utf-8') as f:
|
||
while True:
|
||
# Get next line from file
|
||
line = f.readline()
|
||
# If line is empty then end of file reached
|
||
if not line :
|
||
break;
|
||
#print(line.strip())
|
||
exportinfo = line.strip().replace("\"","").replace("\'","")
|
||
#去除注释#行
|
||
rm_str_list = re.findall(r'^#(.+?)', exportinfo,re.DOTALL)
|
||
#print('rm_str_list数据：{}'.format(rm_str_list))
|
||
exportinfolist = []
|
||
if len(rm_str_list) == 1:
|
||
exportinfo = ""
|
||
#list_all = re.findall(r'export[ ](.+?)', exportinfo,re.DOTALL)
|
||
#print('exportinfo数据：{}'.format(exportinfo))
|
||
#以export分隔，字符前面新增标记作为数组0，数组1为后面需要的数据
|
||
list_all = ("标记"+exportinfo.replace(" ","").replace(" ","")).split("export")
|
||
#print('list_all数据：{}'.format(list_all))
|
||
if len(list_all) > 1:
|
||
#以=分割，查找需要的环境名字
|
||
tmp = list_all[1].split("=")
|
||
if len(tmp) > 1:
|
||
|
||
info = tmp[0]
|
||
if name in info:
|
||
#print('需要查询的环境数据：{}'.format(tmp))
|
||
data_tmp = []
|
||
data_json = {
|
||
'id': None,
|
||
'value': tmp[1],
|
||
'status': 0,
|
||
'name': name,
|
||
'remarks': "",
|
||
'position': None,
|
||
'timestamp': int(time.time()*1000),
|
||
'created': int(time.time()*1000)
|
||
}
|
||
if flag == 'old':
|
||
data_json = {
|
||
'_id': None,
|
||
'value': tmp[1],
|
||
'status': 0,
|
||
'name': name,
|
||
'remarks': "",
|
||
'position': None,
|
||
'timestamp': int(time.time()*1000),
|
||
'created': int(time.time()*1000)
|
||
}
|
||
#print('需要的数据：{}'.format(data_json))
|
||
data.append(data_json)
|
||
#print('第二次配置数据：{}'.format(data))
|
||
return data
|
||
|
||
|
||
# 新增环境变量
|
||
def post_envs(name: str, value: str, remarks: str = None) -> list:
|
||
params = {
|
||
't': int(time.time() * 1000)
|
||
}
|
||
data = [{
|
||
'name': name,
|
||
'value': value
|
||
}]
|
||
if remarks is not None:
|
||
data[0]['remarks'] = remarks
|
||
res = requests.post(ql_url + '/api/envs', headers=__get__headers(), params=params, json=data)
|
||
j_data = res.json()
|
||
if j_data['code'] == 200:
|
||
return j_data['data']
|
||
return []
|
||
|
||
|
||
# 修改环境变量
|
||
def put_envs(_id: str, name: str, value: str, remarks: str = None) -> bool:
|
||
params = {
|
||
't': int(time.time() * 1000)
|
||
}
|
||
|
||
data = {
|
||
'name': name,
|
||
'value': value,
|
||
'id': _id
|
||
}
|
||
if flag == 'old':
|
||
data = {
|
||
'name': name,
|
||
'value': value,
|
||
'_id': _id
|
||
}
|
||
|
||
if remarks is not None:
|
||
data['remarks'] = remarks
|
||
res = requests.put(ql_url + '/api/envs', headers=__get__headers(), params=params, json=data)
|
||
j_data = res.json()
|
||
if j_data['code'] == 200:
|
||
return True
|
||
return False
|
||
|
||
|
||
# 禁用环境变量
|
||
def disable_env(_id: str) -> bool:
|
||
params = {
|
||
't': int(time.time() * 1000)
|
||
}
|
||
data = [_id]
|
||
res = requests.put(ql_url + '/api/envs/disable', headers=__get__headers(), params=params, json=data)
|
||
j_data = res.json()
|
||
if j_data['code'] == 200:
|
||
return True
|
||
return False
|
||
|
||
|
||
# 启用环境变量
|
||
def enable_env(_id: str) -> bool:
|
||
params = {
|
||
't': int(time.time() * 1000)
|
||
}
|
||
data = [_id]
|
||
res = requests.put(ql_url + '/api/envs/enable', headers=__get__headers(), params=params, json=data)
|
||
j_data = res.json()
|
||
if j_data['code'] == 200:
|
||
return True
|
||
return False
|