mirror of
https://github.com/daiyanan1992/qinglongtest
synced 2025-12-16 23:09:38 +08:00
Add files via upload
This commit is contained in:
daiyanan1992
212
tools/ql_api.py
Normal file
212
tools/ql_api.py
Normal file
@@ -0,0 +1,212 @@
|
||||
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
|
||||
14
tools/ql_util.py
Normal file
14
tools/ql_util.py
Normal file
@@ -0,0 +1,14 @@
|
||||
import random
|
||||
import string
|
||||
|
||||
|
||||
# 随机生成数字与小写字母字符串
|
||||
def get_random_str(rdm_leg: int = 8, status: bool = False):
|
||||
random_str = ''
|
||||
base_str = string.octdigits
|
||||
if status:
|
||||
base_str = base_str + string.ascii_lowercase
|
||||
length = len(base_str) - 1
|
||||
for i in range(rdm_leg):
|
||||
random_str += base_str[random.randint(0, length)]
|
||||
return random_str
|
||||
Reference in New Issue
Block a user