File size: 3,387 Bytes
8f1296d cd8efea 8f1296d cd8efea 8f1296d cd8efea 8f1296d | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 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) |