Create app.py

app.py

@@ -0,0 +1,99 @@
+import gradio as gr
+import pandas as pd
+import requests
+import tempfile
+import os
+import time
+from typing import List
+
+# 全局进度缓存
+progress_log: List[str] = []
+result_file_path = None
+
+# 测试函数，根据你的业务修改这里，这里示例仅做http访问测试
+def test_host(host: str) -> dict:
+    try:
+        url = host if host.startswith("http") else f"http://{host}"
+        resp = requests.get(url, timeout=8)
+        return {"host": host, "status_code": resp.status_code, "ok": True, "error": ""}
+    except Exception as e:
+        return {"host": host, "status_code": 0, "ok": False, "error": str(e)}
+
+
+def upload_csv(file):
+    """接收上传csv，保存全局引用"""
+    global progress_log
+    progress_log.clear()
+    if file is None:
+        return "未选择文件", ""
+    progress_log.append(f"✅ 文件已上传: {file.orig_name}")
+    return "\n".join(progress_log), file
+
+
+def start_batch_test(uploaded_file):
+    global progress_log, result_file_path
+    progress_log.clear()
+    if uploaded_file is None:
+        progress_log.append("❌ 请先上传 hosts.csv")
+        yield "\n".join(progress_log), None
+        return
+
+    try:
+        df = pd.read_csv(uploaded_file)
+        if "host" not in df.columns:
+            progress_log.append("❌ csv必须包含 host 列")
+            yield "\n".join(progress_log), None
+            return
+        host_list = df["host"].dropna().tolist()
+        total = len(host_list)
+        progress_log.append(f"📋 读取到 {total} 个待测试主机")
+        yield "\n".join(progress_log), None
+
+        out_rows = []
+        for idx, h in enumerate(host_list):
+            progress_log.append(f"[{idx+1}/{total}] 测试 {h}")
+            res = test_host(h)
+            out_rows.append(res)
+            if not res["ok"]:
+                progress_log.append(f"    ⚠️失败: {res['error']}")
+            else:
+                progress_log.append(f"    ✅ status:{res['status_code']}")
+            yield "\n".join(progress_log), None
+            time.sleep(0.2)  # 限流，防止CPU/网络打爆
+
+        # 输出临时csv
+        tmp_fd, tmp_path = tempfile.mkstemp(suffix=".csv")
+        os.close(tmp_fd)
+        pd.DataFrame(out_rows).to_csv(tmp_path, index=False)
+        result_file_path = tmp_path
+        progress_log.append("\n🎉全部任务完成，可以下载结果文件")
+        yield "\n".join(progress_log), tmp_path
+
+    except Exception as err:
+        progress_log.append(f"💥任务异常:{str(err)}")
+        yield "\n".join(progress_log), None
+
+
+def get_download_file():
+    global result_file_path
+    return result_file_path
+
+
+with gr.Blocks(title="Csv 🐠") as demo:
+    gr.Markdown("# 🐠 批量主机测试工具")
+    file_input = gr.File(label="上传 hosts.csv", file_types=[".csv"])
+    log_text = gr.Textbox(label="运行日志", lines=15, interactive=False)
+    start_btn = gr.Button("🚀 开始测试", variant="primary")
+    download_output = gr.File(label="下载测试结果csv")
+
+    # 上传事件
+    file_input.upload(upload_csv, inputs=[file_input], outputs=[log_text, file_input])
+    # 启动测试，使用生成器实现实时日志
+    start_btn.click(
+        fn=start_batch_test,
+        inputs=[file_input],
+        outputs=[log_text, download_output]
+    )
+
+if __name__ == "__main__":
+    demo.launch(server_name="0.0.0.0", server_port=7860)