Python运维工程师必会的指令

掌握这些Python运维指令，让你的运维工作自动化、高效化

欢迎大家关注此公众号，后台点击按钮【免费资料】可免费获取【Python入门30节课】电子书

1. 此外小庄推荐一本适合于新手\小白入手一本 Python基础书籍，欢迎大家订阅，也感谢大家支持，我才有更新的动力

前言

Python运维工程师需要掌握一系列用于系统管理、自动化部署、监控告警的指令和工具。本文将系统性地介绍运维工程师必须掌握的Python指令，帮助你构建强大的运维自动化体系。

一、系统信息获取

1.1 获取系统信息

import platform

print(platform.system())       # 操作系统: Windows/Linux/Darwin
print(platform.release())      # 系统版本
print(platform.version())      # 系统详细版本
print(platform.machine())      # 机器架构: x86_64/AMD64
print(platform.processor())    # 处理器信息
print(platform.node())         # 主机名

1.2 获取CPU信息

import psutil

# CPU使用率
cpu_percent = psutil.cpu_percent(interval=1)
print(f'CPU使用率: {cpu_percent}%')

# 每个CPU核心的使用率
cpu_per_core = psutil.cpu_percent(interval=1, percpu=True)
for i, percent in enumerate(cpu_per_core):
print(f'CPU核心{i}: {percent}%')

# CPU数量
print(f'逻辑CPU: {psutil.cpu_count()}')
print(f'物理CPU: {psutil.cpu_count(logical=False)}')

# CPU频率
cpu_freq = psutil.cpu_freq()
print(f'当前频率: {cpu_freq.current}MHz')

1.3 获取内存信息

import psutil

memory = psutil.virtual_memory()
print(f'总内存: {memory.total / 1024 / 1024 / 1024:.2f}GB')
print(f'已使用: {memory.used / 1024 / 1024 / 1024:.2f}GB')
print(f'可用内存: {memory.available / 1024 / 1024 / 1024:.2f}GB')
print(f'使用率: {memory.percent}%')

# 交换内存
swap = psutil.swap_memory()
print(f'交换内存使用率: {swap.percent}%')

1.4 获取磁盘信息

import psutil

# 磁盘分区
partitions = psutil.disk_partitions()
for partition in partitions:
print(f'设备: {partition.device}, 挂载点: {partition.mountpoint}')

    usage = psutil.disk_usage(partition.mountpoint)
print(f'  总空间: {usage.total / 1024 / 1024 / 1024:.2f}GB')
print(f'  已使用: {usage.used / 1024 / 1024 / 1024:.2f}GB')
print(f'  使用率: {usage.percent}%')

# 磁盘IO
disk_io = psutil.disk_io_counters()
print(f'读取字节数: {disk_io.read_bytes}')
print(f'写入字节数: {disk_io.write_bytes}')

1.5 获取网络信息

import psutil

# 网络接口
net_interfaces = psutil.net_if_addrs()
for name, addrs in net_interfaces.items():
print(f'接口: {name}')
for addr in addrs:
if addr.family.name == 'AF_INET':
print(f'  IPv4: {addr.address}')

# 网络IO
net_io = psutil.net_io_counters()
print(f'发送字节数: {net_io.bytes_sent}')
print(f'接收字节数: {net_io.bytes_recv}')

# 网络连接
connections = psutil.net_connections()
for conn in connections[:5]:
print(f'本地地址: {conn.laddr}, 远程地址: {conn.raddr}, 状态: {conn.status}')

二、进程管理

2.1 查看所有进程

import psutil

for proc in psutil.process_iter(['pid', 'name', 'cpu_percent', 'memory_percent']):
print(f"PID: {proc.info['pid']}, 名称: {proc.info['name']}, "
f"CPU: {proc.info['cpu_percent']}%, 内存: {proc.info['memory_percent']:.2f}%")

2.2 查找特定进程

import psutil

def find_process_by_name(name):
"""根据进程名查找进程"""
    processes = []
