AI + Python 巡检远程主机并输出html报告

pip install fastapi==0.115.6 starlette==0.41.3 uvicorn jinja2 python-multipart paramiko report_generator psutil dashscope -i https://pypi.tuna.tsinghua.edu.cn/simple

fastapi：高性能 Web 框架，用来快速搭建 API 接口starlette：FastAPI 的底层框架，提供路由、请求处理等基础功能uvicorn：ASGI 服务器，用来启动和运行 FastAPI 应用jinja2：模板引擎，用于动态渲染 HTML 页面python-multipart：处理文件上传和表单数据paramiko：SSH 客户端库，用于远程连接服务器、执行命令report_generator：报告生成库（非主流库，具体功能需看文档）psutil：获取系统信息，如 CPU、内存、磁盘使用率等dashscope：阿里云大模型 SDK，调用通义千问等 AI 模型

from fastapi import FastAPI, Requestfrom fastapi.responses import HTMLResponsefrom fastapi.templating import Jinja2Templatesimport uvicornimport osfrom sys_inspect import read_ip_list, get_all_server_metrics, generate_ai_reportapp = FastAPI()templates = Jinja2Templates(directory="templates")# 获取当前 main.py 所在目录，拼出 ip.txt 的绝对路径BASE_DIR = os.path.dirname(os.path.abspath(__file__))IP_FILE = os.path.join(BASE_DIR, "ip.txt")@app.get("/", response_class=HTMLResponse)async def read_root(request: Request):    ip_list = read_ip_list(IP_FILE)    return templates.TemplateResponse(        "index.html",        {            "request": request,            "ip_list": ip_list,            "ip_count": len(ip_list)        }    )@app.post("/run_check", response_class=HTMLResponse)async def run_inspection(request: Request):    print("🔍 开始巡检所有远程主机...")    # 1. 从 ip.txt 读取并采集所有远程机器数据    all_metrics = get_all_server_metrics()    # 2. 调用 AI 生成 HTML 报告    html_content = generate_ai_report(all_metrics)    # 3. 返回报告给浏览器    return HTMLResponse(content=html_content)if __name__ == "__main__":    uvicorn.run(app, host="0.0.0.0", port=8000)

