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