当 AI 成为技术人标配工具：覆盖开发、运维、测试、网工全场景

@RestController
@RequestMapping("/api/users")
public class UserController {

    @Autowired
    private UserService userService;

    /**
     * 用户注册接口
     * AI 根据需求文档自动生成的完整参数校验 + 异常处理
     */
    @PostMapping("/register")
    public ResponseEntity<ApiResponse<UserVO>> register(
            @RequestBody @Valid RegisterRequest request,
            BindingResult bindingResult) {

        // 1. JSR-303 注解校验失败的第一道防线
        if (bindingResult.hasErrors()) {
            String errorMsg = bindingResult.getFieldErrors().stream()
                .map(e -> e.getField() + ": " + e.getDefaultMessage())
                .collect(Collectors.joining("; "));
            return ResponseEntity.badRequest()
                .body(ApiResponse.fail(400, errorMsg));
        }

        // 2. 业务层校验——AI 建议补充的边界场景
        // 手机号归属地校验
        if (!PhoneUtil.isValidChinaMobile(request.getPhone())) {
            return ResponseEntity.badRequest()
                .body(ApiResponse.fail(400, "手机号格式不正确或非中国大陆号码"));
        }

        // 密码强度校验——AI 指出弱密码是注册接口最常见的安全风险
        PasswordStrength strength = PasswordUtil.checkStrength(request.getPassword());
        if (strength.getScore() < 3) {
            return ResponseEntity.badRequest()
                .body(ApiResponse.fail(400, 
                    "密码强度不足，建议：" + strength.getSuggestion()));
        }

        // 防刷：同一 IP 60 秒内只能注册一次
        String clientIp = RequestUtil.getClientIp();
        String rateLimitKey = "register:rate:" + clientIp;
        if (!redisTemplate.opsForValue().setIfAbsent(
                rateLimitKey, "1", 60, TimeUnit.SECONDS)) {
            return ResponseEntity.status(429)
                .body(ApiResponse.fail(429, "操作太频繁，请稍后再试"));
        }

        // 3. 调用 Service
        UserVO userVO = userService.register(request);
        return ResponseEntity.ok(ApiResponse.success(userVO));
    }
}

/**
 * AI 生成的请求 DTO，包含完整的校验注解
 * 关键：每个字段的校验规则都有注释说明"为什么这么校验"
 */
@Data
public class RegisterRequest {

    @NotBlank(message = "用户名不能为空")
    @Size(min = 3, max = 20, message = "用户名长度3-20个字符")
    @Pattern(regexp = "^[a-zA-Z0-9_]+$", 
             message = "用户名只能包含字母、数字、下划线")
    private String username;

    @NotBlank(message = "密码不能为空")
    @Size(min = 8, max = 64, message = "密码长度8-64个字符")
    private String password;

    @NotBlank(message = "手机号不能为空")
    @Pattern(regexp = "^1[3-9]\\d{9}$", message = "手机号格式不正确")
    private String phone;

    @Email(message = "邮箱格式不正确")
    private String email;

    // AI 指出：很多团队漏了这一层校验，导致脏数据入库
    @Size(max = 100, message = "昵称长度不能超过100")
    private String nickname;
}

import subprocess
import json
import requests

# 获取当前 PR 的 diff
def get_pr_diff(pr_number):
    result = subprocess.run(
        ["git", "diff", f"origin/main...HEAD"],
        capture_output=True, text=True, encoding="utf-8"
    )
    return result.stdout

# 团队编码规范（精简版，喂给 AI 当上下文）
TEAM_CODING_STANDARDS = """
1. 所有外部输入必须经过校验，不允许直接信任前端参数
2. 数据库查询禁止使用字符串拼接SQL，必须用参数化查询
3. 敏感信息（密码、token）不允许出现在日志中
4. 所有 Controller 方法必须有权限注解
5. 循环内禁止执行数据库查询
6. 异常必须分类处理，不允许 catch(Exception) 后吞掉
7. 新增 API 必须有对应的单元测试
"""

