Python 3.14 遇上 Gemini Pro：手把手教你搭建一套具备“自愈能力”的 AI 自动化运维系统

"""
项目名称：E路领航 AI 自动化运维系统 (v2026.1)
功能：监控 OpenWrt 日志并利用 Gemini Pro 进行异常自愈决策
环境：Python 3.14 + AI Studio Tier 1
作者：苏阳 (blog.oool.cc)
"""

import os
import asyncio
import json
import httpx  # Python 3.14 推荐使用的异步请求库
from datetime import datetime

# --- 配置脱敏处理 ---
# 请在 Win11 系统环境变量中设置以下参数，严禁在代码中硬编码真实 Key
GEMINI_API_KEY = os.getenv("GEMINI_STUDIO_API_KEY")
ENDPOINT_URL = "https://generativelanguage.googleapis.com/v1beta/models/gemini-pro:generateContent"
TARGET_GATEWAY = "172.20.0.1"  # OpenWrt 网关地址

class AIOpsAgent:
    def __init__(self):
        self.headers = {"Content-Type": "application/json"}
        self.log_cache = []

    async def get_system_logs(self):
        """
        模拟抓取 OpenWrt 内核日志
        在实际部署中，建议通过 SSH 密钥对执行: logread -f
        """
        # 演示数据：模拟一次 HomeProxy 导致的连接超时
        sample_log = "[2026-02-08 21:00:05] HomeProxy: Connect to Google failed: timeout."
        return sample_log

    async def ask_gemini_for_solution(self, log_content):
        """
        将异常日志提交给 AI Studio 进行语义分析
        """
        prompt = f"""
        你是一位专业的 Linux 运维专家。以下是来自 OpenWrt (网关 172.20.0.1) 的异常日志：
        '{log_content}'
        
        请根据我的环境（OpenWrt + HomeProxy + AdGuard Home）给出以下建议：
        1. 故障原因简述。
        2. 具体的修复指令（例如重启服务、清理缓存等）。
        3. 给出 JSON 格式的执行逻辑，字段包含: 'action', 'command'。
        """
        
        payload = {
            "contents": [{"parts": [{"text": prompt}]}]
        }

        async with httpx.AsyncClient() as client:
            try:
                response = await client.post(
                    f"{ENDPOINT_URL}?key={GEMINI_API_KEY}",
                    headers=self.headers,
                    json=payload,
                    timeout=30.0
                )
                if response.status_code == 200:
                    result = response.json()
                    # 提取 AI 返回的建议文本
                    advice = result['candidates'][0]['content']['parts'][0]['text']
                    return advice
                else:
                    return f"API 调用失败，状态码: {response.status_code}"
            except Exception as e:
                return f"连接 Gemini 发生错误: {str(e)}"

    async def execute_repair(self, advice):
        """
        解析 AI 返回的建议并执行修复
        警告：在生产环境下，建议增加人工审核环节或指令白名单
        """
        print(f"正在分析 AI 建议: {advice}")
        # 这里可以加入解析 JSON 指令并调用 Paramiko 执行 SSH 命令的逻辑
        # 针对演示，我们仅输出逻辑
        if "restart" in advice.lower():
            print(f"执行决策：正在通过 Cloudflare Tunnel 远程重启相关服务...")

    async def run_loop(self):
        print("E路领航 AI 运维 Agent 已启动...")
        while True:
            log = await self.get_system_logs()
            if "error" in log.lower() or "failed" in log.lower():
                print(f"检测到异常日志: {log}")
                advice = await self.ask_gemini_for_solution(log)
                await self.execute_repair(advice)
            
            # 每一小时轮询一次，或根据需要调整
            await asyncio.sleep(3600)

if __name__ == "__main__":
    agent = AIOpsAgent()
    try:
        asyncio.run(agent.run_loop())
    except KeyboardInterrupt:
        print("Agent 已手动停止。")