a

project-test/大树-模型平台/test_platform.py

@@ -0,0 +1,43 @@
+import json
+import unittest
+
+import requests
+
+
+def print1(obj):
+    print(json.dumps(obj, indent=4, ensure_ascii=False))
+
+
+def get_token():
+    headers = {
+        'authority': 'aimp.aimpdev.k5.bigtree.tech',
+        'accept': 'application/json, text/plain, */*',
+        'accept-language': 'zh-CN,zh;q=0.9,en;q=0.8',
+    }
+
+    json_data = {
+        'username': 'lzq',
+        'password': 'Aa123456',
+        'token': True,
+    }
+
+    response = requests.post('https://aimp.aimpdev.k5.bigtree.tech/api/aimp-auth-web/auth/login', headers=headers, json=json_data)
+    return response.json()['token']
+
+
+class MyTest(unittest.TestCase):
+    def setUp(self) -> None:
+        # self.host = "https://edsp.rxdpdev.k5.bigtree.tech/api/rxdp-xdata-web/rxdp-file"
+        self.host = "http://127.1:8080"
+        self.head = {
+            "Access-Token": get_token()
+        }
+
+    def test_invoke(self):
+        url = "http://127.0.0.1:8080/out_api/processInvoke/shui_test"
+        data = {"taxpayer_id": "91320585MA1NUBLR8L", "apikey": "8290b473ed6c42eaa6e1ed0ec68c1c3d", "apisecret": "n4FBN9youmxXUEIUtM9J5EfwItZGFx34nimJgv0xsP26WbLcG8QcE2HK0jzXu0cRR35Q1QDjg5HUQuApq7mimA"}
+        header = {
+            "appId": "tian_test",
+            "secret": "ZryZsW6er-kaYxSjLwJ8XtAUtP0kowhZ"
+        }
+        print(requests.post(url, json=data, headers=header).json())

test_py_invoke.py

@@ -0,0 +1,26 @@
+def exec_def(cmd_str, obj):
+    import requests
+    print(requests.post("http://172.18.162.173:8000/exec", json={
+        "function_str": cmd_str,
+        "call_str": "exec(obj)",
+        "param_dict": {
+            "obj": obj
+        }
+    }).json())
+
+
+# ================[ 自定义部分 Start ]===============
+exec_str = """
+def exec(obj):
+    import json
+    obj["a"] = 123
+    return obj
+"""
+obj = {"b": 45}
+# ================[ 自定义部分 End ]===============
+
+
+exec_def(exec_str, obj)
+
+
+round(357-128*math.log(float(param["p"])/(1-float(param["p"]))))

tmp.py

@@ -1,23 +1,3 @@
-def exec_def(cmd_str, obj):
-    import requests
-    print(requests.post("http://172.18.162.173:8000/exec", json={
-        "function_str": cmd_str,
-        "call_str": "exec(obj)",
-        "param_dict": {
-            "obj": obj
-        }
-    }).json())
+import time
 
-
-# ================[ 自定义部分 Start ]===============
-exec_str = """
-def exec(obj):
-    import json
-    obj["a"] = 123
-    return obj
-"""
-obj = {"b": 45}
-# ================[ 自定义部分 End ]===============
-
-
-exec_def(exec_str, obj)
+print(time.time())

tmp2.py

@@ -1,11 +1,30 @@
+import base64
+
+
+def file_to_base64(filename):
+    # 文件转base64
+    with open(filename, "rb") as file:
+        return base64.b64encode(file.read()).decode()
+
+
+# 使用示例
+fileStr = file_to_base64("/Users/alvin/Downloads/tmp.csv")
+
+
 def exec(obj):