for proc in psutil.process_iter(['pid', 'name', 'cmdline']):
if name.lower() in proc.info['name'].lower():
            processes.append(proc)
return processes

# 查找Python进程
python_procs = find_process_by_name('python')
for proc in python_procs:
print(f"PID: {proc.pid}, 命令: {proc.cmdline()}")

2.3 进程操作

import psutil
import os

# 获取当前进程
current_proc = psutil.Process(os.getpid())
print(f'当前进程PID: {current_proc.pid}')
print(f'进程名称: {current_proc.name()}')
print(f'进程状态: {current_proc.status()}')

# 终止进程
proc = psutil.Process(1234)
proc.terminate()  # 发送SIGTERM
# proc.kill()     # 发送SIGKILL

# 等待进程结束
proc.wait(timeout=10)

2.4 获取进程详细信息

import psutil

proc = psutil.Process(1234)

# 基本信息
print(f'PID: {proc.pid}')
print(f'名称: {proc.name()}')
print(f'状态: {proc.status()}')
print(f'创建时间: {proc.create_time()}')

# 资源使用
print(f'CPU使用率: {proc.cpu_percent(interval=1)}%')
print(f'内存使用: {proc.memory_info().rss / 1024 / 1024:.2f}MB')
print(f'内存使用率: {proc.memory_percent():.2f}%')

# 命令行
print(f'命令行: {proc.cmdline()}')

# 工作目录
print(f'工作目录: {proc.cwd()}')

# 打开的文件
for file in proc.open_files():
print(f'打开文件: {file.path}')

三、文件系统操作

3.1 文件和目录操作

import os
import shutil
import glob

# 获取当前目录
current_dir = os.getcwd()
print(f'当前目录: {current_dir}')

# 列出目录内容
files = os.listdir('.')
for file in files:
print(file)

# 判断路径
print(os.path.exists('/path/to/file'))
print(os.path.isfile('/path/to/file'))
print(os.path.isdir('/path/to/dir'))

# 获取文件信息
file_path = 'test.txt'
print(f'文件大小: {os.path.getsize(file_path)}字节')
print(f'修改时间: {os.path.getmtime(file_path)}')

# 创建目录
os.makedirs('/path/to/dir', exist_ok=True)

# 删除目录
shutil.rmtree('/path/to/dir')

# 复制文件
shutil.copy('source.txt', 'destination.txt')
shutil.copytree('source_dir', 'destination_dir')

# 移动/重命名
shutil.move('old_name.txt', 'new_name.txt')

# 删除文件
os.remove('file.txt')

# 查找文件
files = glob.glob('*.py')
files = glob.glob('**/*.py', recursive=True)

3.2 文件内容操作

# 读取文件
with open('file.txt', 'r', encoding='utf-8') as f:
    content = f.read()

# 逐行读取
with open('file.txt', 'r', encoding='utf-8') as f:
for line in f:
print(line.strip())

# 写入文件
with open('output.txt', 'w', encoding='utf-8') as f:
    f.write('Hello World\n')

# 追加写入
with open('log.txt', 'a', encoding='utf-8') as f:
    f.write('New log entry\n')

3.3 文件监控

import os
import time

def watch_file(file_path, interval=1):
"""监控文件变化"""
    last_mtime = os.path.getmtime(file_path)

while True:
        current_mtime = os.path.getmtime(file_path)
if current_mtime != last_mtime:
print(f'文件 {file_path} 已更新')
            last_mtime = current_mtime
        time.sleep(interval)

# 使用watchdog库进行更强大的文件监控
# pip install watchdog
from watchdog.observers import Observer
from watchdog.events import FileSystemEventHandler

class MyHandler(FileSystemEventHandler):
def on_modified(self, event):
print(f'文件已修改: {event.src_path}')

def on_created(self, event):
print(f'文件已创建: {event.src_path}')

def on_deleted(self, event):
print(f'文件已删除: {event.src_path}')

observer = Observer()
observer.schedule(MyHandler(), path='.', recursive=True)
observer.start()

四、Shell命令执行

4.1 使用subprocess执行命令

