Python入门——运维的第二把"瑞士军刀"
写在前面:Shell脚本擅长处理文本和系统操作,但当任务变复杂(比如解析JSON、操作数据库、调用API),Shell就显得力不从心了。Python就像运维的"第二把瑞士军刀"——语法简洁、库丰富、跨平台,从自动化运维到数据分析无所不能。本章带你从零开始,用运维视角学Python!
1 Python简介
1.1 Python的历史
Python由Guido van Rossum于1989年圣诞节期间创建,1991年发布第一个版本。名字来自英国喜剧"Monty Python's Flying Circus"(不是蟒蛇!)。
设计哲学:
- 代码可读性优先(强制缩进)
- 简洁优于复杂("There should be one obvious way to do it")
- batteries included(自带大量标准库)
1.2 Python 2 vs Python 3
| 特性 | Python 2 | Python 3 |
|---|---|---|
| 状态 | 已于2020年停止维护 | 当前主流版本 |
| 打印 | print "hello" | print("hello") |
| 字符串 | 默认ASCII | 默认Unicode |
| 除法 | 5/2 = 2 | 5/2 = 2.5 |
| 迭代器 | range()返回列表 | range()返回迭代器 |
结论:现在只用学Python 3! Python 2已彻底退役。
1.3 安装Python
CentOS / RHEL(使用yum):
# CentOS 7默认只有Python 2.7,需要安装Python 3
sudo yum install -y epel-release
sudo yum install -y python3 python3-pip
# 验证
python3 --version
pip3 --version
Ubuntu / Debian(使用apt):
# Ubuntu 20.04+默认自带Python 3
sudo apt update
sudo apt install -y python3 python3-pip python3-venv
# 验证
python3 --version
pip3 --version
从源码编译(获取最新版本):
# 1. 安装编译依赖
sudo yum install -y gcc openssl-devel bzip2-devel libffi-devel \
zlib-devel readline-devel sqlite-devel
# 2. 下载源码
cd /usr/src
sudo wget https://www.python.org/ftp/python/3.12.2/Python-3.12.2.tgz
sudo tar xzf Python-3.12.2.tgz
cd Python-3.12.2
# 3. 编译安装(altinstall不会覆盖系统Python)
sudo ./configure --enable-optimizations
sudo make altinstall
# 4. 验证
python3.12 --version
# 5. 创建软链接(可选)
sudo ln -sf /usr/local/bin/python3.12 /usr/bin/python3
互动提问:为什么源码安装用
make altinstall而不是make install?点击查看答案
因为很多Linux系统的工具(如yum)依赖Python 2。`make install` 会覆盖系统的 `python` 命令,可能导致系统工具崩溃。`altinstall` 只安装 `python3.x` 而不碰 `python`。
2 Python环境管理
2.1 虚拟环境(venv)
为什么需要虚拟环境?
想象你在学校做两个实验:实验A需要Python 3.8 + requests 2.25,实验B需要Python 3.12 + requests 2.31。两个版本冲突了怎么办?虚拟环境就是给每个项目创建一个"独立的房间",各房间的包互不干扰。
# 创建虚拟环境
python3 -m venv myproject_env
# 激活虚拟环境
source myproject_env/bin/activate
# 激活后命令行前面会出现 (myproject_env)
# 在虚拟环境中安装包
pip install requests flask
# 查看已安装的包
pip list
# 导出依赖列表
pip freeze > requirements.txt
# 根据依赖列表安装(部署时常用)
pip install -r requirements.txt
# 退出虚拟环境
deactivate
2.2 pip——Python的包管理器
# 安装包
pip install requests
pip install flask==2.3.0 # 指定版本
pip install "django>=4.0,<5.0" # 版本范围
# 卸载
pip uninstall requests
# 查看包信息
pip show requests
# 更新包
pip install --upgrade requests
# 使用国内镜像加速(非常重要!)
pip install requests -i https://pypi.tuna.tsinghua.edu.cn/simple
# 设置永久镜像
pip config set global.index-url https://pypi.tuna.tsinghua.edu.cn/simple
# 查看过期的包
pip list --outdated
3 交互式模式 vs 脚本模式
交互式模式(Python REPL)
$ python3
Python 3.12.2 (main, Feb 2024)
>>> print("Hello, World!")
Hello, World!
>>> 2 + 3
5
>>> import os
>>> os.getcwd()
'/home/student'
>>> exit()
适合: 快速测试一行代码、验证语法、计算表达式。
脚本模式
# 编写脚本文件
cat > hello.py << 'EOF'
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""我的第一个Python脚本"""
print("Hello, World!")
print("当前时间:", __import__('datetime').datetime.now())
EOF
# 运行
python3 hello.py
# 或者直接赋予执行权限
chmod +x hello.py
./hello.py
适合: 编写完整的程序、自动化脚本。
4 变量——给数据贴标签
生活类比
变量就像一个贴了标签的盒子:标签是变量名,盒子里装的是数据值。你可以随时换盒子里的东西,也可以给同一个东西贴多个标签。
4.1 命名规则
# 合法命名
name = "张三" # 普通变量
user_age = 20 # 推荐:蛇形命名(snake_case)
_private = True # 下划线开头表示私有
MAX_RETRY = 3 # 全大写表示常量
# 非法命名(会报错)
# 2name = "张三" # 不能以数字开头
# my-name = "张三" # 不能用连字符
# class = "高三" # 不能用关键字
4.2 多变量赋值
# 同时给多个变量赋相同的值
a = b = c = 0
# 同时给多个变量赋不同的值
name, age, city = "张三", 20, "北京"
print(f"姓名: {name}, 年龄: {age}, 城市: {city}")
# 交换两个变量(Python独有的优雅写法)
x, y = 10, 20
x, y = y, x # 交换!不需要临时变量
print(f"x={x}, y={y}") # x=20, y=10
4.3 动态类型
# Python是动态类型:变量不需要声明类型,且可以随时改变
x = 100 # x是整数
x = "hello" # 现在x变成了字符串(合法!)
x = [1, 2, 3] # 现在x变成了列表
# 查看变量类型
print(type(x)) # <class 'list'>
5 数据类型——Python的"积木块"
5.1 整数(int)与浮点数(float)
# 整数:没有大小限制
port = 8080
big_num = 10_000_000_000 # 可以用下划线分隔,增加可读性
hex_num = 0xFF # 十六进制
bin_num = 0b1010 # 二进制
oct_num = 0o755 # 八进制(Linux权限常用)
# 浮点数
pi = 3.14159
temperature = -15.5
scientific = 1.5e10 # 科学计数法
# 运维场景:计算磁盘使用率
total_disk = 500 # GB
used_disk = 387 # GB
usage_percent = used_disk / total_disk * 100
print(f"磁盘使用率: {usage_percent:.1f}%") # 磁盘使用率: 77.4%
5.2 字符串(str)
# 创建字符串
s1 = 'hello'
s2 = "world"
s3 = '''这是
多行
字符串'''
# 常用操作
hostname = "web-server-01"
print(len(hostname)) # 12(长度)
print(hostname.upper()) # WEB-SERVER-01
print(hostname.startswith("web")) # True
print(hostname.split("-")) # ['web', 'server', '01']
print(hostname.replace("01", "02")) # web-server-02
# 字符串格式化(三种方式)
name = "Nginx"
port = 80
# 方式1:f-string(推荐,Python 3.6+)
print(f"{name} 监听端口 {port}")
# 方式2:format方法
print("{} 监听端口 {}".format(name, port))
# 方式3:%格式化(老风格)
print("%s 监听端口 %d" % (name, port))
5.3 布尔值(bool)与 None
# 布尔值
is_running = True
has_error = False
# 布尔运算
print(True and False) # False
print(True or False) # True
print(not True) # False
# 哪些值被视为False?("假值")
# False, 0, 0.0, "", [], {}, (), None, set()
# None 表示"没有值"
result = None
if result is None:
print("还没有结果")
# 运维场景:检查服务状态
service_active = True
if service_active:
print("服务正常运行")
else:
print("服务已停止,请检查!")
5.4 类型转换
# 字符串 ↔ 数字
port_str = "8080"
port_int = int(port_str) # 字符串转整数
price = float("19.99") # 字符串转浮点
# 数字 → 字符串
age = 20
msg = "年龄: " + str(age) # 必须转换才能拼接
# 其他转换
num_list = list("abc") # ['a', 'b', 'c']
num_tuple = tuple([1, 2, 3]) # (1, 2, 3)
num_set = set([1, 2, 2, 3]) # {1, 2, 3}
# 进制转换(运维常用)
ip_part = 192
print(bin(ip_part)) # 0b11000000
print(hex(ip_part)) # 0xc0
print(oct(0o755)) # 0o755
互动提问:
int("3.14")会成功吗?点击查看答案
会报错!`int()` 不能直接转换含小数点的字符串。正确做法:`int(float("3.14"))` → 3,先转float再转int。
6 运算符
6.1 算术运算符
a, b = 17, 5
print(a + b) # 22 加法
print(a - b) # 12 减法
print(a * b) # 85 乘法
print(a / b) # 3.4 除法(结果是浮点数)
print(a // b) # 3 整除(向下取整)
print(a % b) # 2 取余
print(a ** b) # 1419857 幂运算(17的5次方)
# 运维场景:计算需要多少个磁盘分区
total_size = 1000 # GB
partition_size = 200 # GB
num_partitions = total_size // partition_size # 5个
remaining = total_size % partition_size # 剩余0GB
6.2 比较运算符
x = 10
print(x == 10) # True 等于
print(x != 10) # False 不等于
print(x > 5) # True 大于
print(x >= 10) # True 大于等于
print(x < 20) # True 小于
print(x <= 10) # True 小于等于
# 链式比较(Python特色)
age = 18
print(16 <= age < 60) # True(等价于 16<=age and age<60)
# is vs ==
a = [1, 2, 3]
b = [1, 2, 3]
print(a == b) # True (值相等)
print(a is b) # False (不是同一个对象)
6.3 逻辑运算符
cpu_usage = 85
mem_usage = 70
# and:两个都为True才是True
if cpu_usage > 80 and mem_usage > 80:
print("CPU和内存都告警!")
# or:至少一个为True就是True
if cpu_usage > 80 or mem_usage > 80:
print("至少一个指标告警!") # 会输出
# not:取反
is_ok = True
if not is_ok:
print("有问题")
6.4 赋值运算符
x = 10
x += 5 # x = x + 5 → 15
x -= 3 # x = x - 3 → 12
x *= 2 # x = x * 2 → 24
x //= 5 # x = x // 5 → 4
x **= 3 # x = x ** 3 → 64
x %= 10 # x = x % 10 → 4
6.5 位运算符
# 位运算在权限处理中非常有用
a = 0b1100 # 12
b = 0b1010 # 10
print(bin(a & b)) # 0b1000 按位与
print(bin(a | b)) # 0b1110 按位或
print(bin(a ^ b)) # 0b0110 按位异或
print(bin(~a)) # -0b1101 按位取反
print(bin(a << 1)) # 0b11000 左移
print(bin(a >> 1)) # 0b110 右移
# 运维场景:Linux权限掩码计算
# rwx = 7 (111), rw- = 6 (110), r-x = 5 (101)
owner = 0b111 # rwx = 7
group = 0b101 # r-x = 5
other = 0b100 # r-- = 4
perm = (owner << 6) | (group << 3) | other
print(f"权限: {oct(perm)}") # 0o754
7 流程控制
7.1 if / elif / else
# 根据CPU使用率判断告警等级
cpu_usage = 92
if cpu_usage >= 90:
level = "严重"
action = "立即处理"
elif cpu_usage >= 80:
level = "警告"
action = "关注观察"
elif cpu_usage >= 60:
level = "提示"
action = "正常偏高"
else:
level = "正常"
action = "无需处理"
print(f"CPU: {cpu_usage}% => [{level}] {action}")
7.2 三元表达式
# 条件为真时的值 if 条件 else 条件为假时的值
age = 20
status = "成年" if age >= 18 else "未成年"
print(status) # 成年
# 运维场景:一行判断服务状态
is_active = True
display = "运行中" if is_active else "已停止"
# 端口检查
port = 80
port_type = "标准端口" if port < 1024 else "非标准端口"
12.8 for循环
8.1 range()——数字序列
# range(stop)
for i in range(5): # 0, 1, 2, 3, 4
print(i, end=" ") # 输出:0 1 2 3 4
print()
# range(start, stop)
for i in range(1, 6): # 1, 2, 3, 4, 5
print(i, end=" ")
print()
# range(start, stop, step)
for i in range(0, 100, 10): # 0, 10, 20, ..., 90
print(i, end=" ")
print()
# 倒序
for i in range(10, 0, -1): # 10, 9, 8, ..., 1
print(i, end=" ")
print()
# 运维场景:批量检测端口
ports = [22, 80, 443, 3306, 6379, 8080]
for port in ports:
print(f"检查端口 {port} ...")
8.2 enumerate()——带索引遍历
# enumerate给每个元素加上编号
servers = ["web-01", "web-02", "db-01", "redis-01"]
for i, server in enumerate(servers, start=1):
print(f"第{i}台服务器: {server}")
# 输出:
# 第1台服务器: web-01
# 第2台服务器: web-02
# 第3台服务器: db-01
# 第4台服务器: redis-01
# 运维场景:给配置文件中的每个server编号
config_lines = [
"server 192.168.1.1:8080 weight=5;",
"server 192.168.1.2:8080 weight=3;",
"server 192.168.1.3:8080 backup;",
]
for idx, line in enumerate(config_lines, start=1):
print(f"# server_{idx}")
print(line)
8.3 zip()——并行遍历
# zip把多个列表"拉链"在一起
hosts = ["web-01", "web-02", "db-01"]
ips = ["192.168.1.10", "192.168.1.11", "192.168.1.20"]
roles = ["web", "web", "database"]
for host, ip, role in zip(hosts, ips, roles):
print(f"{host:10s} => {ip:16s} 角色: {role}")
# 运维场景:生成hosts文件
print("\n=== /etc/hosts ===")
for ip, host in zip(ips, hosts):
print(f"{ip} {host}")
# 用zip构造字典
keys = ["hostname", "ip", "os", "cpu_cores"]
values = ["web-01", "10.0.0.1", "Ubuntu 22.04", 4]
server_info = dict(zip(keys, values))
print(server_info)
12.9 while循环
# 基本while循环
count = 0
while count < 5:
print(f"第 {count+1} 次检测...")
count += 1
# 运维场景:重试机制
import time
max_retries = 3
retry = 0
success = False
while retry < max_retries:
retry += 1
print(f"尝试连接数据库... (第{retry}次)")
# 模拟:第2次成功
if retry == 2:
success = True
break
time.sleep(1)
if success:
print("连接成功!")
else:
print(f"连接失败,已重试{max_retries}次")
9.1 break与continue
# break:找到第一个异常服务就停止
services = ["nginx", "mysql", "redis", "docker"]
for svc in services:
if svc == "redis":
print(f"[错误] {svc} 已停止运行!")
break
print(f"[正常] {svc} 运行中")
# continue:跳过某个,继续检查其他的
print("\n--- 磁盘检查 ---")
mounts = ["/", "/boot", "/tmp", "/home", "/var"]
for mount in mounts:
if mount == "/tmp":
print(f" 跳过临时目录 {mount}")
continue
print(f" 检查 {mount} ...")
9.2 while-else(Python特色)
# else块在循环正常结束(没有break)时执行
port = 8080
while port <= 8090:
# 模拟:端口8085被占用
if port == 8085:
print(f"端口 {port} 已被占用!")
break
port += 1
else:
# 只有循环没有被break时才执行
print(f"找到可用端口: {port}")
# 运维场景:检查服务是否全部启动
services_to_check = ["nginx", "mysql", "redis"]
all_ok = True
for svc in services_to_check:
# 模拟检查
is_running = (svc != "redis") # redis没启动
if not is_running:
print(f"[失败] {svc} 未启动")
all_ok = False
break
else:
print("[成功] 所有服务正常运行")
if not all_ok:
print("请检查失败的服务")
10 列表(list)——最常用的数据容器
生活类比
列表就像一个购物清单:可以添加项目、删除项目、排序、查找——而且清单里的东西可以重复,也可以随时更换。
10.1 创建与基本操作
# 创建列表
fruits = ["apple", "banana", "cherry"]
ports = [22, 80, 443, 3306, 6379]
mixed = [1, "hello", True, 3.14] # 可以混合类型
empty = []
# 访问元素(索引从0开始)
print(fruits[0]) # apple
print(fruits[-1]) # cherry(最后一个)
# 修改元素
fruits[1] = "orange"
# 长度
print(len(fruits)) # 3
# 判断元素是否在列表中
if "apple" in fruits:
print("有苹果!")
if "grape" not in fruits:
print("没有葡萄")
12.10.2 列表方法
servers = ["web-01", "web-02"]
# 添加
servers.append("web-03") # 末尾添加
servers.insert(0, "web-00") # 在指定位置插入
servers.extend(["db-01", "db-02"]) # 批量添加
# 删除
servers.remove("web-00") # 按值删除
popped = servers.pop() # 弹出最后一个
servers.pop(0) # 弹出指定位置
# 查找
idx = servers.index("web-02") # 查找索引
count = servers.count("web-01") # 统计出现次数
# 排序
numbers = [3, 1, 4, 1, 5, 9, 2, 6]
numbers.sort() # 原地排序:[1, 1, 2, 3, 4, 5, 6, 9]
numbers.sort(reverse=True) # 降序
sorted_nums = sorted(numbers) # 返回新列表,不改变原列表
# 翻转
servers.reverse()
print(servers)
10.3 列表切片——强大的子集提取
log_entries = [
"2024-01-01 INFO 启动成功",
"2024-01-01 WARN 磁盘80%",
"2024-01-01 ERROR 连接失败",
"2024-01-02 INFO 服务恢复",
"2024-01-02 INFO 用户登录",
"2024-01-03 ERROR 超时",
"2024-01-03 INFO 备份完成",
]
# 基本切片 [start:stop:step]
print(log_entries[0:3]) # 前3条
print(log_entries[3:]) # 第4条到最后
print(log_entries[:3]) # 前3条(等价于0:3)
print(log_entries[-2:]) # 最后2条
print(log_entries[::2]) # 每隔一条取一个
print(log_entries[::-1]) # 反转列表
# 切片赋值
nums = [0, 1, 2, 3, 4, 5]
nums[1:4] = [10, 20, 30] # 替换1~3号元素
print(nums) # [0, 10, 20, 30, 4, 5]
10.4 列表推导式——一行搞定列表生成
# 基本语法:[表达式 for 变量 in 可迭代对象 if 条件]
# 生成1~10的平方
squares = [x**2 for x in range(1, 11)]
print(squares) # [1, 4, 9, 16, 25, 36, 49, 64, 81, 100]
# 筛选偶数
evens = [x for x in range(1, 21) if x % 2 == 0]
# 运维场景:批量生成服务器名
servers = [f"web-{i:02d}" for i in range(1, 11)]
print(servers) # ['web-01', 'web-02', ..., 'web-10']
# 生成IP列表
ips = [f"192.168.1.{i}" for i in range(1, 255)]
# 过滤ERROR日志
logs = ["INFO ok", "ERROR fail", "INFO done", "ERROR timeout"]
errors = [log for log in logs if "ERROR" in log]
print(errors) # ['ERROR fail', 'ERROR timeout']
# 嵌套推导式:生成端口矩阵
port_matrix = [(host, port) for host in ["web", "db"] for port in [80, 443]]
print(port_matrix)
11 元组(tuple)——不可变的列表
# 元组一旦创建就不能修改(不可变)
# 适合存储不应该被改变的数据
# 创建
point = (3, 4)
http_status = (200, 301, 404, 500)
single = (42,) # 单元素元组必须加逗号
# 解包
x, y = point
print(f"x={x}, y={y}")
# 用元组做函数多返回值
def check_server(host):
is_alive = True
response_time = 25 # ms
return is_alive, response_time
alive, rt = check_server("web-01")
print(f"在线: {alive}, 响应: {rt}ms")
# 运维场景:定义不可变的配置
SERVERS = (
("web-01", "192.168.1.10", 80),
("web-02", "192.168.1.11", 80),
("db-01", "192.168.1.20", 3306),
)
for name, ip, port in SERVERS:
print(f"{name}: {ip}:{port}")
12 字典(dict)——键值对的"通讯录"
生活类比
字典就像通讯录:用名字(键)查电话号码(值),不需要从头翻到尾。
12.1 创建与基本操作
# 创建字典
server = {
"hostname": "web-01",
"ip": "192.168.1.10",
"os": "Ubuntu 22.04",
"cpu_cores": 4,
"memory_gb": 16,
"services": ["nginx", "docker"],
}
# 访问
print(server["hostname"]) # web-01
print(server.get("disk", "未知")) # 未知(键不存在时返回默认值)
# 添加/修改
server["disk_gb"] = 500
server["cpu_cores"] = 8 # 修改
# 删除
del server["disk_gb"]
removed = server.pop("memory_gb") # 删除并返回值
# 判断键是否存在
if "ip" in server:
print(f"IP: {server['ip']}")
# 长度
print(len(server))
12.2 字典方法
server = {"name": "web-01", "ip": "10.0.0.1", "port": 80}
# 所有键
print(server.keys()) # dict_keys(['name', 'ip', 'port'])
# 所有值
print(server.values()) # dict_values(['web-01', '10.0.0.1', 80])
# 所有键值对
print(server.items()) # dict_items([('name', 'web-01'), ...])
# 更新字典
server.update({"port": 8080, "env": "production"})
# 设置默认值(键不存在时才设置)
server.setdefault("timeout", 30)
# 清空
# server.clear()
# fromkeys:批量创建
ports_status = dict.fromkeys([22, 80, 443, 3306], "unknown")
print(ports_status) # {22: 'unknown', 80: 'unknown', ...}
12.3 遍历字典
server = {"name": "web-01", "ip": "10.0.0.1", "port": 80, "env": "prod"}
# 遍历键值对
print("=== 服务器信息 ===")
for key, value in server.items():
print(f" {key:10s} : {value}")
# 只遍历键
for key in server:
print(key)
# 字典推导式
# 生成端口→服务名的映射
port_service = {port: f"service_{port}" for port in [22, 80, 443]}
print(port_service)
# 过滤:只保留使用率>80%的磁盘
disk_usage = {"/": 75, "/boot": 45, "/var": 92, "/tmp": 88, "/home": 30}
high_usage = {k: v for k, v in disk_usage.items() if v > 80}
print(f"高使用率磁盘: {high_usage}")
13 集合(set)——去重与集合运算
# 集合:无序、不重复
# 主要用于:去重、集合运算
# 创建
fruits = {"apple", "banana", "cherry"}
nums = set([1, 2, 2, 3, 3, 3]) # {1, 2, 3} 自动去重
# 添加/删除
fruits.add("orange")
fruits.discard("banana") # 删除(不存在也不报错)
fruits.remove("cherry") # 删除(不存在会报错)
# 判断
if "apple" in fruits:
print("有苹果")
# 集合运算(数学中的集合操作)
web_servers = {"web-01", "web-02", "web-03"}
monitoring = {"web-01", "db-01", "redis-01"}
# 交集:同时出现在两个集合中的
both = web_servers & monitoring # {'web-01'}
print(f"交集(都有的): {both}")
# 并集:两个集合合并去重
all_servers = web_servers | monitoring # {'web-01','web-02','web-03','db-01','redis-01'}
print(f"并集(全部): {all_servers}")
# 差集:在A中但不在B中的
only_web = web_servers - monitoring # {'web-02', 'web-03'}
print(f"差集(仅Web): {only_web}")
# 对称差集:两个集合各自独有的
unique = web_servers ^ monitoring
print(f"对称差集: {unique}")
# 运维场景:找出哪些服务器没有安装监控agent
all_servers = {"web-01", "web-02", "db-01", "redis-01", "cache-01"}
monitored = {"web-01", "web-02", "db-01"}
unmonitored = all_servers - monitored
print(f"未安装监控的服务器: {unmonitored}")
14 字符串方法详解
14.1 split()——拆分字符串
# 按空格拆分
cmd_output = "web-01 192.168.1.10 Ubuntu running"
parts = cmd_output.split()
print(parts) # ['web-01', '192.168.1.10', 'Ubuntu', 'running']
# 按指定分隔符拆分
csv_line = "web-01,192.168.1.10,Ubuntu,running"
fields = csv_line.split(",")
print(fields)
# 拆分/etc/passwd格式
passwd_line = "nginx:x:997:995:Nginx web server:/var/lib/nginx:/sbin/nologin"
parts = passwd_line.split(":")
print(f"用户名: {parts[0]}, Shell: {parts[-1]}")
# 限制拆分次数
log = "2024-01-01 12:00:00 ERROR something happened"
date_time, rest = log.split(" ", 2)[:2], log.split(" ", 2)[2]
14.2 join()——合并字符串
# join是split的逆操作
servers = ["web-01", "web-02", "web-03"]
# 用逗号连接
csv = ",".join(servers)
print(csv) # web-01,web-02,web-03
# 用换行连接
lines = "\n".join(servers)
print(lines)
# 生成SQL IN条件
ids = ["1", "2", "3", "4"]
sql_in = "IN ({})".format(",".join(ids))
print(sql_in) # IN (1,2,3,4)
# 生成路径
import os
path_parts = ["/var", "log", "nginx", "access.log"]
full_path = "/".join(path_parts)
print(full_path)
14.3 strip()——去除空白
# 去除两端空白(处理用户输入、文件读取时常用)
user_input = " hello world \n"
clean = user_input.strip() # "hello world"
left = user_input.lstrip() # "hello world \n"
right = user_input.rstrip() # " hello world"
# 去除指定字符
filename = "report_2024.csv.bak"
clean_name = filename.rstrip(".bak") # 注意:这会去除任意组合
# 运维场景:读取配置文件
config_line = " server_name = example.com ;"
# 去除空白和注释
clean_config = config_line.strip().split(";")[0].strip()
print(clean_config) # server_name = example.com
14.4 format() 与 f-string
hostname = "web-01"
ip = "10.0.0.1"
cpu = 75.5
mem = 82.3
# format方法
print("服务器 {} ({}) CPU: {:.1f}% MEM: {:.1f}%".format(hostname, ip, cpu, mem))
# f-string(推荐,Python 3.6+)
print(f"服务器 {hostname} ({ip}) CPU: {cpu:.1f}% MEM: {mem:.1f}%")
# f-string高级用法
# 对齐
for name, status in [("nginx", "running"), ("mysql", "stopped"), ("redis", "running")]:
print(f" {name:<10s} {status:>10s}")
# 输出:
# nginx running
# mysql stopped
# redis running
# 千分位
traffic = 1234567890
print(f"流量: {traffic:,} bytes") # 流量: 1,234,567,890 bytes
# 百分比
success = 987
total = 1000
print(f"成功率: {success/total:.2%}") # 成功率: 98.70%
# 日期格式化
from datetime import datetime
now = datetime.now()
print(f"当前时间: {now:%Y-%m-%d %H:%M:%S}")
15 综合实战:运维脚本示例
15.1 服务器信息收集脚本
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
服务器信息收集脚本
功能:收集系统基本信息并生成报告
"""
import os
import platform
import socket
from datetime import datetime
def get_system_info():
"""收集系统信息"""
info = {
"hostname": socket.gethostname(),
"platform": platform.platform(),
"python_version": platform.python_version(),
"architecture": platform.machine(),
"processor": platform.processor() or "unknown",
"username": os.getenv("USER", os.getenv("USERNAME", "unknown")),
"home_dir": os.path.expanduser("~"),
"cwd": os.getcwd(),
"pid": os.getpid(),
}
return info
def get_disk_info():
"""获取磁盘信息(简化版)"""
result = os.popen("df -h 2>/dev/null").readlines()
disks = []
for line in result[1:]:
parts = line.split()
if len(parts) >= 6:
disks.append({
"filesystem": parts[0],
"size": parts[1],
"used": parts[2],
"available": parts[3],
"use_percent": parts[4],
"mount": parts[5],
})
return disks
def generate_report(info, disks):
"""生成报告"""
report_lines = []
report_lines.append("=" * 60)
report_lines.append(f" 服务器巡检报告")
report_lines.append(f" 生成时间: {datetime.now():%Y-%m-%d %H:%M:%S}")
report_lines.append("=" * 60)
report_lines.append("\n【系统信息】")
for key, value in info.items():
report_lines.append(f" {key:<20s}: {value}")
report_lines.append("\n【磁盘信息】")
report_lines.append(f" {'挂载点':<15s} {'大小':>8s} {'已用':>8s} {'使用率':>8s}")
report_lines.append(" " + "-" * 45)
for d in disks:
pct = d["use_percent"].replace("%", "")
flag = " !!!" if int(pct) > 80 else ""
report_lines.append(
f" {d['mount']:<15s} {d['size']:>8s} {d['used']:>8s} "
f"{d['use_percent']:>8s}{flag}"
)
report_lines.append("\n" + "=" * 60)
return "\n".join(report_lines)
# 主程序
if __name__ == "__main__":
sys_info = get_system_info()
disk_info = get_disk_info()
report = generate_report(sys_info, disk_info)
print(report)
# 保存到文件
filename = f"report_{datetime.now():%Y%m%d_%H%M%S}.txt"
with open(f"/tmp/{filename}", "w", encoding="utf-8") as f:
f.write(report)
print(f"\n报告已保存到: /tmp/{filename}")
15.2 日志分析脚本
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
Nginx日志分析脚本
分析访问日志,统计访问量、状态码、热门URL等
"""
from collections import Counter
def parse_log_line(line):
"""解析单行Nginx日志"""
parts = line.split()
if len(parts) < 9:
return None
return {
"ip": parts[0],
"time": parts[3].strip("["),
"method": parts[5].strip('"'),
"url": parts[6],
"status": int(parts[8]),
}
def analyze_log(log_file):
"""分析日志文件"""
ip_counter = Counter()
status_counter = Counter()
url_counter = Counter()
total = 0
errors = 0
with open(log_file, "r") as f:
for line in f:
entry = parse_log_line(line)
if entry is None:
continue
total += 1
ip_counter[entry["ip"]] += 1
status_counter[entry["status"]] += 1
url_counter[entry["url"]] += 1
if entry["status"] >= 400:
errors += 1
# 打印报告
print(f"\n{'='*50}")
print(f" Nginx日志分析报告")
print(f"{'='*50}")
print(f"\n总请求数: {total:,}")
print(f"错误请求: {errors:,} ({errors/max(total,1)*100:.1f}%)")
print(f"\n--- Top 5 访问IP ---")
for ip, count in ip_counter.most_common(5):
print(f" {ip:<20s} {count:>6,} 次")
print(f"\n--- 状态码分布 ---")
for code, count in sorted(status_counter.items()):
pct = count / total * 100
print(f" {code}: {count:>6,} ({pct:.1f}%)")
print(f"\n--- Top 5 热门URL ---")
for url, count in url_counter.most_common(5):
print(f" {url:<40s} {count:>6,} 次")
if __name__ == "__main__":
import sys
log_path = sys.argv[1] if len(sys.argv) > 1 else "/var/log/nginx/access.log"
try:
analyze_log(log_path)
except FileNotFoundError:
print(f"日志文件不存在: {log_path}")
except PermissionError:
print(f"权限不足,请使用 sudo 运行")
运维常用命令精选
环境与包管理
python3 -m venv myenv && source myenv/bin/activate # 创建并激活虚拟环境
pip install -r requirements.txt # 安装依赖
pip freeze > requirements.txt # 导出依赖
pip install package -i https://pypi.tuna.tsinghua.edu.cn/simple # 国内镜像
运行与调试
python3 script.py # 运行脚本
python3 -i script.py # 运行后进入交互模式(调试用)
python3 -m py_compile script.py # 检查语法错误
python3 -c "import os; print(os.getcwd())" # 运行单行命令
运维常用Python单行命令
# 生成随机密码
python3 -c "import secrets; print(secrets.token_urlsafe(16))"
# 格式化JSON
echo '{"a":1,"b":2}' | python3 -m json.tool
# 启动简易HTTP服务器(分享文件)
python3 -m http.server 8080
# 计算MD5
python3 -c "import hashlib; print(hashlib.md5(b'hello').hexdigest())"
趣味命令专区
# 1. Python之禅(The Zen of Python)
python3 -c "import this"
# 2. 用Python画一个ASCII爱心
python3 -c "
for y in range(15,-16,-1):
line = ''
for x in range(-30,31):
a, b = x/10, y/7
line += '*' if (a**2+b**2-1)**3 - a**2*b**3 <= 0 else ' '
print(line)
"
# 3. 随机名言
python3 -c "
import random
quotes = [
'代码是写给人看的,顺便能在机器上运行。—— Harold Abelson',
'简单优于复杂。—— Python之禅',
'过早优化是万恶之源。—— Donald Knuth',
'好的软件,像好的酒,需要时间。—— Joel Spolsky',
]
print(random.choice(quotes))
"
# 4. 终端进度条
python3 -c "
import time, sys
for i in range(51):
bar = '█' * i + '░' * (50-i)
sys.stdout.write(f'\r[{bar}] {i*2}%')
sys.stdout.flush()
time.sleep(0.05)
print()
"
# 5. 用Python查看当前IP
python3 -c "
import urllib.request
import json
try:
data = json.loads(urllib.request.urlopen('https://httpbin.org/ip').read())
print(f'你的公网IP: {data[\"origin\"]}')
except:
print('无法获取IP')
"
# 6. 一行生成二维码(需安装qrcode库)
# pip install qrcode
python3 -c "
import qrcode
img = qrcode.make('https://github.com')
img.save('/tmp/qrcode.png')
print('二维码已保存到 /tmp/qrcode.png')
"
本章小结:Python以其简洁的语法和强大的生态,成为运维工程师的"第二语言"。本章覆盖了变量、数据类型、流程控制、四大容器(列表/元组/字典/集合)和字符串操作。记住:Shell处理系统操作和简单文本,Python处理复杂逻辑和数据。两者配合,运维自动化效率翻倍!下一章我们将用Python编写真正的运维自动化工具。