Update app.py

app.py

@@ -1,239 +1,211 @@
-# import os
-# import json
-# import random
-# import uuid
-# from flask import Flask, request, jsonify, session, render_template
-# from flask_cors import CORS
-# from datetime import datetime
-# from elo_rank import EloRank
-
-# app = Flask(__name__)
-# CORS(app)
-# app.secret_key = 'supersecretkey'
-
-# base_dir = os.path.dirname(os.path.abspath(__file__))
-
-# DATA_DIR = os.path.join(base_dir, '/app/data')
-# RESULTS_DIR = os.path.join(base_dir, '/app/results')
-
-# # 实例化 EloRank 系统
-# elo_rank_system = EloRank()
-
-# # 初始化 Elo 排名的模型
-# models = ['output_path_4o', 'output_path_miniomni', 'output_path_speechgpt', 'output_path_funaudio', 'output_path_4o_cascade', 'output_path_4o_llama_omni']
-# for model in models:
-#     elo_rank_system.add_model(model)
-
-# import os
-
-# def print_directory_structure(start_path, indent=''):
-#     for item in os.listdir(start_path):
-#         item_path = os.path.join(start_path, item)
-#         if os.path.isdir(item_path):
-#             print(f"{indent}📁 {item}/")
-#             print_directory_structure(item_path, indent + '    ')
-#         else:
-#             print(f"{indent}📄 {item}")
-
-
-# def load_test_data(task):
-#     """Load the JSON file corresponding to the selected task"""
-#     # 调用函数，打印当前目录结构
-#     with open('/app/test_text.txt', 'r') as file:
-#             content = file.read()
-#             print(content)
-#     # with open(os.path.join(DATA_DIR, f"{task}.json"), "r", encoding='utf-8') as f:
-#     #     test_data = json.load(f)
-
-#     try:
-#         with open(os.path.join(DATA_DIR, f"{task}.json"), "r", encoding='utf-8') as f:
-#             test_data = json.load(f)
-#     except FileNotFoundError:
-#         return jsonify({"message": "Test data file not found"}), 400
+import os
+import json
+import random
+import uuid
+from flask import Flask, request, jsonify, session, render_template
+from flask_cors import CORS
+from datetime import datetime
+from elo_rank import EloRank
 
