网工如何利用 AI 优化链路巡检？聊聊各技术岗位的 AI 实践路径

巡检员 → SSH登录设备 → 逐条敲命令 → 肉眼对比历史数据 → 手动填Excel → 提交巡检报告

#!/usr/bin/env python3
"""
网络链路智能巡检脚本
- 批量采集设备接口状态、CRC 错误包、光功率、路由表摘要
- AI 分析链路健康度并给出根因建议
"""

import json
import re
from datetime import datetime
from concurrent.futures import ThreadPoolExecutor, as_completed
from netmiko import ConnectHandler
from openai import OpenAI

# ============ 设备清单 ============
DEVICES = [
    {"host": "10.10.1.1", "name": "CORE-SW-01", "type": "cisco_ios"},
    {"host": "10.10.1.2", "name": "CORE-SW-02", "type": "cisco_ios"},
    {"host": "10.10.2.1", "name": "EDGE-RT-01",  "type": "cisco_ios"},
    {"host": "10.10.2.2", "name": "EDGE-RT-02",  "type": "huawei"},
    {"host": "10.10.3.1", "name": "DC-FW-01",    "type": "fortinet"},
]

# 巡检命令模板（按设备类型区分）
INSPECT_COMMANDS = {
    "cisco_ios": [
        "show interfaces | include (Ethernet|input errors|output errors|CRC|runts|giants)",
        "show ip interface brief | exclude unassigned",
        "show logging | last 50",
        "show ip route summary",
    ],
    "huawei": [
        "display interface | include (GigabitEthernet|input error|output error|CRC)",
        "display ip interface brief",
        "display logbuffer | last 50",
        "display ip routing-table summary",
    ],
    "fortinet": [
        "diagnose hardware deviceinfo nic",
        "diagnose sys top",
        "execute log filter category traffic",
    ],
}

# ============ 第一步：批量采集 ============
def collect_device(device):
    """SSH 登录设备执行巡检命令，返回原始输出"""
    result = {
        "device": device["name"],
        "host": device["host"],
        "timestamp": datetime.now().isoformat(),
        "raw_output": {},
    }
    try:
        conn = ConnectHandler(
            host=device["host"],
            device_type=device["type"],
            username="inspect_bot",
            password="REPLACE_WITH_VAULT_TOKEN",
            timeout=20,
        )
        commands = INSPECT_COMMANDS.get(device["type"], [])
        for cmd in commands:
            output = conn.send_command(cmd, read_timeout=30)
            result["raw_output"][cmd] = output
        conn.disconnect()
        result["status"] = "success"
    except Exception as e:
        result["status"] = "failed"
        result["error"] = str(e)
    return result


def batch_collect(devices, max_workers=5):
    """多线程并行采集所有设备"""
    all_results = []
    with ThreadPoolExecutor(max_workers=max_workers) as executor:
        futures = {executor.submit(collect_device, dev): dev for dev in devices}
        for future in as_completed(futures):
            dev = futures[future]
            try:
                result = future.result()
                all_results.append(result)
                print(f"[OK] {dev['name']} 采集完成")
            except Exception as e:
                print(f"[FAIL] {dev['name']} 采集失败: {e}")
                all_results.append({
                    "device": dev["name"],
                    "status": "failed",
                    "error": str(e),
                })
    return all_results


# ============ 第二步：数据清洗 ============
def parse_interface_errors(raw_text):
    """从 show interfaces 输出中提取错误计数"""
    issues = []
    # 匹配类似 "0 input errors, 0 CRC, 0 frame" 的行
    error_pattern = re.compile(
        r'(\d+)\s+input errors.*?(\d+)\s+CRC.*?(\d+)\s+frame', re.IGNORECASE
    )
    for line in raw_text.split("\n"):
        match = error_pattern.search(line)
        if match:
            input_errors, crc, frame = map(int, match.groups())
            if input_errors > 0 or crc > 0:
                issues.append({
                    "input_errors": input_errors,
                    "crc": crc,
                    "frame": frame,
                    "raw_line": line.strip(),
                })
    return issues


def clean_results(raw_results):
    """清洗原始采集数据，提取关键指标"""
    cleaned = []
    for item in raw_results:
        if item["status"] != "success":
            cleaned.append({
                "device": item["device"],
                "status": "unreachable",
                "issues": [],
            })
            continue
        
        issues = []
        for cmd, output in item["raw_output"].items():
            if "interface" in cmd.lower():
                iface_issues = parse_interface_errors(output)
                issues.extend(iface_issues)
        
        cleaned.append({
            "device": item["device"],
            "host": item["host"],
            "timestamp": item["timestamp"],
            "status": "reachable",
            "interface_issues": issues,
            "raw_output": item["raw_output"],
        })
    return cleaned


