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)