import subprocess

# 执行简单命令
result = subprocess.run(['ls', '-la'], capture_output=True, text=True)
print(result.stdout)
print(result.stderr)
print(result.returncode)

# 执行shell命令
result = subprocess.run('ls -la | grep .py', shell=True, capture_output=True, text=True)

# 超时控制
try:
    result = subprocess.run(['ping', '-c', '4', 'google.com'], 
                          capture_output=True, text=True, timeout=10)
except subprocess.TimeoutExpired:
print('命令执行超时')

4.2 使用subprocess.Popen

import subprocess

# 启动进程
process = subprocess.Popen(['ping', '-c', '4', 'google.com'],
                          stdout=subprocess.PIPE,
                          stderr=subprocess.PIPE,
                          text=True)

# 获取输出
stdout, stderr = process.communicate()
print(f'输出: {stdout}')
print(f'错误: {stderr}')
print(f'返回码: {process.returncode}')

4.3 使用os.system

import os

# 简单执行命令
os.system('ls -la')

# 获取命令输出（使用popen）
output = os.popen('ls -la').read()
print(output)

4.4 使用paramiko进行SSH

import paramiko

# 创建SSH客户端
ssh = paramiko.SSHClient()
ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())

# 连接服务器
ssh.connect('hostname', username='user', password='password')

# 执行命令
stdin, stdout, stderr = ssh.exec_command('ls -la')
print(stdout.read().decode())

# 使用密钥连接
private_key = paramiko.RSAKey.from_private_key_file('/path/to/key')
ssh.connect('hostname', username='user', pkey=private_key)

# 关闭连接
ssh.close()

五、日志管理

5.1 使用logging模块

import logging

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

logger = logging.getLogger(__name__)

# 记录日志
logger.info('这是一条信息日志')
logger.warning('这是一条警告日志')
logger.error('这是一条错误日志')
logger.debug('这是一条调试日志')

5.2 日志轮转

import logging
from logging.handlers import RotatingFileHandler, TimedRotatingFileHandler

# 按大小轮转
handler = RotatingFileHandler(
'app.log', 
    maxBytes=10*1024*1024,  # 10MB
    backupCount=5
)

# 按时间轮转
handler = TimedRotatingFileHandler(
'app.log',
    when='midnight',
    interval=1,
    backupCount=30
)

5.3 解析日志文件

import re
from collections import Counter

def parse_nginx_log(log_file):
"""解析Nginx日志"""
    ip_pattern = r'\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3}'

    ip_list = []
with open(log_file, 'r') as f:
for line in f:
match = re.search(ip_pattern, line)
if match:
                ip_list.append(match.group())

# 统计IP访问次数
    ip_counter = Counter(ip_list)
return ip_counter.most_common(10)

六、定时任务

6.1 使用schedule库

import schedule
import time

def job():
print("执行定时任务...")

# 每隔10分钟执行
schedule.every(10).minutes.do(job)

# 每小时执行
schedule.every().hour.do(job)

# 每天10:30执行
schedule.every().day.at("10:30").do(job)

# 每周一执行
schedule.every().monday.do(job)

# 运行定时任务
while True:
    schedule.run_pending()
    time.sleep(1)

6.2 使用APScheduler

from apscheduler.schedulers.blocking import BlockingScheduler
from apscheduler.triggers.cron import CronTrigger
from apscheduler.triggers.interval import IntervalTrigger

scheduler = BlockingScheduler()

# 间隔执行
@scheduler.scheduled_job(IntervalTrigger(seconds=30))
def interval_job():
print("每30秒执行一次")

# Cron表达式
@scheduler.scheduled_job(CronTrigger(hour=8, minute=30))
def cron_job():
print("每天8:30执行")

scheduler.start()

6.3 使用crontab

# 生成crontab配置
def generate_crontab(command, schedule):
"""生成crontab配置"""
return f"{schedule} {command}"

# 示例：每天凌晨2点执行备份
crontab = generate_crontab(
    command="/usr/bin/python3 /path/to/backup.py",
    schedule="0 2 * * *"
)
print(crontab)

