在 Ubuntu 上使用 Python 做自动化运维的实操指南
一 环境准备与基础脚本
sudo apt update && sudo apt install -y python3 python3-pippython3 -m venv venv && source venv/bin/activatepip install paramiko scp psutil requests# check_service.py
import subprocess
def check_service(service: str) -> None:
res = subprocess.run(
["sudo", "systemctl", "is-active", service],
capture_output=True, text=True
)
if res.stdout.strip() != "active":
subprocess.run(["sudo", "systemctl", "restart", service])
if __name__ == "__main__":
check_service("nginx")
二 将脚本做成系统服务与定时任务
sudo nano /etc/systemd/system/my_service.service[Unit]
Description=My Python Service
After=network.target
[Service]
User=your_username
ExecStart=/path/to/venv/bin/python /path/to/your_script.py
Restart=always
RestartSec=10
Environment="PATH=/path/to/venv/bin:$PATH"
[Install]
WantedBy=multi-user.target
sudo systemctl daemon-reload && sudo systemctl enable --now my_servicesudo systemctl status my_servicecrontab -e0 1 * * * /path/to/venv/bin/python /path/to/your_script.pyExecStart 改为 Conda 环境解释器路径：
/home/your_username/miniconda3/envs/<env>/bin/python /path/to/script.py三 远程批量执行与自动化部署
pip install fabric# fabfile.py
from fabric import Connection
def whoami(c):
result = c.run("whoami", hide=True)
print("当前用户:", result.stdout.strip())
fab -H your_server_ip whoami# fabfile.py
from fabric import task, SerialGroup
@task
def update_system(c):
c.run("sudo apt update && sudo apt upgrade -y", hide=True)
c.run("sudo apt autoremove -y", hide=True)
print("系统更新完成")
@task
def deploy(c):
c.run("mkdir -p /var/www/my_site")
c.put("local/path/", "/var/www/my_site/")
c.run("chmod -R 755 /var/www/my_site")
fab -H s1,s2,s3 update_system 或 fab -H s1 deployconnect_kwargs={"key_filename": "/path/to/key"} 指定）connect_kwargs={"password": "..."}，但密钥更安全四 监控与告警实战示例
# monitor.py
import psutil, smtplib
from email.mime.text import MIMEText
from datetime import datetime
CPU_THRESHOLD = 80.0
MEM_THRESHOLD = 80.0
SMTP_HOST = "smtp.example.com"
SMTP_PORT = 587
SMTP_USER = "your_email@example.com"
SMTP_PASS = "your_password"
FROM = "your_email@example.com"
TO = "alert_recipient@example.com"
def send_alert(cpu: float, mem: float):
body = f"告警时间: {datetime.now()}\nCPU使用率: {cpu}%\n内存使用率: {mem}%"
msg = MIMEText(body, "plain", "utf-8")
msg["Subject"] = "服务器性能告警"
msg["From"] = FROM
msg["To"] = TO
with smtplib.SMTP(SMTP_HOST, SMTP_PORT) as s:
s.starttls()
s.login(SMTP_USER, SMTP_PASS)
s.send_message(msg)
def main():
cpu = psutil.cpu_percent(interval=1)
mem = psutil.virtual_memory().percent
if cpu > CPU_THRESHOLD or mem > MEM_THRESHOLD:
send_alert(cpu, mem)
if __name__ == "__main__":
main()
五 规模化与进阶方案
sudo apt install salt-master（Master）、sudo apt install salt-minion（Minion）echo "master: <MASTER_IP>" | sudo tee /etc/salt/minionsudo systemctl enable --now salt-minionsalt '*' test.pingsalt '*' cmd.run 'df -h'salt '*' pkg.install nginxsalt '*' cp.get_file salt://path/to/file /dest/pathsalt '*' service.start nginx、salt '*' service.enable nginx