# ============ 第三步：AI 智能分析 ============
client = OpenAI(
    base_url="http://10.10.100.50:8000/v1",  # 内网部署的开源模型
    api_key="internal-not-needed",
)

ANALYSIS_PROMPT = """你是一个资深网络工程师，正在分析网络设备巡检数据。

以下是本次巡检采集到的设备状态信息，包括接口错误计数、日志摘要、路由表摘要。
请完成以下分析：

1. **健康度评分**：为每台设备打分（0-100），100 表示完全健康。
2. **异常识别**：找出所有异常项，包括但不限于：
   - 接口 CRC 错误 / input errors 增长
   - 链路带宽利用率异常
   - 路由表条目突变
   - 日志中的告警信息
3. **根因推测**：对每个异常给出可能的原因（光模块老化？线缆问题？对端设备问题？配置变更？）
4. **关联分析**：如果多台设备出现关联性异常（比如两台对端设备的同一链路都有 CRC），指出关联关系。
5. **处置建议**：给出具体的处理步骤和优先级（P0/P1/P2）。

巡检数据：
{data}

请以 JSON 格式输出分析结果。
"""


def ai_analyze(cleaned_data):
    """将清洗后的数据喂给大模型做智能分析"""
    # 压缩数据，避免超出上下文窗口
    compressed = []
    for item in cleaned_data:
        # 只传关键信息，不传完整 raw_output
        entry = {
            "device": item["device"],
            "status": item["status"],
        }
        if "interface_issues" in item:
            entry["interface_issues"] = item["interface_issues"]
        # 日志只取最近的关键行
        if "raw_output" in item:
            for cmd, output in item["raw_output"].items():
                if "log" in cmd.lower():
                    lines = output.strip().split("\n")
                    entry["recent_logs"] = lines[-10:] if len(lines) > 10 else lines
        compressed.append(entry)
    
    prompt = ANALYSIS_PROMPT.format(data=json.dumps(compressed, ensure_ascii=False, indent=2))
    
    response = client.chat.completions.create(
        model="qwen2.5-14b-instruct",
        messages=[
            {"role": "system", "content": "你是网络运维专家，擅长从设备巡检数据中发现潜在问题并给出专业建议。"},
            {"role": "user", "content": prompt},
        ],
        temperature=0.3,
        max_tokens=4096,
    )
    
    return response.choices[0].message.content


# ============ 第四步：生成报告 ============
def generate_report(cleaned_data, ai_analysis):
    """生成最终的巡检报告"""
    report = f"""
# 网络链路智能巡检报告

**巡检时间**: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}
**巡检设备数**: {len(cleaned_data)}
**正常设备**: {sum(1 for d in cleaned_data if d['status'] == 'reachable' and not d.get('interface_issues'))}
**异常设备**: {sum(1 for d in cleaned_data if d.get('interface_issues'))}
**不可达设备**: {sum(1 for d in cleaned_data if d['status'] == 'unreachable')}

---

## AI 分析结果

{ai_analysis}

---

## 设备明细

"""
    for item in cleaned_data:
        status_emoji = "✅" if item["status"] == "reachable" and not item.get("interface_issues") else "⚠️"
        report += f"### {status_emoji} {item['device']}\n"
        report += f"- 状态: {item['status']}\n"
        if item.get("interface_issues"):
            report += f"- 接口异常: {len(item['interface_issues'])} 项\n"
            for issue in item["interface_issues"]:
                report += f"  - input_errors={issue['input_errors']}, CRC={issue['crc']}\n"
        report += "\n"
    
    return report


# ============ 主流程 ============
if __name__ == "__main__":
    print("=" * 60)
    print("开始网络链路智能巡检")
    print("=" * 60)
    
    # 1. 批量采集
    print("\n[1/4] 采集设备状态...")
    raw_data = batch_collect(DEVICES)
    
    # 2. 数据清洗
    print("\n[2/4] 清洗巡检数据...")
    cleaned = clean_results(raw_data)
    
    # 3. AI 分析
    print("\n[3/4] AI 智能分析中...")
    ai_result = ai_analyze(cleaned)
    print(f"AI 分析完成，结果长度: {len(ai_result)} 字符")
    
    # 4. 生成报告
    print("\n[4/4] 生成巡检报告...")
    report = generate_report(cleaned, ai_result)
    
    # 保存报告
    report_file = f"inspection_report_{datetime.now().strftime('%Y%m%d_%H%M%S')}.md"
    with open(report_file, "w", encoding="utf-8") as f:
        f.write(report)
    
    print(f"\n巡检完成！报告已保存到 {report_file}")

