Add files via upload

tools/ql_api.py

@@ -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

tools/ql_util.py

@@ -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