-    import json
-    all_key = ['zScore', 'yingyunzijinZhouzhuancishu', 'aeM6IdOrgnumD', 'area', 'bh10017', 'bh10018', 'bh10022', 'bhScore', 'cnGudongChange6m', 'cnHigh', 'cnZeroincome24m', 'credit15', 'creditcardScoreV4', 'cunhuoZhouzhuanratio', 'cvVatSales12m', 'drsNodebtscore', 'fraud18', 'gudingzichanZongzichanRatio', 'hades1', 'highMostCnRuleType', 'incomeTaxReal12m', 'largeAmountPaymentV4', 'lirunRatio', 'liudongRatio', 'liudongzichanZhouzhuanratio', 'maxRiskLevel', 'registcapi', 'riskDegree', 'riskScore', 'slopeVatSales12m', 'stockpercent', 'stockpercentJicha', 'sudongRatio', 'tlIdM6NbankPassorg', 'vatMonth12m', 'vatSales3mRatio', 'vatSalesHuanbi6m', 'yingfuZhouzhuanratio', 'yingshouZhouzhuanratio']
-    # 创建一个对象，里面有all_key对应的key，假如这个key在obj里面，则赋值为obj里的value，假如没有则赋值为null
-    data_obj = {key: obj[key] if key in obj else None for key in all_key}
-    # 然后把对象json化赋值到下面的obj的data属性里面
-    obj['data'] = json.dumps(data_obj)
+    import base64
+    import io
+    import csv
+    # 解析里面的file，把list存入obj
+    base64_file = obj.get("file")
+    if base64_file is not None:
+        base64_bytes = base64_file.encode()
+        decoded_bytes = base64.b64decode(base64_bytes)
+        file_like = io.StringIO(decoded_bytes.decode())
+        reader = csv.reader(file_like)
+        csv_list = list(reader)
+        obj['list'] = csv_list
     return obj
 
 
-print(exec({"zScore": 1}))
+print(exec({"file": fileStr}))

tmp3.py

@@ -1,21 +0,0 @@
-import os
-
-import uvicorn
-from fastapi import FastAPI
-
-app = FastAPI(title="project name ", description="通用系统 ", version="v 0.0.0")
-
-
-@app.get("/call_sh")
-def call_sh(path: str):
-    os.system(f"sh {path} &")
-    return "请等待结果"
-
-
-@app.get("/")
-def call_sh():
-    return {"code": 200}
-
-
-if __name__ == '__main__':
-    uvicorn.run(app='tmp3:app', host="0.0.0.0", port=9990, reload=True)

tps_util.py

@@ -0,0 +1,52 @@
+import concurrent.futures
+import requests
+import time
+import statistics
+
+
+def call_api(url, method, params, json):
+    start = time.time()
+    response = requests.request(method, url, params=params, json=json)
+    # print(response.json())
+    end = time.time()
+    return (end - start) * 1000, response.status_code, len(response.content)
+
+
+def test_api_tps(url, method, params=None, json=None, concurrency=1, duration=1):
+    successes = 0
+    failures = 0
+    times = []
+    data_lengths = []
+
+    with concurrent.futures.ThreadPoolExecutor(max_workers=concurrency) as executor:
+        start = time.time()
+        futures = [executor.submit(call_api, url, method, params, json) for i in range(concurrency)]
+
+        while time.time() - start < duration:
+            completed, futures = concurrent.futures.wait(futures, timeout=0)
+            for future in completed:
+                elapsed, status, length = future.result()
+                times.append(elapsed)
+                data_lengths.append(length)
+                if status == 200:
+                    successes += 1
+                else:
+                    failures += 1
+                futures.add(executor.submit(call_api, url, method, params, json))
+
+    total = successes + failures
+    tps = total / duration
+    mean = statistics.mean(times)
+    min_time = min(times)
+    max_time = max(times)
+
+    print(f'TPS:{tps}; ', end=' ')
+    print(f'平均时间(ms): {mean:.0f}; ', end=' ')
+    print(f'最小时间(ms): {min_time:.0f}; ', end=' ')
+    print(f'最大时间(ms): {max_time:.0f}; ', end=' ')
+    print(f'成功数: {successes}; ', end=' ')
+    print(f'失败数: {failures}; ', end=' ')
+    print(f'总数据长度: {sum(data_lengths)}; ', end=' ')
+
+
+test_api_tps('https://httpbin.tianyunperfect.cn/get', 'get', params={"a": 1}, duration=2)