#!/usr/bin/env python3
"""
AI 辅助代码审查
- 从 GitLab/GitHub 获取 PR diff
- 按文件拆分后喂给 AI 审查
- 输出结构化的审查意见
"""

import re
import json
import subprocess
from openai import OpenAI

client = OpenAI(
    base_url="http://10.10.100.50:8000/v1",
    api_key="internal-not-needed",
)

# 团队代码规范（喂给 AI 作为审查标准）
TEAM_CODING_STANDARDS = """
团队代码规范摘要：
1. 所有外部输入必须做参数校验，推荐使用 @Valid + 自定义 Validator
2. Service 层方法必须有日志记录（入参 + 执行耗时）
3. 数据库查询禁止使用 SELECT *，必须指定字段
4. 所有 HTTP 接口必须有 @Operation 注解（Swagger 文档）
5. 异常处理统一走 GlobalExceptionHandler，禁止在 Controller 里 try-catch
6. 线程池禁止使用 new Thread()，必须走 ThreadPoolExecutor 或 @Async
7. 敏感信息（密码、密钥）禁止硬编码，必须走配置中心
8. SQL 语句中禁止拼接字符串，必须使用参数化查询
"""

REVIEW_PROMPT = """你是一个严格的代码审查员。请基于以下团队代码规范审查这段代码变更。

## 团队代码规范
{standards}

## 变更文件
文件名: {filename}
变更类型: {change_type}

## 代码 Diff
```diff
{diff_content}

for line in diff_text.split("\n"):
    if line.startswith("diff --git"):
        if current_file:
            files.append({
                "filename": current_file,
                "diff": "\n".join(current_lines),
            })
        match = re.search(r'diff --git a/(.+?) b/', line)
        current_file = match.group(1) if match else "unknown"
        current_lines = [line]
    else:
        current_lines.append(line)

if current_file:
    files.append({
        "filename": current_file,
        "diff": "\n".join(current_lines),
    })

return files

prompt = REVIEW_PROMPT.format(
    standards=TEAM_CODING_STANDARDS,
    filename=file_info["filename"],
    change_type="modified",
    diff_content=file_info["diff"][:8000],  # 截断超长 diff
)

response = client.chat.completions.create(
    model="qwen2.5-coder-14b",
    messages=[
        {"role": "system", "content": "你是资深代码审查员，严格但不过度吹毛求疵。"},
        {"role": "user", "content": prompt},
    ],
    temperature=0.2,
    max_tokens=2048,
)

return {
    "filename": file_info["filename"],
    "review": response.choices[0].message.content,
}

all_reviews = []
for f in files:
    print(f"  审查中: {f['filename']}")
    review = review_file(f)
    all_reviews.append(review)

return all_reviews

print("\n" + "=" * 60)
print("代码审查报告")
print("=" * 60)
for r in reviews:
    if r.get("skipped"):
        continue
    print(f"\n📄 {r['filename']}")
    print(r["review"])

### 2.3 实际场景：一次差点上线的 SQL 注入

有次我们团队一个新人提了个 PR，在 Controller 里直接拼 SQL 查数据。代码大概是：

```java
// ❌ 危险代码
@GetMapping("/search")
public List<User> searchUsers(@RequestParam String keyword) {
    String sql = "SELECT * FROM users WHERE name LIKE '%" + keyword + "%'";
    return jdbcTemplate.query(sql, new UserRowMapper());
}

{
  "severity": "blocker",
  "issues": [
    {
      "line": "第 42 行",
      "type": "security",
      "severity": "blocker",
      "message": "SQL 注入风险：keyword 参数直接拼接进 SQL 语句，攻击者可通过构造特殊输入执行任意 SQL。",
      "suggestion": "使用参数化查询：jdbcTemplate.query(\"SELECT id, name, email FROM users WHERE name LIKE ?\", new Object[]{\"%\" + keyword + \"%\"}, new UserRowMapper())"
    },
    {
      "line": "第 42 行",
      "type": "style",
      "severity": "major",
      "message": "违反规范第 3 条：使用了 SELECT *，应指定具体字段。",
      "suggestion": "改为 SELECT id, name, email FROM users"
    }
  ]
}

#!/usr/bin/env python3
"""
AI 容量趋势预测
- 从 Prometheus 拉取历史指标
- AI 分析趋势并给出扩容建议
"""

import json
import requests
from datetime import datetime, timedelta
from openai import OpenAI

PROMETHEUS_URL = "http://prometheus:9090"
client = OpenAI(base_url="http://10.10.100.50:8000/v1", api_key="internal")

# 关键监控指标
METRICS_TO_TRACK = {
    "cpu_usage": '100 - (avg(rate(node_cpu_seconds_total{{instance="{inst}",mode="idle"}}[5m])) * 100)',
    "memory_usage": '(1 - (node_memory_MemAvailable_bytes{{instance="{inst}"}} / node_memory_MemTotal_bytes{{instance="{inst}"}})) * 100',
    "disk_usage": '100 - ((node_filesystem_avail_bytes{{instance="{inst}",mountpoint="/"}} / node_filesystem_size_bytes{{instance="{inst}",mountpoint="/"}}) * 100)',
}

# 要监控的服务器
SERVERS = [
    {"name": "app-server-01", "instance": "10.10.1.10:9100"},
    {"name": "app-server-02", "instance": "10.10.1.11:9100"},
    {"name": "db-server-01",  "instance": "10.10.1.20:9100"},
]


def query_prometheus(query, days=30):
    """从 Prometheus 查询过去 N 天的历史数据（每天一个数据点）"""
    end = datetime.now()
    start = end - timedelta(days=days)
    
    resp = requests.get(f"{PROMETHEUS_URL}/api/v1/query_range", params={
        "query": query,
        "start": start.timestamp(),
        "end": end.timestamp(),
        "step": "1d",  # 每天一个点
    })
    data = resp.json()
    if data["status"] != "success":
        return []
    
    result = data["data"]["result"]
    if not result:
        return []
    
    # 提取时间序列
    values = []
    for ts, val in result[0]["values"]:
        values.append({
            "date": datetime.fromtimestamp(float(ts)).strftime("%m-%d"),
            "value": float(val),
        })
    return values


def collect_capacity_data():
    """采集所有服务器的容量数据"""
    all_data = []
    for server in SERVERS:
        server_data = {"name": server["name"], "metrics": {}}
        for metric_name, query_template in METRICS_TO_TRACK.items():
            query = query_template.format(inst=server["instance"])
            history = query_prometheus(query)
            server_data["metrics"][metric_name] = history
        all_data.append(server_data)
        print(f"[OK] {server['name']} 数据采集完成")
    return all_data


PREDICT_PROMPT = """你是一个运维容量规划专家。以下是服务器过去 30 天的资源使用数据。

