1. 引言

在运维日常工作中，脚本编写往往占据大量时间：日志分析、批量部署、故障排查、数据备份……这些重复性工作不仅繁琐，还容易出错。随着 AI 编程工具的兴起，Codex 正在改变运维工程师的工作方式——它能把自然语言描述直接转化为可运行的脚本，让运维人员从"写代码"转向"提需求"。

本文将带你从零开始，掌握用 Codex 编写运维脚本的完整流程，涵盖环境准备、提示词技巧、实战案例与避坑指南。

2. Codex 简介与核心能力

2.1 什么是 Codex

Codex 是 OpenAI 推出的 AI 编程助手，能够理解自然语言指令并生成、修改、调试代码。它基于大规模代码语料训练，支持多种编程语言，尤其擅长 Python、Bash、PowerShell 等运维常用语言。

2.2 Codex 在运维场景的优势

• 快速生成脚本骨架：输入需求描述，秒级生成可运行脚本
• 自动处理边界情况：自动添加异常处理、日志记录、参数校验
• 多语言切换：同一需求可生成 Python、Bash、Go 等多种实现
• 代码解释与优化：对现有脚本进行重构、优化和注释

2.3 Codex 与 ChatGPT 的区别

ChatGPT 是通用对话助手，而 Codex 更专注于代码生成与执行，支持在沙箱环境中运行代码、查看执行结果并迭代修正，更适合实际工程场景。

3. 环境准备与工具链

3.1 注册与获取 API Key

访问 OpenAI 平台注册账号，在 API Keys 页面创建密钥。注意保管好密钥，避免泄露。

3.2 安装 Codex CLI

# 安装 Codex CLI
npm install -g @openai/codex

# 验证安装
codex --version

3.3 配置环境变量

export OPENAI_API_KEY="sk-xxxxx"

3.4 推荐的工作目录结构

~/ops-scripts/
├── scripts/          # 存放生成的脚本
├── logs/             # 运行日志
├── config/           # 配置文件
└── tests/            # 测试脚本

4. 提示词工程：让 Codex 听懂你的需求

4.1 提示词的基本结构

一个高质量的提示词应包含以下要素：

• 角色设定：告诉 Codex 它是什么角色
• 任务描述：清晰说明要做什么
• 输入输出：明确输入参数和期望输出
• 约束条件：指定语言、平台、错误处理方式

4.2 提示词模板示例

你是一名资深运维工程师，请用 Python 编写一个脚本：
- 功能：扫描指定目录下超过 100MB 的日志文件
- 输入：目录路径（命令行参数）
- 输出：文件列表及大小，按大小降序排列
- 要求：支持 --dry-run 参数，添加日志记录，处理权限异常

4.3 常见提示词技巧

• 分步拆解：复杂任务拆成多个小步骤逐步生成
• 提供示例：给出输入输出示例帮助 Codex 理解
• 迭代修正：生成结果不满意时，用追加指令修正
• 明确技术栈：指定 Python 3.10+、Bash 5.0 等版本要求

5. 实战案例一：日志分析脚本

5.1 需求描述

分析 Nginx 访问日志，统计每个 IP 的访问次数、状态码分布，并输出 Top 10 异常 IP。

5.2 提示词编写

请用 Python 编写日志分析脚本：
- 读取 /var/log/nginx/access.log
- 统计每个 IP 的请求次数
- 统计 HTTP 状态码分布（2xx/3xx/4xx/5xx）
- 输出访问量 Top 10 的 IP 及其状态码分布
- 使用正则表达式解析日志行
- 添加异常处理，日志文件不存在时给出友好提示

5.3 Codex 生成结果与解读

#!/usr/bin/env python3
"""Nginx 访问日志分析脚本"""

import re
import sys
from collections import Counter, defaultdict
from pathlib import Path

LOG_PATTERN = re.compile(
    r'(?P<ip>\d+\.\d+\.\d+\.\d+)\s+-\s+-\s+'
    r'\[(?P<time>[^\]]+)\]\s+'
    r'"(?P<request>[^"]+)"\s+'
    r'(?P<status>\d{3})'
)