七、网络监控

7.1 Ping检测

import subprocess
import platform

def ping(host, count=4):
"""Ping主机"""
    param = '-n' if platform.system().lower() == 'windows' else '-c'
    command = ['ping', param, str(count), host]

try:
        result = subprocess.run(command, capture_output=True, text=True, timeout=10)
return result.returncode == 0
except subprocess.TimeoutExpired:
return False

# 批量检测
hosts = ['google.com', 'github.com', 'example.com']
for host in hosts:
    status = '在线' if ping(host) else '离线'
print(f'{host}: {status}')

7.2 端口检测

import socket

def check_port(host, port, timeout=2):
"""检测端口是否开放"""
try:
        sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        sock.settimeout(timeout)
        result = sock.connect_ex((host, port))
        sock.close()
return result == 0
except socket.error:
return False

# 检测常用端口
ports = [22, 80, 443, 3306, 6379]
for port in ports:
    status = '开放' if check_port('localhost', port) else '关闭'
print(f'端口 {port}: {status}')

7.3 HTTP健康检查

import requests
from datetime import datetime

def health_check(url, timeout=5):
"""HTTP健康检查"""
try:
        start_time = datetime.now()
        response = requests.get(url, timeout=timeout)
        end_time = datetime.now()

        response_time = (end_time - start_time).total_seconds()

return {
'url': url,
'status': response.status_code,
'response_time': response_time,
'healthy': response.status_code == 200
        }
except requests.RequestException as e:
return {
'url': url,
'status': None,
'response_time': None,
'healthy': False,
'error': str(e)
        }

# 批量健康检查
urls = [
'https://www.google.com',
'https://www.github.com',
'https://api.example.com'
]

for url in urls:
    result = health_check(url)
    status = '健康' if result['healthy'] else '异常'
print(f"{url}: {status}")

八、配置管理

8.1 使用YAML配置

import yaml

# 读取YAML配置
with open('config.yaml', 'r', encoding='utf-8') as f:
    config = yaml.safe_load(f)

print(config['database']['host'])
print(config['database']['port'])

# 写入YAML配置
config = {
'database': {
'host': 'localhost',
'port': 3306,
'name': 'mydb'
    },
'server': {
'host': '0.0.0.0',
'port': 8080
    }
}

with open('config.yaml', 'w', encoding='utf-8') as f:
    yaml.dump(config, f, default_flow_style=False)

8.2 使用JSON配置

import json

# 读取JSON配置
with open('config.json', 'r', encoding='utf-8') as f:
    config = json.load(f)

# 写入JSON配置
with open('config.json', 'w', encoding='utf-8') as f:
    json.dump(config, f, indent=2, ensure_ascii=False)

8.3 使用环境变量

import os

# 读取环境变量
db_host = os.environ.get('DB_HOST', 'localhost')
db_port = int(os.environ.get('DB_PORT', 3306))

# 设置环境变量
os.environ['MY_VAR'] = 'my_value'

8.4 使用python-dotenv

from dotenv import load_dotenv
import os

# 加载.env文件
load_dotenv()

# 读取环境变量
secret_key = os.getenv('SECRET_KEY')
database_url = os.getenv('DATABASE_URL')

九、备份与恢复

9.1 文件备份

import shutil
import os
from datetime import datetime

def backup_file(source, backup_dir):
"""备份文件"""
if not os.path.exists(backup_dir):
        os.makedirs(backup_dir)

    timestamp = datetime.now().strftime('%Y%m%d_%H%M%S')
    filename = os.path.basename(source)
    backup_path = os.path.join(backup_dir, f'{filename}.{timestamp}.bak')

    shutil.copy2(source, backup_path)
return backup_path

# 使用示例
backup_path = backup_file('important.db', './backups')
print(f'备份已创建: {backup_path}')

9.2 数据库备份

import subprocess
from datetime import datetime

