Walter's Blog

主题

Python 自动化 ​

概述 ​

Python 是做自动化的首选语言——语法简洁、标准库强大、第三方生态丰富。从简单的文件批处理到复杂的集群运维，Python 都能胜任。

自动化的核心思路 ​

不管做什么自动化，套路都一样：

1. 读取输入（文件 / API / 命令输出 / 数据库）
2. 处理逻辑（过滤 / 转换 / 计算 / 判断）
3. 执行动作（写文件 / 调 API / 执行命令 / 发通知）
4. 记录结果（日志 / 报告 / 数据库）

场景一：文件与目录批量操作 ​

最基础的自动化，纯标准库就能搞定。

批量重命名文件 ​

python
from pathlib import Path
import re

# 把所有 .txt 文件加上日期前缀
for f in Path(".").glob("*.txt"):
    new_name = f"2024-01-15_{f.name}"
    f.rename(f.parent / new_name)
    print(f"重命名: {f.name} → {new_name}")

批量查找和替换文本 ​

python
from pathlib import Path

def replace_in_file(file_path: Path, old: str, new: str):
    content = file_path.read_text(encoding="utf-8")
    if old in content:
        content = content.replace(old, new)
        file_path.write_text(content, encoding="utf-8")
        print(f"已替换: {file_path}")

# 批量替换所有 YAML 文件中的镜像版本
for f in Path(".").rglob("*.yaml"):
    replace_in_file(f, "image: nginx:1.24", "image: nginx:1.25")

批量清理旧文件 ​

python
from pathlib import Path
from datetime import datetime, timedelta

# 清理 30 天前的日志文件
log_dir = Path("/var/log/myapp")
cutoff = datetime.now() - timedelta(days=30)

for f in log_dir.glob("*.log"):
    mtime = datetime.fromtimestamp(f.stat().st_mtime)
    if mtime < cutoff:
        f.unlink()
        print(f"已删除: {f.name} (最后修改: {mtime:%Y-%m-%d})")

场景二：调用系统命令 ​

用 subprocess 调用外部命令，适合已有 CLI 工具但需要批量/编排的场景。

封装命令执行函数 ​

python
import subprocess
import logging