def analyze_log(log_path: str) -> None:
    """分析日志文件，输出统计结果"""
    path = Path(log_path)
    if not path.exists():
        print(f"错误：日志文件 {log_path} 不存在", file=sys.stderr)
        sys.exit(1)

    ip_counter = Counter()
    status_counter = Counter()
    ip_status = defaultdict(Counter)

    with path.open('r', encoding='utf-8') as f:
        for line in f:
            match = LOG_PATTERN.search(line)
            if not match:
                continue
            ip = match.group('ip')
            status = match.group('status')
            ip_counter[ip] += 1
            status_counter[status[0] + 'xx'] += 1
            ip_status[ip][status] += 1

    print("=== 状态码分布 ===")
    for status, count in status_counter.most_common():
        print(f"{status}: {count}")

    print("\n=== Top 10 访问 IP ===")
    for ip, count in ip_counter.most_common(10):
        statuses = dict(ip_status[ip].most_common(3))
        print(f"{ip}: {count} 次, 状态码 {statuses}")

if __name__ == '__main__':
    if len(sys.argv) != 2:
        print(f"用法: {sys.argv[0]} <日志文件路径>", file=sys.stderr)
        sys.exit(1)
    analyze_log(sys.argv[1])

5.4 运行与验证

python3 analyze_log.py /var/log/nginx/access.log

6. 实战案例二：批量服务器状态检查

6.1 需求描述

编写脚本批量检查多台服务器的 CPU、内存、磁盘使用率，并汇总输出报告。

6.2 提示词编写

请用 Bash 编写服务器批量检查脚本：
- 从 servers.txt 读取服务器列表（每行一个 IP）
- 通过 SSH 连接每台服务器
- 执行命令获取 CPU 使用率、内存使用率、磁盘使用率
- 输出格式化的汇总表格
- 支持超时设置，避免卡死
- 服务器连接失败时记录错误并继续

6.3 Codex 生成结果与解读

#!/bin/bash
# 批量服务器状态检查脚本

SERVERS_FILE="servers.txt"
TIMEOUT=5

echo "=== 服务器状态检查报告 ==="
echo "时间: $(date '+%Y-%m-%d %H:%M:%S')"
echo "----------------------------------------"
printf "%-15s %-8s %-8s %-8s\n" "服务器" "CPU%" "内存%" "磁盘%"
echo "----------------------------------------"

while IFS= read -r server; do
    [ -z "$server" ] && continue
    result=$(ssh -o ConnectTimeout=$TIMEOUT -o StrictHostKeyChecking=no \
        "$server" \
        "cpu=\$(top -bn1 | grep 'Cpu(s)' | awk '{print \$2}' | cut -d'%' -f1); \
         mem=\$(free | awk '/Mem:/ {printf \"%.1f\", \$3/\$2*100}'); \
         disk=\$(df -h / | awk 'NR==2 {print \$5}' | tr -d '%'); \
         echo \"\$cpu \$mem \$disk\"" 2>/dev/null)

    if [ -n "$result" ]; then
        read -r cpu mem disk <<< "$result"
        printf "%-15s %-8s %-8s %-8s\n" "$server" "$cpu%" "$mem%" "$disk%"
    else
        printf "%-15s %-8s\n" "$server" "连接失败"
    fi
done < "$SERVERS_FILE"

6.4 运行与验证

chmod +x check_servers.sh
./check_servers.sh

7. 实战案例三：自动化备份脚本

7.1 需求描述

编写数据库自动备份脚本，支持全量备份、增量备份、备份文件保留策略。

7.2 提示词编写

请用 Python 编写 MySQL 自动备份脚本：
- 支持全量备份和增量备份两种模式
- 备份文件按日期命名，存放到指定目录
- 自动清理超过 7 天的备份文件
- 备份完成后发送邮件通知
- 使用配置文件管理数据库连接信息
- 添加日志记录，记录每次备份的开始、结束、耗时

7.3 Codex 生成结果与解读

#!/usr/bin/env python3
"""MySQL 自动备份脚本"""

import configparser
import datetime
import logging
import os
import smtplib
import subprocess
import sys
from email.mime.text import MIMEText
from pathlib import Path

# 配置日志
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s [%(levelname)s] %(message)s',
    handlers=[
        logging.FileHandler('backup.log'),
        logging.StreamHandler()
    ]
)

def load_config(config_path: str) -> dict:
    """加载配置文件"""
    config = configparser.ConfigParser()
    config.read(config_path)
    return {
        'host': config['mysql']['host'],
        'user': config['mysql']['user'],
        'password': config['mysql']['password'],
        'database': config['mysql']['database'],
        'backup_dir': config['backup']['dir'],
        'retention_days': int(config['backup']['retention_days']),
        'smtp_server': config['mail']['smtp_server'],
        'smtp_port': int(config['mail']['smtp_port']),
        'mail_from': config['mail']['from'],
        'mail_to': config['mail']['to'],
    }