-    
-#     # 更新音频路径，将它们指向 Flask 静态文件夹
-#     for item in test_data:
-#         item['input_path'] = f"/app/static/audio{item['input_path']}"
-#         item['output_path_4o'] = f"/app/static/audio{item['output_path_4o']}"
-#         item['output_path_miniomni'] = f"/app/static/audio{item['output_path_miniomni']}"
-#         item['output_path_speechgpt'] = f"/app/static/audio{item['output_path_speechgpt']}"
-#         item['output_path_funaudio'] = f"/app/static/audio{item['output_path_funaudio']}"
-#         item['output_path_4o_cascade'] = f"/app/static/audio{item['output_path_4o_cascade']}"
-#         item['output_path_4o_llama_omni'] = f"/app/static/audio{item['output_path_4o_llama_omni']}"
-    
-#     return test_data
-
-
-# def save_result(task, username, result_data, session_id):
-#     """Save user's result in a separate file"""
-#     file_path = os.path.join(RESULTS_DIR, f"{task}_{username}_{session_id}.jsonl")
-#     # 获取所有模型的 Elo 分数
-#     elo_scores = {model: elo_rank_system.get_rating(model) for model in models}
-    
-#     # 添加 Elo 分数和时间戳到结果数据
-#     result_data['elo_scores'] = elo_scores
-#     result_data['timestamp'] = datetime.now().isoformat()
-#     with open(file_path, "a", encoding='utf-8') as f:
-#         f.write(json.dumps(result_data) + "\n")
-
-# @app.route('/start_test', methods=['POST'])
-# def start_test():
-#     """Initiate the test for a user with the selected task"""
-#     data = request.json
-#     task = data['task']
-#     username = data['username']
-
-#     # Load the test data
-#     test_data = load_test_data(task)
-#     if not test_data:
-#         return jsonify({"message": "No test data available"}), 400
-    
-#     # Shuffle test data for the user
-#     random.shuffle(test_data)
-
-#     # Generate a unique session ID (for example using uuid)
-#     session_id = str(uuid.uuid4())
-
-#     # Store in session
-#     session['task'] = task
-#     session['username'] = username
-#     session['test_data'] = test_data
-#     session['current_index'] = 0
-#     session['session_id'] = session_id  # Store the session ID in the session
-
-#     task_description = test_data[0].get('task_description', '')
-
-#     print(session)
-
-
-#     return jsonify({"message": "Test started", "total_tests": len(test_data), "task_description": task_description})
-
-
-
-# @app.route('/next_test', methods=['GET'])
-# def next_test():
-#     """Serve the next test item"""
-#     # if 'current_index' not in session or 'test_data' not in session:
-#     #     return jsonify({"message": "No active test found"}), 400
-
-#     if 'current_index' not in session or 'test_data' not in session:
-#         # 转换 session 为字典避免序列化错误
-#         print("Debug Session:", dict(session))  # 调试信息
-#         return jsonify({"message": "Session data missing"}), 400
-
-
-#     current_index = session['current_index']
-#     test_data = session['test_data']
-
-#     if current_index >= len(test_data):
-#         # Return the "Test completed" message when all tests are done
-#         return jsonify({"message": "Test completed"}), 200
-
-#     # 使用 EloRank 的 sample_next_match 来选择两款模型
-#     selected_models = elo_rank_system.sample_next_match()
-
-#     if not selected_models or len(selected_models) != 2:
-#         return jsonify({"message": "Error selecting models"}), 500
-
-#     # Serve test data with the two selected models
-#     current_test = test_data[current_index]
-#     session['selected_models'] = selected_models
-#     session['current_index'] += 1
-
-#     return jsonify({
-#         "text": current_test["text"],
-#         "input_path": current_test["input_path"],
-#         "model_a": selected_models[0],
-#         "model_b": selected_models[1],
-#         "audio_a": current_test[selected_models[0]],
-#         "audio_b": current_test[selected_models[1]]
-#     })
+app = Flask(__name__)
+CORS(app)
+app.secret_key = 'supersecretkey'
 
-# @app.route('/submit_result', methods=['POST'])
-# def submit_result():
-#     """Submit the user's result and save it"""
-#     data = request.json
-#     chosen_model = data['chosen_model']
+base_dir = os.path.dirname(os.path.abspath(__file__))
 
-#     username = session.get('username')
-#     task = session.get('task')
-#     current_index = session.get('current_index') - 1  # Subtract since we increment after serving
-#     session_id = session.get('session_id')  # Get the session ID
+DATA_DIR = os.path.join(base_dir, '/app/data')
+RESULTS_DIR = os.path.join(base_dir, '/app/results')
 
-#     if not username or not task or current_index < 0:
-#         return jsonify({"message": "No active test found"}), 400
+# 实例化 EloRank 系统
+elo_rank_system = EloRank()
 
-#     # Retrieve the selected models
-#     selected_models = session['selected_models']
-#     model_a = selected_models[0]
-#     model_b = selected_models[1]
+# 初始化 Elo 排名的模型
+models = ['output_path_4o', 'output_path_miniomni', 'output_path_speechgpt', 'output_path_funaudio', 'output_path_4o_cascade', 'output_path_4o_llama_omni']
+for model in models:
+    elo_rank_system.add_model(model)
 
-#     result = {
-#         "name": username,
-#         "chosen_model": chosen_model,
-#         "model_a": model_a,
-#         "model_b": model_b,
-#         "result": {
-#             model_a: 1 if chosen_model == 'A' else 0,
-#             model_b: 1 if chosen_model == 'B' else 0
-#         }
-#     }
+import os
 
-#     # Save the result for the current test using session_id to avoid filename conflict
-#     test_data = session['test_data'][current_index]
-#     result_data = {**test_data, **result}
-#     save_result(task, username, result_data, session_id)
+def print_directory_structure(start_path, indent=''):
+    for item in os.listdir(start_path):
+        item_path = os.path.join(start_path, item)
+        if os.path.isdir(item_path):
+            print(f"{indent}📁 {item}/")
+            print_directory_structure(item_path, indent + '    ')
+        else:
+            print(f"{indent}📄 {item}")
 