logging.basicConfig(level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s")

def run_cmd(cmd: list[str], check: bool = True) -> subprocess.CompletedProcess:
    """执行系统命令，记录日志"""
    cmd_str = " ".join(cmd)
    logging.info(f"执行: {cmd_str}")

    result = subprocess.run(
        cmd,
        capture_output=True,
        text=True,
        check=check
    )

    if result.stdout:
        logging.info(result.stdout.strip())
    if result.stderr:
        logging.warning(result.stderr.strip())

    return result

批量操作 K8s 资源 ​

python
# 批量重启 Deployment
namespaces = ["default", "staging", "production"]
deployments = ["nginx", "redis", "api-server"]

for ns in namespaces:
    for deploy in deployments:
        try:
            run_cmd([
                "kubectl", "rollout", "restart",
                f"deployment/{deploy}", "-n", ns
            ])
        except subprocess.CalledProcessError as e:
            logging.error(f"重启失败: {ns}/{deploy} - {e.stderr}")

批量检查服务状态 ​

python
# 检查多个节点的连通性
nodes = ["node1", "node2", "node3"]

for node in nodes:
    result = subprocess.run(
        ["ssh", node, "uptime"],
        capture_output=True, text=True, timeout=10
    )
    if result.returncode == 0:
        print(f"✅ {node}: {result.stdout.strip()}")
    else:
        print(f"❌ {node}: 连接失败")

场景三：SSH 远程执行 ​

当需要远程操作多台服务器时，用 paramiko 比 subprocess + ssh 更灵活。

安装 ​

bash
uv add paramiko

单台服务器操作 ​

python
import paramiko

def ssh_exec(host: str, command: str, username: str = "root",
             key_file: str = "~/.ssh/id_rsa") -> str:
    """SSH 执行远程命令"""
    client = paramiko.SSHClient()
    client.set_missing_host_key_policy(paramiko.AutoAddPolicy())
    client.connect(host, username=username, key_filename=key_file)

    stdin, stdout, stderr = client.exec_command(command)
    output = stdout.read().decode()
    error = stderr.read().decode()
    client.close()

    if error:
        raise RuntimeError(f"命令执行失败: {error}")
    return output

# 使用
result = ssh_exec("192.168.1.10", "df -h")
print(result)

批量操作多台服务器 ​

python
import paramiko
from concurrent.futures import ThreadPoolExecutor, as_completed

def ssh_exec(host: str, command: str) -> dict:
    """在单台服务器执行命令，返回结果"""
    try:
        client = paramiko.SSHClient()
        client.set_missing_host_key_policy(paramiko.AutoAddPolicy())
        client.connect(host, username="root", key_filename="~/.ssh/id_rsa",
                       timeout=10)
        stdin, stdout, stderr = client.exec_command(command)
        output = stdout.read().decode().strip()
        client.close()
        return {"host": host, "status": "success", "output": output}
    except Exception as e:
        return {"host": host, "status": "failed", "error": str(e)}

# 并发执行
hosts = ["192.168.1.10", "192.168.1.11", "192.168.1.12"]
command = "uptime"

with ThreadPoolExecutor(max_workers=5) as pool:
    futures = {pool.submit(ssh_exec, h, command): h for h in hosts}
    for future in as_completed(futures):
        result = future.result()
        if result["status"] == "success":
            print(f"✅ {result['host']}: {result['output']}")
        else:
            print(f"❌ {result['host']}: {result['error']}")

场景四：调用 HTTP API ​

用 httpx（或 requests）调用 REST API，适合操作 K8s、云平台、内部服务等。

安装 ​

bash
uv add httpx

基本用法 ​

python
import httpx

# GET 请求
resp = httpx.get("https://httpbin.org/get", params={"key": "value"})
data = resp.json()

# POST 请求
resp = httpx.post("https://httpbin.org/post", json={"name": "test"})

# 带认证
resp = httpx.get(
    "https://api.example.com/data",
    headers={"Authorization": "Bearer sk-xxx"}
)

批量调用 API ​

python
import httpx
import logging

logging.basicConfig(level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s")

# 批量检查服务健康状态
services = [
    {"name": "用户服务", "url": "http://user-svc:8080/health"},
    {"name": "订单服务", "url": "http://order-svc:8080/health"},
    {"name": "支付服务", "url": "http://pay-svc:8080/health"},
]

with httpx.Client(timeout=5.0) as client:
    for svc in services:
        try:
            resp = client.get(svc["url"])
            if resp.status_code == 200:
                logging.info(f"✅ {svc['name']}: 正常")
            else:
                logging.warning(f"⚠️ {svc['name']}: 状态码 {resp.status_code}")
        except httpx.ConnectError:
            logging.error(f"❌ {svc['name']}: 无法连接")

场景五：定时任务 ​

让脚本按计划自动执行。

方式一：系统 crontab（推荐生产使用） ​

bash
# 编辑 crontab
crontab -e

# 每 5 分钟检查一次
*/5 * * * * /path/to/python /path/to/check_pods.py >> /var/log/check_pods.log 2>&1

# 每天凌晨 2 点清理
0 2 * * * /path/to/python /path/to/cleanup.py

方式二：Python 内置 schedule 库 ​

bash
uv add schedule
python
import schedule
import time
import logging

logging.basicConfig(level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s")

def check_pods():
    logging.info("开始检查 Pod 状态...")
    # 你的检查逻辑

def cleanup():
    logging.info("开始清理...")
    # 你的清理逻辑

# 定义计划
schedule.every(5).minutes.do(check_pods)
schedule.every().day.at("02:00").do(cleanup)
schedule.every().monday.do(cleanup)

# 运行
while True:
    schedule.run_pending()
    time.sleep(1)

方式三：APScheduler（更强大） ​

bash
uv add apscheduler
python
from apscheduler.schedulers.blocking import BlockingScheduler

scheduler = BlockingScheduler()

@scheduler.scheduled_job("interval", minutes=5)
def check_pods():
    print("检查 Pod 状态...")

@scheduler.scheduled_job("cron", hour=2, minute=0)
def cleanup():
    print("执行清理...")

scheduler.start()

场景六：配置文件管理 ​

自动化脚本通常需要配置（服务器列表、API 地址、认证信息等），不要硬编码。

YAML 配置文件 ​

yaml
# config.yaml
servers:
  - host: 192.168.1.10
    role: master
  - host: 192.168.1.11
    role: worker

k8s:
  namespace: production
  context: prod-cluster

notifications:
  webhook_url: https://hooks.example.com/alert
  enabled: true
python
# 读取配置
import yaml
from pathlib import Path

config = yaml.safe_load(Path("config.yaml").read_text(encoding="utf-8"))
servers = config["servers"]
namespace = config["k8s"]["namespace"]

环境变量（敏感信息） ​

python
import os

# 敏感信息用环境变量，不要写在配置文件里
api_key = os.getenv("API_KEY")
db_password = os.getenv("DB_PASSWORD")

.env 文件（开发环境） ​

bash
uv add python-dotenv
ini
# .env
API_KEY=sk-xxxxx
DB_PASSWORD=secret123
K8S_NAMESPACE=production
python
from dotenv import load_dotenv
import os

load_dotenv()  # 加载 .env 文件
api_key = os.getenv("API_KEY")

场景七：通知与告警 ​

自动化执行完需要通知结果。

飞书 / 钉钉 Webhook ​

python
import httpx

def send_feishu_alert(title: str, content: str, webhook_url: str):
    """发送飞书告警"""
    payload = {
        "msg_type": "interactive",
        "card": {
            "header": {"title": {"tag": "plain_text", "content": title}},
            "elements": [{"tag": "markdown", "content": content}]
        }
    }
    httpx.post(webhook_url, json=payload)

# 使用
send_feishu_alert(
    "⚠️ Pod 异常告警",
    "命名空间 production 中有 3 个 Pod 处于 CrashLoopBackOff 状态",
    os.getenv("FEISHU_WEBHOOK_URL")
)

邮件通知 ​

python
import smtplib
from email.mime.text import MIMEText

def send_email(subject: str, body: str, to: str):
    msg = MIMEText(body, "plain", "utf-8")
    msg["Subject"] = subject
    msg["From"] = "monitor@example.com"
    msg["To"] = to

    with smtplib.SMTP("smtp.example.com", 587) as server:
        server.starttls()
        server.login("monitor@example.com", os.getenv("SMTP_PASSWORD"))
        server.send_message(msg)

完整示例：K8s Pod 监控脚本 ​

把上面的技巧组合起来，写一个完整的监控脚本：

python
#!/usr/bin/env python3
"""K8s Pod 状态监控脚本 - 检查异常 Pod 并发送告警"""

import os
import logging
import httpx
from kubernetes import client, config
from dotenv import load_dotenv

load_dotenv()

logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s [%(levelname)s] %(message)s"
)

# ========== 配置 ==========
NAMESPACE = os.getenv("K8S_NAMESPACE", "default")
WEBHOOK_URL = os.getenv("FEISHU_WEBHOOK_URL", "")
ALERT_PHASES = {"Failed", "Pending", "Unknown"}
ALERT_RESTART_THRESHOLD = 5

# ========== 核心逻辑 ==========
def check_pods() -> list[dict]:
    """检查异常 Pod，返回告警列表"""
    config.load_kube_config()
    v1 = client.CoreV1Api()
    pods = v1.list_namespaced_pod(namespace=NAMESPACE)

    alerts = []
    for pod in pods.items:
        # 检查异常状态
        if pod.status.phase in ALERT_PHASES:
            alerts.append({
                "name": pod.metadata.name,
                "namespace": NAMESPACE,
                "reason": f"状态异常: {pod.status.phase}"
            })

        # 检查重启次数
        for container in pod.status.container_statuses or []:
            if container.restart_count > ALERT_RESTART_THRESHOLD:
                alerts.append({
                    "name": pod.metadata.name,
                    "namespace": NAMESPACE,
                    "reason": f"容器 {container.name} 重启 {container.restart_count} 次"
                })

    return alerts

# ========== 通知 ==========
def send_alert(alerts: list[dict]):
    """发送飞书告警"""
    if not WEBHOOK_URL:
        for a in alerts:
            logging.warning(f"⚠️ {a['namespace']}/{a['name']}: {a['reason']}")
        return

    content = "\n".join(
        f"- **{a['namespace']}/{a['name']}**: {a['reason']}"
        for a in alerts
    )
    payload = {
        "msg_type": "interactive",
        "card": {
            "header": {"title": {"tag": "plain_text", "content": f"⚠️ K8s Pod 告警 ({len(alerts)}条)"}},
            "elements": [{"tag": "markdown", "content": content}]
        }
    }
    httpx.post(WEBHOOK_URL, json=payload)
    logging.info(f"已发送 {len(alerts)} 条告警")

# ========== 主流程 ==========
def main():
    logging.info(f"开始检查命名空间 {NAMESPACE} 的 Pod 状态...")
    alerts = check_pods()

    if alerts:
        logging.warning(f"发现 {len(alerts)} 个异常")
        send_alert(alerts)
    else:
        logging.info("所有 Pod 状态正常")

if __name__ == "__main__":
    main()

配合 crontab 定时执行：

bash
# 每 5 分钟检查一次
*/5 * * * * cd /opt/k8s-monitor && .venv/bin/python monitor.py >> /var/log/k8s-monitor.log 2>&1

自动化脚本的项目结构 ​

当脚本多了之后，建议按项目组织：

my-automation/
├── .env                    # 敏感配置（不提交 Git）
├── .env.example            # 配置模板（提交 Git）
├── config.yaml             # 非敏感配置
├── requirements.txt        # 依赖
├── scripts/
│   ├── check_pods.py       # Pod 监控
│   ├── cleanup.py          # 清理脚本
│   └── deploy.py           # 部署脚本
├── lib/
│   ├── k8s_helper.py       # K8s 操作封装
│   ├── notify.py           # 通知模块
│   └── config.py           # 配置加载
└── logs/                   # 日志目录

参考资料 ​

基于 MIT 许可发布

版权所有 © 2024-present Walter