Create app.py

app.py

@@ -0,0 +1,170 @@
+import gradio as gr
+import subprocess
+import threading
+import time
+import os
+import signal
+
+class TrafficGenerator:
+    def __init__(self):
+        self.process = None
+        self.is_running = False
+        self.is_paused = False
+    
+    def start_traffic(self, target_url, requests_per_second, duration):
+        if self.is_running:
+            return "流量生成已在运行中！"
+        
+        try:
+            # 构建命令行参数
+            cmd = [
+                "python", "web_traffic_generator.py",
+                "--url", target_url,
+                "--rps", str(requests_per_second),
+                "--duration", str(duration)
+            ]
+            
+            self.process = subprocess.Popen(
+                cmd,
+                stdout=subprocess.PIPE,
+                stderr=subprocess.PIPE,
+                preexec_fn=os.setsid
+            )
+            self.is_running = True
+            self.is_paused = False
+            return f"开始生成流量到: {target_url}"
+        except Exception as e:
+            return f"启动失败: {str(e)}"
+    
+    def pause_traffic(self):
+        if not self.is_running:
+            return "没有运行的流量生成进程"
+        
+        if self.is_paused:
+            return "流量生成已经暂停"
+        
+        try:
+            if self.process and self.process.poll() is None:
+                os.killpg(os.getpgid(self.process.pid), signal.SIGSTOP)
+                self.is_paused = True
+                return "流量生成已暂停"
+        except Exception as e:
+            return f"暂停失败: {str(e)}"
+        return "暂停操作失败"
+    
+    def resume_traffic(self):
+        if not self.is_running:
+            return "没有运行的流量生成进程"
+        
+        if not self.is_paused:
+            return "流量生成已经在运行中"
+        
+        try:
+            if self.process and self.process.poll() is None:
+                os.killpg(os.getpgid(self.process.pid), signal.SIGCONT)
+                self.is_paused = False
+                return "流量生成已恢复"
+        except Exception as e:
+            return f"恢复失败: {str(e)}"
+        return "恢复操作失败"
+    
+    def stop_traffic(self):
+        if not self.is_running:
+            return "没有运行的流量生成进程"
+        
+        try:
+            if self.process:
+                os.killpg(os.getpgid(self.process.pid), signal.SIGTERM)
+                self.process.wait(timeout=5)
+        except:
+            try:
+                self.process.kill()
+            except:
+                pass
+        
+        self.is_running = False
+        self.is_paused = False
+        return "流量生成已停止"
+
+# 创建流量生成器实例
+traffic_gen = TrafficGenerator()
+
+def control_traffic(action, target_url, requests_per_second, duration):
+    if action == "开始":
+        return traffic_gen.start_traffic(target_url, requests_per_second, duration)
+    elif action == "暂停":
+        return traffic_gen.pause_traffic()
+    elif action == "继续":
+        return traffic_gen.resume_traffic()
+    elif action == "停止":
+        return traffic_gen.stop_traffic()
+    else:
+        return "未知操作"
+
+# 创建Gradio界面
+with gr.Blocks(title="网络流量生成器") as demo:
+    gr.Markdown("# 🌐 网络流量生成器控制面板")
+    gr.Markdown("使用此界面控制网络流量生成器的启动、暂停、继续和停止操作")
+    
+    with gr.Row():
+        with gr.Column():
+            target_url = gr.Textbox(
+                label="目标URL",
+                value="https://example.com",
+                placeholder="请输入目标URL"
+            )
+            requests_per_second = gr.Number(
+                label="每秒请求数 (RPS)",
+                value=10,
+                minimum=1,
+                maximum=1000
+            )
+            duration = gr.Number(
+                label="持续时间 (秒)",
+                value=60,
+                minimum=10,
+                maximum=3600
+            )
+        
+        with gr.Column():
+            action = gr.Radio(
+                choices=["开始", "暂停", "继续", "停止"],
+                label="控制操作",
+                value="开始"
+            )
+            control_button = gr.Button("执行操作", variant="primary")
+    
+    output = gr.Textbox(
+        label="操作结果",
+        interactive=False,
+        lines=3
+    )
+    
+    # 状态显示
+    status_display = gr.Textbox(
+        label="当前状态",
+        value="就绪",
+        interactive=False
+    )
+    
+    def update_status():
+        if traffic_gen.is_running:
+            if traffic_gen.is_paused:
+                return "🟡 已暂停"
+            else:
+                return "🟢 运行中"
+        else:
+            return "⚪ 就绪"
+    
+    # 绑定事件
+    control_button.click(
+        fn=control_traffic,
+        inputs=[action, target_url, requests_per_second, duration],
+        outputs=output
+    ).then(
+        fn=update_status,
+        outputs=status_display
+    )
+
+if __name__ == "__main__":
+    demo.launch(server_name="0.0.0.0", server_port=7860)