requirements update31

app.py

@@ -6,6 +6,10 @@ import sys
 from pathlib import Path
 import concurrent.futures
 import multiprocessing
+import time
+import threading
+import queue
+from datetime import datetime, timedelta
 from src.containerized_eval import eval_string_script
 
 # 添加当前目录和src目录到模块搜索路径
@@ -97,14 +101,236 @@ def evaluate_code(code, language):
     except Exception as e:
         return {"status": "Exception", "error": str(e)}
 
-# 创建Gradio接口
-demo = gr.Interface(
-    fn=evaluate,
-    inputs=gr.JSON(),
-    outputs=gr.JSON(),
-    title="代码评估服务",
-    description="支持多种编程语言的代码评估服务"
-)
+# 创建任务队列和相关变量
+task_queue = queue.Queue()
+tasks_info = []
+tasks_lock = threading.Lock()
+total_tasks_count = 0
+completed_tasks_count = 0
+estimated_completion_time = None
+average_task_time = 5  # 默认平均任务时间（秒）
+
+def add_task(input_data):
+    """添加任务到队列
+    
+    Args:
+        input_data: 任务数据
+        
+    Returns:
+        dict: 包含任务ID和状态的字典
+    """
+    global total_tasks_count, tasks_info
+    
+    try:
+        if not isinstance(input_data, list):
+            return {"status": "错误", "message": "输入必须是列表格式"}
+        
+        task_id = int(time.time() * 1000)  # 使用时间戳作为任务ID
+        task_count = len(input_data)
+        
+        with tasks_lock:
+            for item in input_data:
+                task_queue.put((task_id, item))
+                total_tasks_count += 1
+            
+            # 更新任务信息
+            task_info = {
+                "id": task_id,
+                "count": task_count,
+                "status": "等待中",
+                "submit_time": datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
+                "completed": 0
+            }
+            tasks_info.append(task_info)
+            
+            # 更新预计完成时间
+            update_estimated_completion_time()
+        
+        return {"status": "成功", "task_id": task_id, "message": f"已添加{task_count}个任务到队列"}
+    
+    except Exception as e:
+        return {"status": "错误", "message": str(e)}
+
+def update_estimated_completion_time():
+    """更新预计完成时间"""
+    global estimated_completion_time, average_task_time
+    
+    remaining_tasks = total_tasks_count - completed_tasks_count
+    if remaining_tasks > 0:
+        # 计算预计完成时间
+        estimated_seconds = remaining_tasks * average_task_time
+        estimated_completion_time = datetime.now() + timedelta(seconds=estimated_seconds)
+    else:
+        estimated_completion_time = None
+
+def process_tasks():
+    """处理队列中的任务"""
+    global completed_tasks_count, average_task_time, tasks_info
+    
+    while True:
+        try:
+            if not task_queue.empty():
+                start_time = time.time()
+                
+                # 获取任务并处理
+                task_id, task_data = task_queue.get()
+                result = evaluate_single_case(task_data)
+                
+                # 更新任务状态
+                with tasks_lock:
+                    completed_tasks_count += 1
+                    
+                    # 更新平均任务时间（使用移动平均）
+                    task_time = time.time() - start_time
+                    average_task_time = (average_task_time * 0.9) + (task_time * 0.1)
+                    
+                    # 更新任务信息
+                    for task in tasks_info:
+                        if task["id"] == task_id:
+                            task["completed"] += 1
+                            if task["completed"] >= task["count"]:
+                                task["status"] = "已完成"
+                            break
+                    
+                    # 更新预计完成时间
+                    update_estimated_completion_time()
+                
+                task_queue.task_done()
+            else:
+                time.sleep(0.1)
+        except Exception as e:
+            print(f"处理任务时出错: {str(e)}")
+            time.sleep(1)
+
+# 启动任务处理线程
+for _ in range(multiprocessing.cpu_count()):
+    threading.Thread(target=process_tasks, daemon=True).start()
+
+def get_queue_status():
+    """获取队列状态
+    
+    Returns:
+        tuple: 包含任务信息、完成进度和预计完成时间的元组
+    """
+    with tasks_lock:
+        # 复制任务信息以避免并发问题
+        current_tasks = tasks_info.copy()
+        
+        # 计算进度
+        if total_tasks_count > 0:
+            progress = completed_tasks_count / total_tasks_count
+        else:
+            progress = 0
+        
+        # 格式化预计完成时间
+        if estimated_completion_time:
+            eta = estimated_completion_time.strftime("%Y-%m-%d %H:%M:%S")
+            remaining_seconds = (estimated_completion_time - datetime.now()).total_seconds()
+            
+            if remaining_seconds < 60:
+                eta_text = f"{int(remaining_seconds)}秒后完成"
+            elif remaining_seconds < 3600:
+                eta_text = f"{int(remaining_seconds/60)}分钟后完成"
+            else:
+                eta_text = f"{int(remaining_seconds/3600)}小时{int((remaining_seconds%3600)/60)}分钟后完成"
+        else:
+            eta = "无任务"
+            eta_text = "无任务"
+    
+    return current_tasks, progress, eta, eta_text
+
+# 创建Gradio界面
+with gr.Blocks(title="代码评估服务", theme=gr.themes.Soft(primary_hue="blue")) as demo:
+    gr.Markdown(
+        """
+        # 代码评估服务任务队列监控
+        
+        实时监控系统中的任务队列状态和预计完成时间
+        """
+    )
+    
+    with gr.Row():
+        with gr.Column(scale=2):
+            # 任务提交区域
+            with gr.Box():
+                gr.Markdown("### 提交新任务")
+                input_json = gr.JSON(label="输入数据（JSON格式列表）")
+                submit_btn = gr.Button("提交任务", variant="primary")
+        
+        with gr.Column(scale=3):
+            # 状态概览区域
+            with gr.Box():
+                gr.Markdown("### 队列状态概览")
+                with gr.Row():
+                    with gr.Column():
+                        total_tasks = gr.Textbox(label="总任务数", value="0")
+                    with gr.Column():
+                        completed = gr.Textbox(label="已完成任务", value="0")
+                    with gr.Column():
+                        remaining = gr.Textbox(label="剩余任务", value="0")
+                
+                progress_bar = gr.Slider(minimum=0, maximum=1, value=0, label="完成进度", interactive=False)
+                eta_display = gr.Textbox(label="预计完成时间", value="无任务")
+    
+    # 任务列表区域
+    with gr.Box():
+        gr.Markdown("### 任务队列详情")
+        task_table = gr.Dataframe(
+            headers=["任务ID", "任务数量", "状态", "提交时间", "已完成/总数"],
+            datatype=["str", "number", "str", "str", "str"],
+            row_count=10,
+            col_count=(5, "fixed"),
+            interactive=False
+        )
+    
+    # 刷新按钮
+    refresh_btn = gr.Button("刷新状态", variant="secondary")
+    
+    # 添加任务的处理函数
+    def handle_submit(input_data):
+        result = add_task(input_data)
+        return update_ui_status(), result["message"]
+    
+    # 更新UI状态的函数
+    def update_ui_status():
+        tasks, progress, eta, eta_text = get_queue_status()
+        
+        # 准备表格数据
+        table_data = []
+        for task in tasks:
+            table_data.append([
+                str(task["id"]),
+                task["count"],
+                task["status"],
+                task["submit_time"],
+                f"{task['completed']}/{task['count']}"
+            ])
+        
+        return (
+            table_data,
+            str(total_tasks_count),
+            str(completed_tasks_count),
+            str(total_tasks_count - completed_tasks_count),
+            progress,
+            eta_text
+        )
+    
+    # 设置事件处理
+    submit_btn.click(
+        handle_submit,
+        inputs=[input_json],
+        outputs=[task_table, total_tasks, completed, remaining, progress_bar, eta_display, gr.Textbox(visible=False)]
+    )
+    
+    refresh_btn.click(
+        update_ui_status,
+        inputs=[],
+        outputs=[task_table, total_tasks, completed, remaining, progress_bar, eta_display]
+    )
+    
+    # 定时刷新UI
+    demo.load(update_ui_status, inputs=[], outputs=[task_table, total_tasks, completed, remaining, progress_bar, eta_display])
+    demo.every(3, update_ui_status, inputs=[], outputs=[task_table, total_tasks, completed, remaining, progress_bar, eta_display])
 
 if __name__ == "__main__":
     demo.launch()