更换发型成功

test/test_hair.py

@@ -0,0 +1,126 @@
+import threading
+import requests
+import json
+import base64
+import time
+import random
+
+url = 'http://xiangsilian.com:18801/api/swapHair/v1'
+
+headers = {'Content-Type': 'application/json'}
+data = {
+    "hair_id": "1907651680352395265",
+    "task_id": "345564352345",
+    "user_img_path": "https://cdn.meidaojia.com/ZoeFiles/testuser_8.JPG",
+    "is_hr": "false",
+    "output_format": "base64"
+}
+
+
+
+def image_to_base64(file_path, mime_type=None):
+    """
+    将图片文件转换为带Base64前缀的Data URI字符串
+    
+    参数:
+        file_path (str): 图片文件路径
+        mime_type (str): 可选，指定MIME类型。如果为None，则根据文件扩展名自动判断
+    
+    返回:
+        str: 带Base64前缀的Data URI字符串
+    """
+    # 如果没有指定MIME类型，根据文件扩展名推断
+    if mime_type is None:
+        extension = file_path.split('.')[-1].lower()
+        mime_types = {
+            'jpg': 'image/jpeg',
+            'jpeg': 'image/jpeg',
+            'png': 'image/png',
+            'gif': 'image/gif',
+            'webp': 'image/webp',
+            'bmp': 'image/bmp'
+        }
+        mime_type = mime_types.get(extension, 'application/octet-stream')
+    
+    # 读取文件内容并编码为Base64
+    with open(file_path, 'rb') as image_file:
+        encoded_string = base64.b64encode(image_file.read()).decode('utf-8')
+    
+    # 组合成Data URI格式
+    return f"data:{mime_type};base64,{encoded_string}"
+
+img64_str = image_to_base64("wertwert.jpg")
+data['user_img_path'] = img64_str
+
+responses = []
+
+lock = threading.Lock()
+
+
+send_index = 0
+
+def send_request(start_time):
+    try:
+        sleep_time = start_time 
+        time.sleep(sleep_time)
+        global send_index
+        send_data = data
+        response = requests.post(url, headers=headers, json=send_data)
+        with lock:  # 加锁，防止多线程同时修改 responses
+            responses.append(response)
+    except Exception as e:
+        with lock:
+            responses.append({"error": str(e)})
+
+threads = []
+num_requests = 1 
+
+time_start = time.time()
+
+start_time = 0
+for _ in range(num_requests):
+    start_time = 0
+    t = threading.Thread(target=send_request, args=(start_time,))
+    t.start()
+    threads.append(t)
+
+
+for t in threads:
+    t.join()
+
+end_time = time.time()
+
+total_time = (end_time - time_start)
+qps = num_requests/total_time
+print(f"num_requests:{num_requests} total_time:{total_time} qps:{qps}")
+
+
+
+import base64
+import os
+
+def save_base64_image(base64_str, filename):
+    """
+    将 Base64 编码的图片字符串保存为本地 JPG 文件。
+
+    :param base64_str: Base64 编码的图片字符串（可能包含 data:image/jpeg;base64, 前缀）
+    :param filename: 保存的文件名（例如 'image.jpg'）
+    """
+    # 如果包含 data URL 前缀，去掉它
+    if base64_str.startswith("data:image"):
+        base64_str = base64_str.split(",", 1)[1]
+
+    try:
+        image_data = base64.b64decode(base64_str)
+        with open(filename, "wb") as f:
+            f.write(image_data)
+        print(f"图片已保存为 {filename}")
+    except Exception as e:
+        print(f"保存图片失败: {e}")
+
+# 打印所有响应
+for i, resp in enumerate(responses, 1):
+    print(f"请求 {i} 结果: {resp.status_code, resp.text[:128]}")
+    json_obj = json.loads(resp.text)
+    data_string = json_obj['data']
+    save_base64_image(data_string, f"response_{i}.jpg")