def run_backup(config: dict, mode: str = 'full') -> Path:
    """执行备份"""
    timestamp = datetime.datetime.now().strftime('%Y%m%d_%H%M%S')
    backup_file = Path(config['backup_dir']) / f"{config['database']}_{mode}_{timestamp}.sql"

    cmd = [
        'mysqldump',
        '-h', config['host'],
        '-u', config['user'],
        f"-p{config['password']}",
        '--single-transaction',
        '--routines',
        '--triggers',
    ]
    if mode == 'incremental':
        cmd.append('--incremental')
    cmd.append(config['database'])

    logging.info(f"开始{mode}备份: {backup_file.name}")
    with backup_file.open('w') as f:
        result = subprocess.run(cmd, stdout=f, stderr=subprocess.PIPE)
    if result.returncode != 0:
        logging.error(f"备份失败: {result.stderr.decode()}")
        raise RuntimeError("备份失败")
    logging.info(f"备份完成: {backup_file.name}")
    return backup_file

def cleanup_old_backups(config: dict) -> None:
    """清理过期备份"""
    backup_dir = Path(config['backup_dir'])
    cutoff = datetime.datetime.now() - datetime.timedelta(days=config['retention_days'])
    for f in backup_dir.glob('*.sql'):
        mtime = datetime.datetime.fromtimestamp(f.stat().st_mtime)
        if mtime < cutoff:
            f.unlink()
            logging.info(f"清理过期备份: {f.name}")

def send_notification(config: dict, subject: str, body: str) -> None:
    """发送邮件通知"""
    msg = MIMEText(body)
    msg['Subject'] = subject
    msg['From'] = config['mail_from']
    msg['To'] = config['mail_to']
    with smtplib.SMTP(config['smtp_server'], config['smtp_port']) as server:
        server.send_message(msg)
    logging.info("邮件通知已发送")

if __name__ == '__main__':
    if len(sys.argv) < 2:
        print(f"用法: {sys.argv[0]} <配置文件> [full|incremental]", file=sys.stderr)
        sys.exit(1)
    cfg = load_config(sys.argv[1])
    mode = sys.argv[2] if len(sys.argv) > 2 else 'full'
    try:
        backup_file = run_backup(cfg, mode)
        cleanup_old_backups(cfg)
        send_notification(cfg, "备份完成", f"备份文件: {backup_file}")
    except Exception as e:
        send_notification(cfg, "备份失败", str(e))
        sys.exit(1)

7.4 运行与验证

python3 backup.py config.ini full

8. 常见问题与避坑指南

8.1 提示词不清晰导致结果偏差

• 问题：需求描述模糊，Codex 生成结果与预期不符
• 解决：使用结构化提示词模板，明确输入输出和约束条件

8.2 生成的脚本缺少错误处理

• 问题：脚本在异常场景下崩溃
• 解决：在提示词中明确要求添加异常处理、参数校验、日志记录

8.3 安全风险

• 问题：脚本中硬编码密码、密钥等敏感信息
• 解决：使用环境变量或配置文件管理敏感信息，避免写入代码

8.4 版本兼容性问题

• 问题：生成的代码依赖特定版本库，在目标环境无法运行
• 解决：在提示词中指定 Python 版本、依赖库版本，并在测试环境验证

8.5 过度依赖 AI 生成

• 问题：完全信任生成结果，不进行代码审查
• 解决：建立代码审查流程，关键脚本必须人工 review 后再上线

9. 最佳实践与效率提升

9.1 建立提示词模板库

将常用运维场景的提示词沉淀为模板，形成团队知识库，提高复用效率。

9.2 结合版本控制

将生成的脚本纳入 Git 管理，记录每次修改，便于回溯和协作。

9.3 自动化测试

为生成的脚本编写单元测试和集成测试，确保功能正确性。

9.4 持续迭代优化

将 Codex 生成的脚本作为初稿，结合人工 review 持续优化，形成"AI 生成 + 人工打磨"的工作流。

10. 总结与展望

Codex 正在重塑运维工程师的工作方式。通过本文的实战案例，你已经掌握了用 Codex 编写日志分析、服务器检查、自动备份等运维脚本的核心方法。关键在于：

• 写好提示词：清晰、结构化地表达需求
• 验证生成结果：不盲信 AI，建立审查机制
• 沉淀最佳实践：将经验固化为模板和流程

未来，随着 AI 编程工具的持续进化，运维工作将更加智能化、自动化。掌握 Codex，就是掌握运维自动化的下一个风口。