-#     # 更新 Elo 排名系统
-#     if chosen_model == 'A':
-#         elo_rank_system.record_match(model_a, model_b)
-#     else:
-#         elo_rank_system.record_match(model_b, model_a)
 
-#     return jsonify({"message": "Result submitted", "model_a": model_a, "model_b": model_b, "chosen_model": chosen_model})
+def load_test_data(task):
+    """Load the JSON file corresponding to the selected task"""
+    # 调用函数，打印当前目录结构
+    with open('/app/test_text.txt', 'r') as file:
+            content = file.read()
+            print(content)
+    # with open(os.path.join(DATA_DIR, f"{task}.json"), "r", encoding='utf-8') as f:
+    #     test_data = json.load(f)
 
-# @app.route('/end_test', methods=['GET'])
-# def end_test():
-#     """End the test session"""
-#     session.clear()
-#     return jsonify({"message": "Test completed"})
+    try:
+        with open(os.path.join(DATA_DIR, f"{task}.json"), "r", encoding='utf-8') as f:
+            test_data = json.load(f)
+    except FileNotFoundError:
+        return jsonify({"message": "Test data file not found"}), 400
 
-# # 渲染index.html页面
-# @app.route('/')
-# def index():
-#     return render_template('index.html')
+    
+    # 更新音频路径，将它们指向 Flask 静态文件夹
+    for item in test_data:
+        item['input_path'] = f"/app/static/audio{item['input_path']}"
+        item['output_path_4o'] = f"/app/static/audio{item['output_path_4o']}"
+        item['output_path_miniomni'] = f"/app/static/audio{item['output_path_miniomni']}"
+        item['output_path_speechgpt'] = f"/app/static/audio{item['output_path_speechgpt']}"
+        item['output_path_funaudio'] = f"/app/static/audio{item['output_path_funaudio']}"
+        item['output_path_4o_cascade'] = f"/app/static/audio{item['output_path_4o_cascade']}"
+        item['output_path_4o_llama_omni'] = f"/app/static/audio{item['output_path_4o_llama_omni']}"
+    
+    return test_data
 
-# if __name__ == '__main__':
-#     if not os.path.exists(RESULTS_DIR):
-#         os.makedirs(RESULTS_DIR)
-#     # 允许局域网访问
-#     app.run(host="0.0.0.0", debug=True, port=8080)
 
+def save_result(task, username, result_data, session_id):
+    """Save user's result in a separate file"""
+    file_path = os.path.join(RESULTS_DIR, f"{task}_{username}_{session_id}.jsonl")
+    # 获取所有模型的 Elo 分数
+    elo_scores = {model: elo_rank_system.get_rating(model) for model in models}
+    
+    # 添加 Elo 分数和时间戳到结果数据
+    result_data['elo_scores'] = elo_scores
+    result_data['timestamp'] = datetime.now().isoformat()
+    with open(file_path, "a", encoding='utf-8') as f:
+        f.write(json.dumps(result_data) + "\n")
+
+@app.route('/start_test', methods=['POST'])
+def start_test():
+    """Initiate the test for a user with the selected task"""
+    data = request.json
+    task = data['task']
+    username = data['username']
+
+    # Load the test data
+    test_data = load_test_data(task)
+    if not test_data:
+        return jsonify({"message": "No test data available"}), 400
+    
+    # Shuffle test data for the user
+    random.shuffle(test_data)
 
+    # Generate a unique session ID (for example using uuid)
+    session_id = str(uuid.uuid4())
 
+    # Store in session
+    session['task'] = task
+    session['username'] = username
+    session['test_data'] = test_data
+    session['current_index'] = 0
+    session['session_id'] = session_id  # Store the session ID in the session
+
+    task_description = test_data[0].get('task_description', '')
 
