快捷菜单
常用功能一站直达
更多功能请点顶栏「快捷菜单」
给我出一份关于 运维方向的 python学习指导,我想在学习python语法的同时,可以做一些脚本或者工具,让自己有成就感,这样学习起来就不会很枯燥了。
核心思路:按语法难度分层,每学一个知识点就写一个运维实用小脚本,从「能跑通」到「能解决工作问题」,循序渐进不枯燥。全程聚焦运维真实场景:文件操作、日志处理、服务器监控、批量执行、备份、告警等。
变量、字符串/列表/字典、if/else 条件判断、for/while 循环、函数def、文件读写、异常处理try/except
场景:服务器上日志/备份文件命名不规范,一键批量重命名
1import os
2
3# 目标目录(改成你的路径)
4target_dir = "/tmp/test_logs"
5# 前缀自定义
6prefix = "app_log_"
7
8for index, filename in enumerate(os.listdir(target_dir)):
9 # 拼接新文件名
10 new_name = f"{prefix}{index}.log"
11 # 重命名
12 os.rename(os.path.join(target_dir, filename), os.path.join(target_dir, new_name))
13
14print("批量重命名完成!")场景:服务器日志占满磁盘,自动删除7天前日志
1import os
2import time
3
4log_path = "/var/log/nginx"
5# 7天前的时间戳
6seven_days_ago = time.time() - 7 * 24 * 60 * 60
7
8for file in os.listdir(log_path):
9 file_path = os.path.join(log_path, file)
10 # 判断文件修改时间
11 if os.path.isfile(file_path) and os.path.getmtime(file_path) < seven_days_ago:
12 os.remove(file_path)
13 print(f"已删除过期日志:{file}")场景:快速查看本机IP、磁盘根目录使用率
1import os
2import socket
3
4# 查看本机IP
5hostname = socket.gethostname()
6ip = socket.gethostbyname(hostname)
7print(f"主机名:{hostname},IP:{ip}")
8
9# 查看磁盘根目录使用情况(Linux)
10disk_info = os.popen("df -h /").read()
11print("磁盘根目录信息:")
12print(disk_info)os/shutil(文件/目录)、subprocess(执行shell命令)、datetime(时间)、logging(日志)、sys(系统参数)
场景:定时备份配置文件/代码目录,生成带日期的备份包
1import shutil
2import datetime
3
4# 源目录、备份目录
5source = "/etc/nginx"
6backup_dir = "/data/backup"
7# 备份文件名(日期+目录名)
8today = datetime.datetime.now().strftime("%Y%m%d")
9backup_name = f"nginx_backup_{today}.tar.gz"
10
11# 打包备份
12shutil.make_archive(base_name=f"{backup_dir}/{backup_name}", format='gztar', root_dir=source)
13print(f"备份完成:{backup_dir}/{backup_name}")场景:在Python中调用Linux命令,统一收集服务器信息
1import subprocess
2
3# 执行shell命令,捕获输出
4def run_cmd(cmd):
5 result = subprocess.run(cmd, shell=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE, encoding="utf-8")
6 return result.stdout, result.stderr
7
8# 示例:查看CPU信息
9cpu_info, err = run_cmd("lscpu | grep 'Model name'")
10print("CPU型号:", cpu_info.strip())
11
12# 示例:查看内存
13mem_info, _ = run_cmd("free -h")
14print("内存信息:\n", mem_info)场景:统计日志中ERROR/Exception出现次数,快速定位问题
1error_count = 0
2log_file = "/var/log/messages"
3
4with open(log_file, "r", encoding="utf-8", errors="ignore") as f:
5 for line in f:
6 if "ERROR" in line or "Exception" in line:
7 error_count += 1
8 print(f"错误行:{line.strip()}")
9
10print(f"日志中总错误数:{error_count}")1pip install psutil paramiko requests pyyamlpsutil:获取系统CPU/内存/磁盘/网络实时数据paramiko:SSH远程连接服务器,批量执行命令requests:调用接口、监控HTTP服务、发告警pyyaml:解析YAML配置文件(运维常用配置格式)场景:实时监控CPU、内存、磁盘使用率,超阈值提醒
1import psutil
2
3# 获取CPU使用率
4cpu_usage = psutil.cpu_percent(interval=1)
5# 获取内存使用率
6mem = psutil.virtual_memory()
7mem_usage = mem.percent
8# 获取磁盘根目录使用率
9disk = psutil.disk_usage("/")
10disk_usage = disk.percent
11
12print(f"CPU使用率:{cpu_usage}%")
13print(f"内存使用率:{mem_usage}%")
14print(f"磁盘使用率:{disk_usage}%")
15
16# 阈值告警
17if cpu_usage > 80 or mem_usage > 85 or disk_usage > 90:
18 print("⚠️ 资源使用率过高,请检查服务器!")场景:批量给10台服务器执行相同命令(如查看负载、重启服务)
1import paramiko
2
3# 服务器列表
4servers = [
5 {"host": "192.168.1.10", "user": "root", "pwd": "your_password"},
6 {"host": "192.168.1.11", "user": "root", "pwd": "your_password"},
7]
8# 要执行的命令
9cmd = "uptime"
10
11for server in servers:
12 ssh = paramiko.SSHClient()
13 ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())
14 ssh.connect(server["host"], username=server["user"], password=server["pwd"])
15 stdin, stdout, stderr = ssh.exec_command(cmd)
16 print(f"\n=== {server['host']} 执行结果 ===")
17 print(stdout.read().decode("utf-8"))
18 ssh.close()场景:监控业务接口是否正常,异常自动发钉钉群告警
1import requests
2
3def check_api(url):
4 try:
5 res = requests.get(url, timeout=5)
6 if res.status_code == 200:
7 return True, "接口正常"
8 else:
9 return False, f"状态码异常:{res.status_code}"
10 except Exception as e:
11 return False, f"请求失败:{str(e)}"
12
13# 监控接口
14api_url = "https://www.baidu.com"
15status, msg = check_api(api_url)
16
17if not status:
18 # 钉钉机器人Webhook(替换成你的)
19 ding_url = "https://oapi.dingtalk.com/robot/send?access_token=xxx"
20 data = {"msgtype": "text", "text": {"content": f"⚠️ 接口告警:{msg}"}}
21 requests.post(ding_url, json=data)
22 print("告警已发送!")
23else:
24 print("接口正常")多线程/多进程、数据库操作(pymysql)、简易Web界面(Flask)、定时任务
schedule库,定时巡检+邮件/钉钉告警需要我把某一个脚本扩展成完整可直接部署的版本(比如带配置文件、日志、异常处理的完整版)吗?
日历 · 精选 · 友链 · 更多
如果内容对你有帮助,欢迎请我喝杯咖啡 ☕



One的公众号
爱折腾博客的小白