import paramikofrom dashscope import Generationimport osimport re# 【重要】请替换为你自己的阿里云 API KeyDASHSCOPE_API_KEY = "sk-ws-H.EIDXMHI.wn9XXXX"# 远程机器的统一账号密码SSH_USERNAME = "root"SSH_PASSWORD = "123456"SSH_PORT = 22def read_ip_list(filepath="ip.txt"):    """从 ip.txt 读取远程主机 IP 列表"""    with open(filepath, "r") as f:        ips = [line.strip() for line in f if line.strip()]    return ipsdef get_remote_metrics(hostname):    """通过 SSH 连接远程机器，采集 CPU、内存、磁盘、端口、进程信息"""    ssh = paramiko.SSHClient()    ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())    try:        ssh.connect(hostname=hostname, port=SSH_PORT, username=SSH_USERNAME, password=SSH_PASSWORD, timeout=10)        # 定义要执行的命令        commands = {            "hostname": "hostname",            "os_info": "uname -r",            "cpu_usage": "top -bn1 | grep 'Cpu(s)' | awk '{print $2}'",            "mem_info": "free -m | grep Mem",            "disk_usage": "df -h / | tail -1 | awk '{print $5}'",            # ===== 新增：端口和进程 =====            "listening_ports": "ss -tlnp 2>/dev/null || netstat -tlnp 2>/dev/null",            "top_processes": "ps aux --sort=-%mem | head -16",            "uptime": "uptime",        }        results = {"target": hostname}        for key, cmd in commands.items():            stdin, stdout, stderr = ssh.exec_command(cmd)            output = stdout.read().decode("utf-8").strip()            results[key] = output if output else "N/A"        # 解析内存数据        if results["mem_info"] and results["mem_info"] != "N/A":            parts = results["mem_info"].split()            mem_total = int(parts[1])            mem_used = int(parts[2])            results["mem_total"] = f"{round(mem_total / 1024, 2)} GB"            results["mem_used_percent"] = f"{round(mem_used / mem_total * 100, 1)}%"        else:            results["mem_total"] = "N/A"            results["mem_used_percent"] = "N/A"        # 清理中间字段        results.pop("mem_info", None)        return results    except Exception as e:        return {"target": hostname, "error": str(e)}    finally:        ssh.close()def get_all_server_metrics():    """读取 ip.txt，逐台采集所有远程机器的数据"""    # 用和 main.py 同样的方式定位 ip.txt    base_dir = os.path.dirname(os.path.abspath(__file__))    ip_file = os.path.join(base_dir, "ip.txt")    ips = read_ip_list(ip_file)    all_metrics = []    for ip in ips:        print(f"  🔍 正在采集: {ip} ...")        metrics = get_remote_metrics(ip)        all_metrics.append(metrics)    return all_metricsdef generate_ai_report(metrics_list):    """将多台机器的指标发送给通义千问，生成 HTML 巡检报告"""    # 把所有机器的数据拼成文本    data_text = ""    for i, m in enumerate(metrics_list, 1):        if "error" in m:            data_text += f"\n【主机{i}】{m['target']} — 连接失败: {m['error']}\n"        else:            data_text += f"""【主机{i}】- 目标IP: {m['target']}- 主机名: {m['hostname']}- 操作系统内核: {m['os_info']}- 运行时长: {m.get('uptime', 'N/A')}- CPU使用率: {m['cpu_usage']}%- 内存总量: {m['mem_total']}- 内存使用率: {m['mem_used_percent']}- 磁盘使用率: {m['disk_usage']}- 监听端口及对应进程:{m.get('listening_ports', 'N/A')}- 资源占用最高的进程（按内存排序前15）:{m.get('top_processes', 'N/A')}"""    prompt = f"""你是一个专业的运维专家。根据以下多台服务器的巡检数据，生成一份 HTML 格式的巡检报告。要求：1. 使用内联 CSS 样式，风格简洁现代（白底黑字，关键数据高亮）。2. 包含标题、巡检时间。3. 每台机器单独一个区块，包含：   - 基础信息表格（IP、主机名、系统、运行时长、CPU、内存、磁盘）   - 监听端口表格（协议、端口、进程名），从 ss/netstat 输出中提取关键信息   - 业务进程表格（用户、PID、CPU%、内存%、命令），从 ps 输出中提取关键信息4. 如果某台机器连接失败，单独标红提示。5. 根据数据给出"健康建议"，例如：   - CPU 使用率超过 80% 提示检查高负载进程   - 内存使用率超过 85% 提示可能存在内存泄漏   - 磁盘使用率超过 80% 提示清理日志或扩容   - 发现异常端口或可疑进程时给出安全提醒数据如下：{data_text}"""    messages = [{"role": "user", "content": prompt}]    response = Generation.call(        model="qwen-turbo",        messages=messages,        api_key=DASHSCOPE_API_KEY,        result_format="message",    )    if response.status_code == 200:        html_content = response.output.choices[0].message.content        # 去掉 AI 返回的 Markdown 代码块标记        html_content = re.sub(r'^```html\s*', '', html_content, flags=re.IGNORECASE)        html_content = re.sub(r'^```\s*', '', html_content)        html_content = re.sub(r'```\s* $ ', '', html_content)        html_content = html_content.strip()        return html_content    else:        return f"<h1>AI 生成失败</h1><p>Error: {response.message}</p>"

<!DOCTYPE html><htmllang="zh-CN"><head>    <metacharset="UTF-8">    <title>智能运维巡检系统</title>    <style>        body { font-family: sans-serif; max-width: 600px; margin: 50px auto; padding: 20px; text-align: center; }        button { padding: 12px 24px; background-color: #007bff; color: white; border: none; border-radius: 4px; cursor: pointer; font-size: 16px; }        button:hover { background-color: #0056b3; }        /* 新增样式：用于显示 IP 列表 */        .ip-box {            margin-top: 30px;            padding: 15px;            background-color: #f8f9fa;            border: 1px solid #e9ecef;            border-radius: 8px;            text-align: left;        }        .ip-title { font-weight: bold; color: #495057; margin-bottom: 10px; display: block; }        .ip-tag {            display: inline-block;            background-color: #e2e6ea;            color: #212529;            padding: 4px 8px;            margin: 4px;            border-radius: 4px;            font-family: monospace;            font-size: 14px;        }    </style></head><body>    <h1>🛡️ 服务器智能巡检</h1>    <p>点击下方按钮，自动巡检配置好的远程主机</p>    <!-- 表单 -->    <formaction="/run_check"method="post">        <buttontype="submit">开始一键巡检 (共 {{ ip_count }} 台)</button>    </form>    <!-- 新增：IP 列表显示区域 -->    <divclass="ip-box">        <spanclass="ip-title">📋 待巡检目标列表 (来自 ip.txt):</span>        {% for ip in ip_list %}            <spanclass="ip-tag">{{ ip }}</span>        {% endfor %}    </div></body></html>