请分析：
1. **当前状态**：每台服务器的 CPU/内存/磁盘当前使用率
2. **增长趋势**：哪个指标在持续增长？日均增长率是多少？
3. **预测**：按当前趋势，预计多少天后会触及告警阈值（CPU>80%, 内存>85%, 磁盘>80%）？
4. **建议**：是否需要扩容？扩容方案是什么（加机器？升配？清理数据？）
5. **风险提示**：有没有突增迹象？是否需要关注特定业务事件的影响？

数据：
{data}

以 JSON 格式输出，包含每台服务器的分析结果。
"""


def ai_predict(data):
    """AI 分析容量趋势"""
    prompt = PREDICT_PROMPT.format(data=json.dumps(data, ensure_ascii=False, indent=2))
    response = client.chat.completions.create(
        model="qwen2.5-14b-instruct",
        messages=[
            {"role": "system", "content": "你是运维容量规划专家。"},
            {"role": "user", "content": prompt},
        ],
        temperature=0.3,
        max_tokens=4096,
    )
    return response.choices[0].message.content


if __name__ == "__main__":
    print("采集容量数据...")
    data = collect_capacity_data()
    
    print("\nAI 趋势分析中...")
    result = ai_predict(data)
    
    print("\n" + "=" * 60)
    print("容量规划报告")
    print("=" * 60)
    print(result)

#!/usr/bin/env python3
"""
AI 慢查询分析助手
- 从 MySQL slow log 采集慢查询
- AI 分析执行计划 + 给出优化建议
"""

import re
import json
import pymysql
from collections import defaultdict
from openai import OpenAI

client = OpenAI(base_url="http://10.10.100.50:8000/v1", api_key="internal")

# 数据库连接配置
DB_CONFIG = {
    "host": "10.10.1.20",
    "port": 3306,
    "user": "monitor",
    "password": "REPLACE_WITH_VAULT",
    "database": "mysql",
}


def get_slow_queries(limit=50):
    """从 performance_schema 获取慢查询"""
    conn = pymysql.connect(**DB_CONFIG)
    try:
        sql = """
        SELECT 
            DIGEST_TEXT as normalized_sql,
            COUNT_STAR as exec_count,
            AVG_TIMER_WAIT/1000000000 as avg_ms,
            SUM_ROWS_EXAMINED as total_rows_scanned,
            SUM_ROWS_SENT as total_rows_returned,
            FIRST_SEEN,
            LAST_SEEN
        FROM performance_schema.events_statements_summary_by_digest
        WHERE AVG_TIMER_WAIT/1000000000 > 1000  -- 平均执行时间 > 1秒
        ORDER BY AVG_TIMER_WAIT DESC
        LIMIT %s
        """
        cursor = conn.cursor(pymysql.cursors.DictCursor)
        cursor.execute(sql, (limit,))
        return cursor.fetchall()
    finally:
        conn.close()


def get_explain(sql_text):
    """获取查询的执行计划"""
    conn = pymysql.connect(**DB_CONFIG)
    try:
        cursor = conn.cursor(pymysql.cursors.DictCursor)
        # 注意：生产环境需要做 SQL 安全校验
        cursor.execute(f"EXPLAIN {sql_text}")
        return cursor.fetchall()
    except Exception as e:
        return [{"error": str(e)}]
    finally:
        conn.close()


ANALYSIS_PROMPT = """你是资深 DBA。请分析以下慢查询并给出优化建议。