-from flask import Flask, render_template_string
-
-app = Flask(__name__)
-
-@app.route("/")
-def home():
-    # 使用 iframe 嵌入目标网站
-    return render_template_string("""
-    <!DOCTYPE html>
-    <html lang="en">
-    <head>
-        <meta charset="UTF-8">
-        <meta name="viewport" content="width=device-width, initial-scale=1.0">
-        <title>Embedded Website</title>
-    </head>
-    <body style="margin: 0; padding: 0; height: 100%; overflow: hidden;">
-        <iframe src="http://71.132.14.167:6002/" frameborder="0" style="width: 100%; height: 100%; border: none;"></iframe>
-    </body>
-    </html>
-    """)
-
-if __name__ == "__main__":
-    app.run(host="0.0.0.0", port=8080)
-
+    print(session)
+
+
+    return jsonify({"message": "Test started", "total_tests": len(test_data), "task_description": task_description})
+
+
+
+@app.route('/next_test', methods=['GET'])
+def next_test():
+    """Serve the next test item"""
+    # if 'current_index' not in session or 'test_data' not in session:
+    #     return jsonify({"message": "No active test found"}), 400
+
+    if 'current_index' not in session or 'test_data' not in session:
+        # 转换 session 为字典避免序列化错误
+        print("Debug Session:", dict(session))  # 调试信息
+        return jsonify({"message": "Session data missing"}), 400
+
+
+    current_index = session['current_index']
+    test_data = session['test_data']
+
+    if current_index >= len(test_data):
+        # Return the "Test completed" message when all tests are done
+        return jsonify({"message": "Test completed"}), 200
+
+    # 使用 EloRank 的 sample_next_match 来选择两款模型
+    selected_models = elo_rank_system.sample_next_match()
+
+    if not selected_models or len(selected_models) != 2:
+        return jsonify({"message": "Error selecting models"}), 500
+
+    # Serve test data with the two selected models
+    current_test = test_data[current_index]
+    session['selected_models'] = selected_models
+    session['current_index'] += 1
+
+    return jsonify({
+        "text": current_test["text"],
+        "input_path": current_test["input_path"],
+        "model_a": selected_models[0],
+        "model_b": selected_models[1],
+        "audio_a": current_test[selected_models[0]],
+        "audio_b": current_test[selected_models[1]]
+    })
+
+@app.route('/submit_result', methods=['POST'])
+def submit_result():
+    """Submit the user's result and save it"""
+    data = request.json
+    chosen_model = data['chosen_model']
+
+    username = session.get('username')
+    task = session.get('task')
+    current_index = session.get('current_index') - 1  # Subtract since we increment after serving
+    session_id = session.get('session_id')  # Get the session ID
+
+    if not username or not task or current_index < 0:
+        return jsonify({"message": "No active test found"}), 400
+
+    # Retrieve the selected models
+    selected_models = session['selected_models']
+    model_a = selected_models[0]
+    model_b = selected_models[1]
+
+    result = {
+        "name": username,
+        "chosen_model": chosen_model,
+        "model_a": model_a,
+        "model_b": model_b,
+        "result": {
+            model_a: 1 if chosen_model == 'A' else 0,
+            model_b: 1 if chosen_model == 'B' else 0
+        }
+    }
+
+    # Save the result for the current test using session_id to avoid filename conflict
+    test_data = session['test_data'][current_index]
+    result_data = {**test_data, **result}
+    save_result(task, username, result_data, session_id)
+
+    # 更新 Elo 排名系统
+    if chosen_model == 'A':
+        elo_rank_system.record_match(model_a, model_b)
+    else:
+        elo_rank_system.record_match(model_b, model_a)
+
+    return jsonify({"message": "Result submitted", "model_a": model_a, "model_b": model_b, "chosen_model": chosen_model})
+
+@app.route('/end_test', methods=['GET'])
+def end_test():
+    """End the test session"""
+    session.clear()
+    return jsonify({"message": "Test completed"})
+
+# 渲染index.html页面
+@app.route('/')
+def index():
+    return render_template('index.html')
+
+if __name__ == '__main__':
+    if not os.path.exists(RESULTS_DIR):
+        os.makedirs(RESULTS_DIR)
+    # 允许局域网访问
+    app.run(host="0.0.0.0", debug=True, port=8080)