def ai_code_review(diff_content):
    prompt = f"""你是一位资深 Java Code Reviewer，请审查以下代码变更。

团队编码规范：
{TEAM_CODING_STANDARDS}

代码变更：
{diff_content[:8000]}

请按以下格式输出审查结果：
1. 【严重问题】可能导致 bug 或安全风险的代码
2. 【规范问题】违反团队编码规范的代码
3. 【建议改进】可以写得更好的地方
4. 【通过项】做得好的地方（简述即可）

注意：只关注有实质内容的变更行，忽略 import 语句和空行。"""

    response = requests.post(
        "http://localhost:8000/v1/chat/completions",
        json={
            "model": "qwen2.5-coder-32b",
            "messages": [{"role": "user", "content": prompt}],
            "temperature": 0.3,
        },
        timeout=120
    )
    return response.json()["choices"][0]["message"]["content"]

# 主流程
diff = get_pr_diff(123)
review_result = ai_code_review(diff)
print(review_result)

# 真实输出示例：
# 【严重问题】
# - UserController.java 第45行：phone 参数直接拼接到日志中，
#   如果包含特殊字符可能导致日志注入，建议用 PhoneUtil.sanitize() 过滤
# - OrderMapper.xml 第12行：使用了 ${orderId} 而非 #{orderId}，
#   存在 SQL 注入风险，必须修复
#
# 【规范问题】
# - PaymentController.java：缺少 @PreAuthorize 权限注解
# - UserService.java 第88行：for 循环内调用了 userMapper.selectById()，
#   违反规范第5条

import re
from collections import defaultdict
from datetime import datetime, timedelta
import requests

class AlertCorrelator:
    """告警聚合器：把短时间内相关的告警归到一起"""

    def __init__(self, window_seconds=300):
        self.window = timedelta(seconds=window_seconds)
        self.alert_buffer = []

    def add_alert(self, alert):
        """添加一条告警，返回所属的事件组ID"""
        now = datetime.now()

        # 清理过期告警
        self.alert_buffer = [
            a for a in self.alert_buffer
            if now - a["timestamp"] < self.window
        ]

        # 尝试匹配已有事件组
        group_id = self._find_matching_group(alert)

        if group_id is None:
            group_id = f"INC-{now.strftime('%Y%m%d%H%M%S')}-{len(self.alert_buffer)}"

        alert["group_id"] = group_id
        alert["timestamp"] = now
        self.alert_buffer.append(alert)

        return group_id

    def _find_matching_group(self, alert):
        """简单规则：同一主机/服务的告警归为一组"""
        for existing in reversed(self.alert_buffer):
            if (existing.get("host") == alert.get("host") or
                existing.get("service") == alert.get("service")):
                return existing["group_id"]
        return None


class AIDiagnosticEngine:
    """AI 诊断引擎：分析告警事件组，输出根因判断和处置建议"""

    def __init__(self, model_url, model_name):
        self.url = model_url
        self.model = model_name

    def diagnose(self, alert_group):
        """对一组关联告警进行诊断"""
        alerts_summary = "\n".join([
            f"- [{a['severity']}] {a['service']}: {a['message']}"
            for a in alert_group
        ])

        prompt = f"""你是一位有 10 年经验的运维专家，请分析以下告警事件组，给出诊断。

告警事件组（同一时间窗口内的关联告警）：
{alerts_summary}

请输出：
1. 【根因判断】最可能的根本原因（一句话）
2. 【置信度】高/中/低
3. 【影响评估】影响哪些业务，影响范围多大
4. 【处置建议】按优先级排列的行动项
5. 【是否需要升级】是否需要通知上级或拉群

格式要求：简洁，不要废话，运维看的是效率不是文采。"""

        response = requests.post(
            f"{self.url}/v1/chat/completions",
            json={
                "model": self.model,
                "messages": [{"role": "user", "content": prompt}],
                "temperature": 0.2,
            },
            timeout=60
        )
        return response.json()["choices"][0]["message"]["content"]


# ===== 实战运行 =====
correlator = AlertCorrelator(window_seconds=180)
ai_engine = AIDiagnosticEngine(
    model_url="http://localhost:8000",
    model_name="qwen2.5-32b"
)

