| 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 |
|
|
| |
| 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) |
|
|
| |
| 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) |