## 慢查询信息
- 规范化 SQL: {sql}
- 执行次数: {count}
- 平均耗时: {avg_ms}ms
- 扫描行数: {rows_scanned}
- 返回行数: {rows_returned}
- 扫描/返回比: {scan_ratio}

## 执行计划
{explain}

请分析：
1. **性能瓶颈**：全表扫描？索引未命中？临时表？文件排序？
2. **扫描效率**：扫描/返回比是否过高？是否需要优化 LIMIT 或分页？
3. **索引建议**：建议创建什么索引？给出 CREATE INDEX 语句。
4. **SQL 改写**：是否可以改写 SQL 提升效率？
5. **影响评估**：这个查询影响什么业务？优化后预期提升多少？

以 JSON 格式输出。
"""


def analyze_slow_query(query):
    """AI 分析单条慢查询"""
    rows_scanned = query.get("total_rows_scanned", 0) or 0
    rows_returned = query.get("total_rows_returned", 0) or 1
    scan_ratio = rows_scanned / max(rows_returned, 1)
    
    # 获取执行计划
    explain = get_explain(query["normalized_sql"])
    
    prompt = ANALYSIS_PROMPT.format(
        sql=query["normalized_sql"][:500],
        count=query["exec_count"],
        avg_ms=round(query["avg_ms"], 2),
        rows_scanned=rows_scanned,
        rows_returned=rows_returned,
        scan_ratio=round(scan_ratio, 2),
        explain=json.dumps(explain, default=str, indent=2),
    )
    
    response = client.chat.completions.create(
        model="qwen2.5-14b-instruct",
        messages=[
            {"role": "system", "content": "你是资深 MySQL DBA。"},
            {"role": "user", "content": prompt},
        ],
        temperature=0.2,
        max_tokens=2048,
    )
    
    return response.choices[0].message.content


if __name__ == "__main__":
    print("采集慢查询...")
    slow_queries = get_slow_queries(limit=20)
    print(f"共 {len(slow_queries)} 条慢查询")
    
    results = []
    for i, q in enumerate(slow_queries, 1):
        print(f"[{i}/{len(slow_queries)}] 分析中: {q['normalized_sql'][:60]}...")
        analysis = analyze_slow_query(q)
        results.append({
            "sql": q["normalized_sql"],
            "avg_ms": round(q["avg_ms"], 2),
            "exec_count": q["exec_count"],
            "analysis": analysis,
        })
    
    # 生成报告
    print("\n" + "=" * 60)
    print("慢查询分析报告")
    print("=" * 60)
    for r in results:
        print(f"\n📋 平均 {r['avg_ms']}ms | 执行 {r['exec_count']} 次")
        print(f"SQL: {r['sql'][:100]}")
        print(f"分析: {r['analysis'][:200]}")

SELECT * FROM orders WHERE user_id = ? AND status = 1 ORDER BY create_time DESC LIMIT 20 OFFSET 10000;