# 模拟一次告警风暴：交换机端口抖动导致连锁反应
raw_alerts = [
    {"service": "switch-core-01", "severity": "critical",
     "host": "10.0.1.1", "message": "端口 Gi0/24 link down"},
    {"service": "switch-core-01", "severity": "warning",
     "host": "10.0.1.1", "message": "端口 Gi0/24 link up"},
    {"service": "switch-core-01", "severity": "critical",
     "host": "10.0.1.1", "message": "端口 Gi0/24 link down"},
    {"service": "nginx-web-01", "severity": "critical",
     "host": "10.0.2.5", "message": "upstream timeout, backend unreachable"},
    {"service": "nginx-web-02", "severity": "critical",
     "host": "10.0.2.6", "message": "upstream timeout, backend unreachable"},
    {"service": "order-service", "severity": "critical",
     "host": "10.0.3.10", "message": "健康检查失败，实例被踢出注册中心"},
    {"service": "payment-service", "severity": "warning",
     "host": "10.0.3.11", "message": "调用 order-service 超时率 35%"},
]

# 聚合告警
groups = defaultdict(list)
for alert in raw_alerts:
    gid = correlator.add_alert(alert)
    groups[gid].append(alert)

# 对每个事件组做 AI 诊断
for gid, alerts in groups.items():
    print(f"\n{'='*60}")
    print(f"事件组: {gid} ({len(alerts)} 条告警)")
    print(f"{'='*60}")
    diagnosis = ai_engine.diagnose(alerts)
    print(diagnosis)

# AI 输出示例：
# 1.【根因判断】核心交换机 switch-core-01 的 Gi0/24 端口物理链路不稳定，
#   导致下游 web 服务器和微服务不可达
# 2.【置信度】高
# 3.【影响评估】影响订单和支付服务，核心交易链路受损
# 4.【处置建议】
#   a. 立即检查 Gi0/24 物理连接（光模块/网线）
#   b. 如短期无法恢复，将流量切换至备用链路
#   c. 通知业务侧暂停大促活动
# 5.【是否需要升级】是，核心交易链路受损，建议立即拉群

import json
import requests
import subprocess

class AITestGenerator:
    """AI 测试用例生成器：根据接口变更生成回归用例"""

    def __init__(self, model_url, model_name):
        self.url = model_url
        self.model = model_name

    def get_api_diff(self, old_spec, new_spec):
        """对比新旧接口定义，提取变更"""
        changes = []

        old_paths = set(old_spec.get("paths", {}).keys())
        new_paths = set(new_spec.get("paths", {}).keys())

        # 新增的接口
        for path in new_paths - old_paths:
            for method in new_spec["paths"][path]:
                changes.append({
                    "type": "new_api",
                    "path": path,
                    "method": method,
                    "spec": new_spec["paths"][path][method]
                })

        # 变更的接口
        for path in old_paths & new_paths:
            for method in new_spec["paths"][path]:
                old_def = old_spec["paths"][path].get(method, {})
                new_def = new_spec["paths"][path].get(method, {})
                if old_def != new_def:
                    changes.append({
                        "type": "modified_api",
                        "path": path,
                        "method": method,
                        "old_spec": old_def,
                        "new_spec": new_def
                    })

        return changes

    def generate_test_cases(self, changes):
        """根据接口变更生成测试用例"""
        changes_desc = json.dumps(changes, ensure_ascii=False, indent=2)

        prompt = f"""你是资深测试工程师，请根据以下接口变更生成回归测试用例。

接口变更内容：
{changes_desc}

要求：
1. 为每个变更接口生成正常流程、边界值、异常场景用例
2. 特别关注参数类型变更、必填项变更、权限变更
3. 每条用例包含：用例名称、前置条件、请求参数、预期结果、优先级
4. 输出 JSON 数组格式

输出示例：
[{{"name": "注册-正常流程", "precondition": "数据库无该用户", 
   "params": {{"username": "testuser", "password": "Test@1234"}}, 
   "expected": {{"code": 200, "msg": "success"}}, "priority": "P0"}}]"""

        response = requests.post(
            f"{self.url}/v1/chat/completions",
            json={
                "model": self.model,
                "messages": [{"role": "user", "content": prompt}],
                "temperature": 0.4,
            },
            timeout=120
        )
        return response.json()["choices"][0]["message"]["content"]

    def generate_playwright_script(self, test_cases):
        """把用例转成 Playwright 自动化脚本"""
        prompt = f"""请把以下测试用例转换成 Playwright (Python) 自动化脚本。

测试用例：
{test_cases}

要求：
1. 使用 pytest 框架
2. 每条用例一个 test 函数
3. 包含断言
4. 失败时截图保存
5. 加上中文注释"""

        response = requests.post(
            f"{self.url}/v1/chat/completions",
            json={
                "model": self.model,
                "messages": [{"role": "user", "content": prompt}],
                "temperature": 0.3,
            },
            timeout=120
        )
        return response.json()["choices"][0]["message"]["content"]