def backup_mysql(host, user, password, database, backup_dir):
"""MySQL数据库备份"""
    timestamp = datetime.now().strftime('%Y%m%d_%H%M%S')
    backup_file = f'{backup_dir}/{database}_{timestamp}.sql'

    command = [
'mysqldump',
f'--host={host}',
f'--user={user}',
f'--password={password}',
        database
    ]

with open(backup_file, 'w') as f:
        subprocess.run(command, stdout=f, check=True)

return backup_file

# 使用示例
backup_file = backup_mysql('localhost', 'root', 'password', 'mydb', './backups')
print(f'数据库备份完成: {backup_file}')

9.3 自动清理旧备份

import os
import glob
from datetime import datetime, timedelta

def cleanup_old_backups(backup_dir, days=7):
"""清理超过指定天数的备份文件"""
    cutoff_date = datetime.now() - timedelta(days=days)

for file_path in glob.glob(os.path.join(backup_dir, '*.bak')):
        file_stat = os.stat(file_path)
        file_mtime = datetime.fromtimestamp(file_stat.st_mtime)

if file_mtime < cutoff_date:
            os.remove(file_path)
print(f'已删除旧备份: {file_path}')

# 使用示例
cleanup_old_backups('./backups', days=7)

十、监控告警

10.1 系统监控脚本

import psutil
import smtplib
from email.mime.text import MIMEText
from datetime import datetime

def system_monitor():
"""系统监控"""
    alerts = []

# CPU监控
    cpu_percent = psutil.cpu_percent(interval=1)
if cpu_percent > 90:
        alerts.append(f'CPU使用率过高: {cpu_percent}%')

# 内存监控
    memory = psutil.virtual_memory()
if memory.percent > 90:
        alerts.append(f'内存使用率过高: {memory.percent}%')

# 磁盘监控
    disk = psutil.disk_usage('/')
if disk.percent > 90:
        alerts.append(f'磁盘使用率过高: {disk.percent}%')

return alerts

def send_alert_email(alerts, to_email):
"""发送告警邮件"""
if not alerts:
return

    subject = f'系统告警 - {datetime.now().strftime("%Y-%m-%d %H:%M:%S")}'
    body = '\n'.join(alerts)

    msg = MIMEText(body)
    msg['Subject'] = subject
    msg['From'] = 'monitor@example.com'
    msg['To'] = to_email

with smtplib.SMTP('smtp.example.com', 587) as server:
        server.starttls()
        server.login('user', 'password')
        server.send_message(msg)

# 执行监控
alerts = system_monitor()
if alerts:
    send_alert_email(alerts, 'admin@example.com')

10.2 钉钉/企业微信告警

import requests
import json

def send_dingtalk_alert(webhook, message):
"""发送钉钉告警"""
    headers = {'Content-Type': 'application/json'}
    data = {
'msgtype': 'text',
'text': {
'content': message
        }
    }
    response = requests.post(webhook, headers=headers, data=json.dumps(data))
return response.json()

def send_wechat_alert(webhook, message):
"""发送企业微信告警"""
    headers = {'Content-Type': 'application/json'}
    data = {
'msgtype': 'text',
'text': {
'content': message
        }
    }
    response = requests.post(webhook, headers=headers, data=json.dumps(data))
return response.json()

# 使用示例
dingtalk_webhook = 'https://oapi.dingtalk.com/robot/send?access_token=xxx'
send_dingtalk_alert(dingtalk_webhook, '服务器CPU使用率超过90%!')

总结

作为Python运维工程师，掌握这些指令是必备技能：

1. 系统信息获取 - psutil库的强大功能
2. 进程管理 - 监控和管理进程
3. 文件操作 - 文件和目录管理
4. Shell命令执行 - subprocess和paramiko
5. 日志管理 - 日志记录和分析
6. 定时任务 - 自动化执行
7. 网络监控 - 服务健康检查
8. 配置管理 - YAML/JSON/环境变量
9. 备份恢复 - 数据安全保障
10. 监控告警 - 及时发现问题

掌握这些工具，你就能构建强大的运维自动化体系。

关注我，获取更多Python技术干货！