告别重复造轮子：用 Codex 批量生成 Shell/Python 运维自动化脚本
·
告别重复造轮子：用 Codex 批量生成 Shell/Python 运维自动化脚本
💡 摘要: 本文深入讲解如何使用 AI 代码助手（Codex/Cursor/GitHub Copilot）批量生成高质量的 Shell 和 Python 运维自动化脚本，涵盖提示词工程、代码验证、批量生成技巧及企业级最佳实践。通过 5 个真实场景案例（日志分析、批量部署、监控告警、数据备份、安全加固），带你掌握 AI 辅助运维开发的全套技能。包含完整的 Prompt 模板、10 个提效技巧和常见问题解决方案，助力运维效率提升 20 倍以上。
🎯 场景化开篇：运维脚本开发的痛点
凌晨 2 点的脚本编写噩梦
时间: 2025 年 3 月 15 日，周六凌晨 2:30
事件: 紧急处理生产环境磁盘空间告警
故障现场还原:
📞 2:30 AM - 监控告警：10 台服务器磁盘使用率超过 90%
🔴 传统做法（手工编写脚本）:
1. SSH 登录每台服务器检查大文件
2. 手动编写清理脚本（耗时 1 小时）
3. 逐台执行并验证（耗时 2 小时）
4. 发现脚本有 Bug，重新修改（耗时 30 分钟）
5. 再次执行并验证（耗时 1 小时）
总计：5 小时 → 早上 7:30 才处理完 ❌
💥 痛点:
- 重复性工作多（80% 的脚本都是类似逻辑）
- 语法记忆困难（awk/sed 复杂语法记不住）
- 调试耗时长（一个小错误排查半小时）
- 文档缺失（3 个月后自己都看不懂）
如果有 AI 代码助手...
✅ 描述需求："编写一个脚本，查找 /var/log 下超过 1GB 的文件并压缩"
✅ AI 生成完整脚本（30 秒）
✅ 自动添加注释和错误处理
✅ 一键测试并部署
✅ 总耗时：5 分钟 ✅
📖 Codex 与 AI 代码助手简介
什么是 Codex？
Codex 是 OpenAI 开发的 AI 代码生成模型，能够根据自然语言描述自动生成代码。目前主流的 AI 代码助手包括：
| 工具 | 开发商 | 特点 | 适用场景 |
|---|---|---|---|
| GitHub Copilot | GitHub/Microsoft | IDE 集成好，智能补全 | 日常编码 |
| Cursor | Cursor.sh | 对话式编程，理解上下文 | 复杂任务 |
| 通义灵码 | 阿里云 | 中文支持好，免费 | 国内用户 |
| CodeWhisperer | AWS | 云原生优化 | AWS 环境 |
| Tabnine | Tabnine | 本地部署，隐私性好 | 企业内网 |
AI 辅助运维开发的价值
效率对比:
- ⏱️ 开发速度: 提升 10-20 倍
- 📝 代码质量: 标准化、注释完善
- 🐛 Bug 率: 降低 60%+
- 📚 学习成本: 新手也能写出专业脚本
🔧 核心技巧：提示词工程（Prompt Engineering）
1. 结构化 Prompt 模板
# Role（角色）
你是一位拥有 10 年经验的 Linux 运维专家，擅长编写高效、安全的 Shell/Python 脚本。
# Task（任务）
请编写一个 [脚本类型] 脚本，用于 [具体功能]。
# Requirements（要求）
1. 功能需求：
- [需求 1]
- [需求 2]
- [需求 3]
2. 技术要求：
- 使用 [Shell/Python] 语言
- 包含完善的错误处理
- 添加详细的中文注释
- 遵循 [规范名称] 规范
3. 输出要求：
- 提供完整的可运行代码
- 包含使用说明
- 提供测试方法
# Context（背景）
- 操作系统：[CentOS 7/Ubuntu 20.04]
- 目标用户：运维工程师
- 执行频率：[每天/每周/一次性]
# Example（示例）
输入：[示例输入]
输出：[期望输出]
2. 实战 Prompt 示例
示例 1：日志清理脚本
你是一位 Linux 运维专家。请编写一个 Shell 脚本，用于清理 Nginx 访问日志。
要求：
1. 查找 /var/log/nginx/ 目录下所有超过 7 天的 .log 文件
2. 使用 gzip 压缩这些文件
3. 删除超过 30 天的压缩文件
4. 记录操作日志到 /var/log/cleanup.log
5. 包含完善的错误处理和权限检查
6. 添加详细的中文注释
7. 支持 --dry-run 参数预览操作
操作系统：CentOS 7
执行频率：每天凌晨 2 点（crontab）
请提供完整的脚本代码和使用说明。
AI 生成的代码:
#!/bin/bash
# =====================================================
# Nginx 日志清理脚本
# 功能：压缩旧日志文件，删除过期压缩文件
# 作者：AI Assistant
# 日期：2025-03-15
# =====================================================
set -euo pipefail # 严格模式
# 配置变量
LOG_DIR="/var/log/nginx"
CLEANUP_LOG="/var/log/cleanup.log"
COMPRESS_DAYS=7 # 压缩超过 7 天的日志
DELETE_DAYS=30 # 删除超过 30 天的压缩文件
DRY_RUN=false # 是否为预览模式
# 解析命令行参数
while [[ $# -gt 0 ]]; do
case $1 in
--dry-run)
DRY_RUN=true
shift
;;
*)
echo "未知参数: $1"
exit 1
;;
esac
done
# 日志函数
log() {
local message="[$(date '+%Y-%m-%d %H:%M:%S')] $*"
echo "$message" | tee -a "$CLEANUP_LOG"
}
# 检查权限
check_permissions() {
if [[ $EUID -ne 0 ]]; then
log "错误：需要 root 权限执行此脚本"
exit 1
fi
if [[ ! -d "$LOG_DIR" ]]; then
log "错误：日志目录不存在: $LOG_DIR"
exit 1
fi
}
# 压缩旧日志文件
compress_old_logs() {
log "开始压缩超过 ${COMPRESS_DAYS} 天的日志文件..."
local count=0
while IFS= read -r -d '' file; do
if [[ "$DRY_RUN" == true ]]; then
log "[预览] 将压缩: $file"
else
if gzip "$file"; then
log "已压缩: $file"
((count++))
else
log "警告：压缩失败: $file"
fi
fi
done < <(find "$LOG_DIR" -name "*.log" -type f -mtime +${COMPRESS_DAYS} -print0)
log "压缩完成，共处理 $count 个文件"
}
# 删除过期压缩文件
delete_expired_archives() {
log "开始删除超过 ${DELETE_DAYS} 天的压缩文件..."
local count=0
while IFS= read -r -d '' file; do
if [[ "$DRY_RUN" == true ]]; then
log "[预览] 将删除: $file"
else
if rm -f "$file"; then
log "已删除: $file"
((count++))
else
log "警告：删除失败: $file"
fi
fi
done < <(find "$LOG_DIR" -name "*.gz" -type f -mtime +${DELETE_DAYS} -print0)
log "删除完成，共处理 $count 个文件"
}
# 主函数
main() {
log "========== Nginx 日志清理开始 =========="
check_permissions
compress_old_logs
delete_expired_archives
log "========== Nginx 日志清理完成 =========="
}
# 执行主函数
main "$@"
Crontab 配置:
# 每天凌晨 2 点执行日志清理
0 2 * * * /opt/scripts/nginx_log_cleanup.sh >> /var/log/nginx_cleanup_cron.log 2>&1
🎯 五大典型场景实战
场景一：批量服务器健康检查
Prompt
你是一位资深运维工程师。请编写一个 Python 脚本，用于批量检查服务器健康状态。
要求：
1. 从 hosts.txt 文件读取服务器列表（每行一个 IP）
2. 对每台服务器执行以下检查：
- Ping 连通性
- SSH 连接测试
- CPU 使用率（阈值 80%）
- 内存使用率（阈值 85%）
- 磁盘使用率（阈值 90%）
3. 生成 HTML 格式的检查报告
4. 异常的服务器发送钉钉告警
5. 支持并发检查（ThreadPoolExecutor）
6. 包含完善的异常处理和日志记录
技术栈：Python 3.8+, paramiko, requests
输出：HTML 报告 + 钉钉告警
AI 生成的核心代码
#!/usr/bin/env python3
"""
服务器健康检查脚本
功能：批量检查服务器状态，生成 HTML 报告，发送钉钉告警
"""
import subprocess
import paramiko
import requests
import json
from concurrent.futures import ThreadPoolExecutor, as_completed
from datetime import datetime
from pathlib import Path
class ServerHealthChecker:
"""服务器健康检查器"""
def __init__(self, hosts_file: str, dingtalk_webhook: str):
self.hosts = self._load_hosts(hosts_file)
self.dingtalk_webhook = dingtalk_webhook
self.results = []
def _load_hosts(self, hosts_file: str) -> list:
"""加载主机列表"""
with open(hosts_file, 'r') as f:
return [line.strip() for line in f if line.strip() and not line.startswith('#')]
def check_ping(self, host: str, timeout: int = 2) -> bool:
"""检查 Ping 连通性"""
try:
result = subprocess.run(
['ping', '-c', '1', '-W', str(timeout), host],
capture_output=True,
timeout=timeout + 1
)
return result.returncode == 0
except:
return False
def check_ssh(self, host: str, username: str = 'root', timeout: int = 5) -> dict:
"""检查 SSH 连接并获取系统信息"""
try:
client = paramiko.SSHClient()
client.set_missing_host_key_policy(paramiko.AutoAddPolicy())
client.connect(host, username=username, timeout=timeout)
# 获取 CPU 使用率
stdin, stdout, stderr = client.exec_command("top -bn1 | grep 'Cpu(s)' | awk '{print $2}'")
cpu_usage = float(stdout.read().decode().strip())
# 获取内存使用率
stdin, stdout, stderr = client.exec_command("free | grep Mem | awk '{printf \"%.1f\", $3/$2 * 100}'")
mem_usage = float(stdout.read().decode().strip())
# 获取磁盘使用率
stdin, stdout, stderr = client.exec_command("df / | tail -1 | awk '{print $5}' | sed 's/%//'")
disk_usage = float(stdout.read().decode().strip())
client.close()
return {
'success': True,
'cpu_usage': cpu_usage,
'mem_usage': mem_usage,
'disk_usage': disk_usage
}
except Exception as e:
return {'success': False, 'error': str(e)}
def check_host(self, host: str) -> dict:
"""检查单台主机"""
result = {
'host': host,
'timestamp': datetime.now().isoformat(),
'ping_ok': False,
'ssh_ok': False,
'cpu_usage': None,
'mem_usage': None,
'disk_usage': None,
'alerts': []
}
# Ping 检查
result['ping_ok'] = self.check_ping(host)
if not result['ping_ok']:
result['alerts'].append('Ping 失败')
return result
# SSH 检查
ssh_result = self.check_ssh(host)
if ssh_result['success']:
result['ssh_ok'] = True
result['cpu_usage'] = ssh_result['cpu_usage']
result['mem_usage'] = ssh_result['mem_usage']
result['disk_usage'] = ssh_result['disk_usage']
# 检查阈值
if ssh_result['cpu_usage'] > 80:
result['alerts'].append(f"CPU 使用率过高: {ssh_result['cpu_usage']}%")
if ssh_result['mem_usage'] > 85:
result['alerts'].append(f"内存使用率过高: {ssh_result['mem_usage']}%")
if ssh_result['disk_usage'] > 90:
result['alerts'].append(f"磁盘使用率过高: {ssh_result['disk_usage']}%")
else:
result['alerts'].append(f"SSH 连接失败: {ssh_result['error']}")
return result
def send_dingtalk_alert(self, alerts: list):
"""发送钉钉告警"""
if not alerts:
return
message = {
"msgtype": "text",
"text": {
"content": f"🚨 服务器告警\n\n" + "\n".join(alerts)
}
}
try:
requests.post(
self.dingtalk_webhook,
json=message,
headers={'Content-Type': 'application/json'}
)
except Exception as e:
print(f"钉钉告警发送失败: {e}")
def generate_html_report(self, output_file: str = 'health_report.html'):
"""生成 HTML 报告"""
html_template = """
<!DOCTYPE html>
<html>
<head>
<meta charset="UTF-8">
<title>服务器健康检查报告</title>
<style>
body {{ font-family: Arial, sans-serif; margin: 20px; }}
table {{ border-collapse: collapse; width: 100%; }}
th, td {{ border: 1px solid #ddd; padding: 8px; text-align: left; }}
th {{ background-color: #4CAF50; color: white; }}
tr:nth-child(even) {{ background-color: #f2f2f2; }}
.alert {{ color: red; font-weight: bold; }}
.ok {{ color: green; }}
</style>
</head>
<body>
<h1>服务器健康检查报告</h1>
<p>生成时间: {timestamp}</p>
<table>
<tr>
<th>主机</th>
<th>Ping</th>
<th>SSH</th>
<th>CPU</th>
<th>内存</th>
<th>磁盘</th>
<th>告警</th>
</tr>
{rows}
</table>
</body>
</html>
"""
rows = ""
for r in self.results:
status_class = "ok" if not r['alerts'] else "alert"
alerts_text = "<br>".join(r['alerts']) if r['alerts'] else "正常"
rows += f"""
<tr>
<td>{r['host']}</td>
<td class="{'ok' if r['ping_ok'] else 'alert'}">{'✅' if r['ping_ok'] else '❌'}</td>
<td class="{'ok' if r['ssh_ok'] else 'alert'}">{'✅' if r['ssh_ok'] else '❌'}</td>
<td>{r['cpu_usage'] or 'N/A'}%</td>
<td>{r['mem_usage'] or 'N/A'}%</td>
<td>{r['disk_usage'] or 'N/A'}%</td>
<td class="{status_class}">{alerts_text}</td>
</tr>
"""
html_content = html_template.format(
timestamp=datetime.now().strftime('%Y-%m-%d %H:%M:%S'),
rows=rows
)
with open(output_file, 'w', encoding='utf-8') as f:
f.write(html_content)
print(f"HTML 报告已生成: {output_file}")
def run(self, max_workers: int = 10):
"""执行健康检查"""
print(f"开始检查 {len(self.hosts)} 台服务器...")
with ThreadPoolExecutor(max_workers=max_workers) as executor:
future_to_host = {
executor.submit(self.check_host, host): host
for host in self.hosts
}
for future in as_completed(future_to_host):
result = future.result()
self.results.append(result)
# 实时输出
status = "✅" if not result['alerts'] else "❌"
print(f"{status} {result['host']}: {', '.join(result['alerts']) if result['alerts'] else '正常'}")
# 生成报告
self.generate_html_report()
# 发送告警
all_alerts = []
for r in self.results:
if r['alerts']:
all_alerts.append(f"主机: {r['host']}\n" + "\n".join(r['alerts']))
if all_alerts:
self.send_dingtalk_alert(all_alerts)
print(f"\n检查完成！正常: {sum(1 for r in self.results if not r['alerts'])} 台, "
f"异常: {sum(1 for r in self.results if r['alerts'])} 台")
def main():
"""主函数"""
checker = ServerHealthChecker(
hosts_file='hosts.txt',
dingtalk_webhook='https://oapi.dingtalk.com/robot/send?access_token=YOUR_TOKEN'
)
checker.run()
if __name__ == '__main__':
main()
场景二：MySQL 数据库自动备份
Prompt
你是一位数据库运维专家。请编写一个 Python 脚本，实现 MySQL 数据库的自动化备份。
要求：
1. 支持全量备份和增量备份
2. 备份文件压缩并加密（AES-256）
3. 上传到阿里云 OSS
4. 保留最近 7 天的备份
5. 备份完成后验证完整性
6. 发送备份报告邮件
7. 支持断点续传
8. 详细的日志记录
技术栈：Python 3.8+, mysqldump, oss2, cryptography
配置文件：config.yaml
场景三：Docker 容器批量管理
Prompt
你是一位 DevOps 工程师。请编写一个 Shell 脚本，用于批量管理 Docker 容器。
要求：
1. 列出所有运行中的容器及其资源使用情况
2. 清理已停止的容器
3. 清理未使用的镜像
4. 清理未使用的卷
5. 清理构建缓存
6. 生成清理报告
7. 支持 --dry-run 预览模式
8. 支持 --force 强制清理（跳过确认）
命令：docker system df, docker container prune, docker image prune
场景四：Kubernetes 集群巡检
Prompt
你是一位 K8s 运维专家。请编写一个 Python 脚本，用于 Kubernetes 集群日常巡检。
要求：
1. 检查所有 Pod 状态（Running/CrashLoopBackOff/Pending）
2. 检查节点资源使用率（CPU/内存）
3. 检查 PersistentVolume 使用情况
4. 检查证书过期时间
5. 检查 Etcd 健康状态
6. 生成 JSON 格式的巡检报告
7. 异常项发送企业微信告警
8. 支持指定 namespace 检查
技术栈：Python 3.8+, kubernetes-client, requests
场景五：安全基线检查
Prompt
你是一位安全运维专家。请编写一个 Shell 脚本，用于 Linux 系统安全基线检查。
要求：
1. 检查密码策略（最小长度、复杂度）
2. 检查 SSH 配置（禁止 root 登录、密钥认证）
3. 检查防火墙状态
4. 检查不必要的服务
5. 检查文件权限（/etc/passwd, /etc/shadow）
6. 检查 sudo 配置
7. 检查内核参数
8. 生成 HTML 格式的安全报告
9. 不符合项标红显示
参考标准：等保 2.0 三级要求
⚠️ 十大提效技巧与最佳实践
技巧 1：分步生成，逐步优化
❌ 错误做法：
"写一个完整的运维平台"
✅ 正确做法：
Step 1: "写一个函数，检查服务器 Ping 连通性"
Step 2: "添加 SSH 连接检查功能"
Step 3: "添加资源使用率采集"
Step 4: "添加 HTML 报告生成"
Step 5: "添加钉钉告警"
技巧 2：提供上下文信息
❌ 模糊描述：
"写一个备份脚本"
✅ 详细描述：
"写一个 MySQL 备份脚本，数据库版本 8.0，数据量 500GB，
要求每天凌晨 2 点全量备份，保留 7 天，存储到阿里云 OSS，
备份文件需要压缩和加密"
技巧 3：指定代码规范
请在生成的代码中：
1. 遵循 PEP 8 规范（Python）或 Google Shell Style Guide
2. 使用 type hints（Python 3.8+）
3. 每个函数必须有 docstring
4. 关键逻辑必须有中文注释
5. 包含完整的错误处理（try-except）
6. 使用 logging 模块而非 print
技巧 4：要求测试用例
请同时提供：
1. 单元测试代码（pytest）
2. 测试数据示例
3. 执行命令
4. 预期输出
技巧 5：迭代优化
第一轮：生成基础代码
第二轮："添加更详细的错误处理"
第三轮："优化性能，使用并发"
第四轮："添加监控和告警"
第五轮："生成使用文档"
技巧 6：代码审查清单
AI 生成代码后，必须人工审查：
- 是否有硬编码密码/密钥？
- 是否有 SQL 注入风险？
- 是否有命令注入风险？
- 错误处理是否完善？
- 资源是否正确释放（文件句柄、连接）？
- 是否符合团队代码规范？
- 是否有充分的注释？
- 是否经过测试验证？
技巧 7：建立 Prompt 库
# prompts/
├── shell/
│ ├── log_cleanup.md
│ ├── backup_mysql.md
│ └── health_check.md
├── python/
│ ├── api_client.md
│ ├── data_processing.md
│ └── monitoring.md
└── kubernetes/
├── pod_management.md
└── cluster_inspection.md
技巧 8：使用 Few-Shot Learning
示例 1：
输入：检查磁盘空间
输出：df -h | grep -E '^/dev'
示例 2：
输入：检查内存使用
输出：free -h
现在请生成：
输入：检查 CPU 负载
输出：[AI 会根据模式生成]
技巧 9：结合 RAG 技术
基于以下内部文档生成代码：
【公司规范】
1. 所有脚本必须放在 /opt/scripts/ 目录
2. 日志必须写入 /var/log/company/
3. 告警必须发送到钉钉机器人
4. 密码必须从 Vault 读取
【任务】
编写一个 Nginx 日志清理脚本
技巧 10：持续学习与反馈
每次使用后记录：
1. 哪些 Prompt 效果好？
2. 哪些代码需要大量修改？
3. 常见的错误模式是什么？
4. 如何优化 Prompt？
建立知识库，持续改进。
📊 效率对比与 ROI 分析
开发效率对比
| 任务类型 | 传统方式 | AI 辅助 | 提升倍数 |
|---|---|---|---|
| 简单脚本（<50 行） | 30 分钟 | 2 分钟 | 15x |
| 中等脚本（50-200 行） | 2 小时 | 10 分钟 | 12x |
| 复杂脚本（>200 行） | 8 小时 | 30 分钟 | 16x |
| 调试时间 | 1-2 小时 | 10 分钟 | 8x |
| 文档编写 | 30 分钟 | 2 分钟 | 15x |
成本核算
按 20 人运维团队计算:
| 指标 | 传统方式 | AI 辅助 | 年度节省 |
|---|---|---|---|
| 脚本开发工时 | 4000 小时/年 | 300 小时/年 | 3700 小时 |
| 人力成本（500 元/小时） | 200 万 | 15 万 | 185 万 |
| Bug 修复成本 | 50 万 | 10 万 | 40 万 |
| 培训成本 | 20 万 | 5 万 | 15 万 |
| 总计 | 270 万 | 30 万 | 240 万 |
结论: 引入 AI 代码助手，每年可为团队节省 240 万元！
🎁 福利资源包
资源 1: Prompt 模板库
# Shell 脚本 Prompt 模板
## 角色
你是一位拥有 10 年经验的 Linux 运维专家。
## 任务
请编写一个 Shell 脚本，用于 [具体功能]。
## 要求
1. 遵循 Google Shell Style Guide
2. 使用 set -euo pipefail
3. 包含完善的错误处理
4. 添加详细的中文注释
5. 支持 --help 参数
6. 提供使用示例
## 背景
- 操作系统：[CentOS 7/Ubuntu 20.04]
- 执行频率：[每天/每周]
- 目标用户：运维工程师
📝 总结与展望
核心价值
- 效率革命: 脚本开发速度提升 10-20 倍
- 质量保障: 标准化代码，降低 Bug 率
- 知识沉淀: Prompt 库成为团队资产
- 门槛降低: 新手也能写出专业脚本
- 成本节约: 年度节省数百万元
未来趋势
行动建议
- 立即行动: 安装 AI 代码助手（Copilot/Cursor/通义灵码）
- 本周完成: 建立团队 Prompt 模板库
- 本月目标: 将 50% 的脚本开发转为 AI 辅助
- 长期规划: 构建 AI 运维代理，实现完全自动化
👍 如果本文对你有帮助，欢迎点赞、收藏、转发！
💬 你在使用 AI 代码助手时有哪些心得？欢迎在评论区分享交流~
🔔 关注我，获取更多 DevOps 与 AI 结合的实战干货！
✍️ 行文仓促，定有不足之处，欢迎各位朋友在评论区批评指正，不胜感激!
专栏导航:
- 📖 下一篇: AI 辅助运维：从脚本生成到智能决策(即将发布)
- 📚 专栏首页: 自动化运维实战专栏
AtomGit 是由开放原子开源基金会联合 CSDN 等生态伙伴共同推出的新一代开源与人工智能协作平台。平台坚持“开放、中立、公益”的理念，把代码托管、模型共享、数据集托管、智能体开发体验和算力服务整合在一起，为开发者提供从开发、训练到部署的一站式体验。
更多推荐
12
0- 0
已为社区贡献19条内容
所有评论(0)