# ===== 实战运行 =====
generator = AITestGenerator(
    model_url="http://localhost:8000",
    model_name="qwen2.5-coder-32b"
)

# 对比新旧 API 文档（这里用简化数据演示）
old_api_spec = {
    "paths": {
        "/api/users/register": {
            "post": {
                "summary": "用户注册",
                "requestBody": {
                    "username": "string, required",
                    "password": "string, required, min 6",
                    "phone": "string, required"
                }
            }
        }
    }
}

new_api_spec = {
    "paths": {
        "/api/users/register": {
            "post": {
                "summary": "用户注册",
                "requestBody": {
                    "username": "string, required, 3-20 chars",
                    "password": "string, required, min 8, need upper+lower+digit",
                    "phone": "string, required",
                    "email": "string, optional"
                }
            }
        },
        "/api/users/profile": {
            "get": {
                "summary": "获取用户资料",
                "parameters": {
                    "Authorization": "header, required"
                }
            }
        }
    }
}

# Step 1: 提取变更
changes = generator.get_api_diff(old_api_spec, new_api_spec)
print(f"检测到 {len(changes)} 个接口变更")

# Step 2: 生成用例
test_cases = generator.generate_test_cases(changes)
print("\n生成的测试用例：")
print(test_cases)

# Step 3: 生成自动化脚本
playwright_script = generator.generate_playwright_script(test_cases)

# 保存到文件
with open("test_register_regression.py", "w", encoding="utf-8") as f:
    f.write(playwright_script)

print("\n自动化脚本已保存到 test_register_regression.py")

from netmiko import ConnectHandler
from concurrent.futures import ThreadPoolExecutor, as_completed
import json
import re
import requests

class NetworkInspector:
    """跨厂商网络设备巡检器"""

    # 不同厂商的命令映射
    COMMAND_MAP = {
        "cisco_ios": {
            "interface": "show interfaces",
            "cpu": "show processes cpu sorted",
            "memory": "show memory statistics",
            "log": "show logging | last 50",
            "arp": "show arp",
            "route": "show ip route summary"
        },
        "huawei": {
            "interface": "display interface",
            "cpu": "display cpu-usage",
            "memory": "display memory-usage",
            "log": "display logbuffer | last 50",
            "arp": "display arp",
            "route": "display ip routing-table statistics"
        },
        "hp_comware": {
            "interface": "display interface",
            "cpu": "display cpu-usage",
            "memory": "display memory",
            "log": "display logbuffer reverse | last 50",
            "arp": "display arp",
            "route": "display ip routing-table statistics"
        }
    }

    def inspect_device(self, device_info):
        """巡检单台设备"""
        device_type = device_info["device_type"]
        commands = self.COMMAND_MAP.get(device_type, {})

        results = {"device": device_info["host"], "data": {}}

        try:
            with ConnectHandler(**device_info) as conn:
                for category, cmd in commands.items():
                    output = conn.send_command(cmd, read_timeout=30)
                    results["data"][category] = output

            results["status"] = "success"
        except Exception as e:
            results["status"] = "failed"
            results["error"] = str(e)

        return results

    def batch_inspect(self, device_list, max_workers=10):
        """批量并行巡检"""
        all_results = []

        with ThreadPoolExecutor(max_workers=max_workers) as executor:
            futures = {
                executor.submit(self.inspect_device, dev): dev["host"]
                for dev in device_list
            }

            for future in as_completed(futures):
                host = futures[future]
                try:
                    result = future.result()
                    all_results.append(result)
                except Exception as e:
                    all_results.append({
                        "device": host,
                        "status": "failed",
                        "error": str(e)
                    })

        return all_results


class AINetworkAnalyzer:
    """AI 网络异常分析器"""

    def __init__(self, model_url, model_name):
        self.url = model_url
        self.model = model_name

    def analyze(self, inspection_results):
        """分析巡检结果，找出潜在问题"""
        # 把巡检数据整理成 AI 能看的格式
        device_reports = []
        for r in inspection_results:
            if r["status"] != "success":
                device_reports.append(
                    f"设备 {r['device']}: 巡检失败，错误: {r.get('error', '未知')}")
                continue

            # 只提取关键信息，避免 prompt 太长
            iface_data = r["data"].get("interface", "")[:2000]
            cpu_data = r["data"].get("cpu", "")[:500]
            mem_data = r["data"].get("memory", "")[:500]
            log_data = r["data"].get("log", "")[:1500]

            device_reports.append(
                f"设备 {r['device']}:\n"
                f"[接口状态]\n{iface_data}\n\n"
                f"[CPU使用率]\n{cpu_data}\n\n"
                f"[内存使用率]\n{mem_data}\n\n"
                f"[最近日志]\n{log_data}"
            )

        combined_report = "\n\n---\n\n".join(device_reports)

        prompt = f"""你是一位有 15 年经验的网络工程师，请分析以下网络设备巡检数据，找出潜在问题。

巡检数据：
{combined_report[:12000]}

请输出：
1. 【严重问题】需要立即处理的（如链路 down、CRC 错误持续增长、CPU > 80%）
2. 【潜在风险】目前不影响但可能恶化的（如内存缓慢增长、错误包有上升趋势）
3. 【建议操作】针对每个问题的具体处理建议
4. 【健康度评分】每台设备的健康度（0-100分）

注意：重点关注接口的 CRC 错误、丢包率、CPU 和内存趋势。"""

        response = requests.post(
            f"{self.url}/v1/chat/completions",
            json={
                "model": self.model,
                "messages": [{"role": "user", "content": prompt}],
                "temperature": 0.2,
            },
            timeout=90
        )
        return response.json()["choices"][0]["message"]["content"]


# ===== 实战运行 =====
devices = [
    {"device_type": "cisco_ios", "host": "10.0.1.1",
     "username": "admin", "password": "***", "port": 22},
    {"device_type": "huawei", "host": "10.0.1.2",
     "username": "admin", "password": "***", "port": 22},
    {"device_type": "hp_comware", "host": "10.0.1.3",
     "username": "admin", "password": "***", "port": 22},
]

inspector = NetworkInspector()
analyzer = AINetworkAnalyzer(
    model_url="http://localhost:8000",
    model_name="qwen2.5-32b"
)

# Step 1: 批量巡检
print("开始巡检...")
results = inspector.batch_inspect(devices)
print(f"巡检完成: {len(results)} 台设备")

# Step 2: AI 分析
print("\nAI 分析中...")
analysis = analyzer.analyze(results)
print(analysis)

# AI 输出示例：
# 1.【严重问题】
# 设备 10.0.1.1 (Cisco):
#   - Gi0/24 接口 CRC 错误率异常，过去1小时增长 152 个，
#     结合 input errors 持续增长，大概率光模块或光纤物理层问题
#   - 建议：立即检查光模块发光功率，准备备件更换
#
# 设备 10.0.1.2 (Huawei):
#   - CPU 使用率 85%，持续 10 分钟以上
#   - 日志显示有大量 ARP 请求，可能是 ARP 攻击或环路
#   - 建议：检查 ARP 表项，排查是否有环路
#
# 2.【潜在风险】
# 设备 10.0.1.3 (H3C):
#   - 内存使用率 72%，虽未告警但过去一周呈缓慢上升趋势
#   - 建议：关注内存趋势，排查是否有内存泄漏的进程
#
# 4.【健康度评分】
# 10.0.1.1: 65分（有严重接口问题）
# 10.0.1.2: 55分（CPU 高且有潜在安全风险）
# 10.0.1.3: 85分（正常，需关注内存趋势）