第四十五篇：常用第三方模块——Pillow、requests、chardet、psutil
前言
Python 拥有庞大的第三方生态系统（PyPI），其中一些模块已成为事实上的行业标准。requests 是最流行的 HTTP 库，Pillow 是图像处理的标准工具，chardet 用于编码检测，psutil 提供系统监控能力。本文将逐一讲解这些实用模块的安装与核心用法。
requests —— 优雅的 HTTP 请求库
安装与环境配置
# 使用 pip 安装
pip install requests
# 如果网络较慢，可以使用国内镜像
pip install requests -i https://pypi.tuna.tsinghua.edu.cn/simple
# 验证安装
python -c "import requests; print(requests.__version__)"
GET 请求基本用法
import requests
# 基本 GET 请求
response = requests.get("https://httpbin.org/get")
print(f"状态码: {response.status_code}") # 200
print(f"响应头 Content-Type: {response.headers['Content-Type']}")
print(f"响应内容: {response.text[:100]}...")
# 带查询参数
params = {"q": "python", "page": 1, "size": 20}
response = requests.get("https://httpbin.org/get", params=params)
print(f"完整 URL: {response.url}")
# https://httpbin.org/get?q=python&page=1&size=20
# 检查响应状态
if response.status_code == 200:
data = response.json() # 自动解析 JSON
print(f"返回数据: {data['args']}")
# 使用 raise_for_status 抛出异常
try:
response.raise_for_status()
except requests.exceptions.HTTPError as e:
print(f"HTTP 错误: {e}")
POST 请求与表单处理
import requests
# POST 提交表单数据
form_data = {
"username": "alice",
"password": "secret123",
"remember": True
}
response = requests.post(
"https://httpbin.org/post",
data=form_data
)
result = response.json()
print(f"接收到的表单数据: {result['form']}")
# POST 提交 JSON 数据
json_data = {
"name": "Alice",
"age": 28,
"skills": ["Python", "JavaScript"]
}
response = requests.post(
"https://httpbin.org/post",
json=json_data # 自动序列化 JSON 并设置 Content-Type
)
result = response.json()
print(f"接收到的 JSON 数据: {result['json']}")
# PUT / DELETE 请求
put_response = requests.put("https://httpbin.org/put", json={"action": "update"})
delete_response = requests.delete("https://httpbin.org/delete")
高级请求特性
import requests
import json
# 自定义 Header
headers = {
"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64)",
"Accept": "application/json",
"Authorization": "Bearer your-token-here",
}
response = requests.get("https://httpbin.org/headers", headers=headers)
# Cookie 管理（Session 自动保存 cookie）
session = requests.Session()
# 第一次登录，cookie 会被 session 记住
login_response = session.post(
"https://httpbin.org/post",
data={"username": "alice", "password": "secret"}
)
# Session 持久化配置（一次设置，所有请求生效）
session.headers.update({"X-Custom-Header": "my-value"})
session.params.update({"api_key": "abc123"})
session.verify = True # 启用 SSL 证书验证
session.timeout = 10 # 超时时间（秒）
# 流式下载大文件
response = requests.get("https://httpbin.org/stream/20", stream=True)
for chunk in response.iter_content(chunk_size=8192):
if chunk:
print(f"收到 {len(chunk)} bytes")
异常处理
import requests
import time
def fetch_with_retry(url, max_retries=3, backoff_factor=1):
"""带重试机制的请求函数"""
for attempt in range(max_retries):
try:
response = requests.get(url, timeout=5)
response.raise_for_status()
return response
except requests.exceptions.Timeout:
wait_time = backoff_factor * (2 ** attempt)
print(f"超时，{wait_time}s 后重试 ({attempt + 1}/{max_retries})...")
except requests.exceptions.ConnectionError:
wait_time = backoff_factor * (2 ** attempt)
print(f"连接失败，{wait_time}s 后重试...")
except requests.exceptions.HTTPError as e:
print(f"HTTP 错误: {e}")
raise
time.sleep(wait_time)
raise requests.exceptions.ConnectionError(f"重试 {max_retries} 次后仍然失败")
try:
resp = fetch_with_retry("https://httpbin.org/delay/1")
print(f"成功! 状态码: {resp.status_code}")
except Exception as e:
print(f"最终失败: {e}")
实际用例：API 客户端封装
import requests
class APIClient:
"""通用 API 客户端封装"""
def __init__(self, base_url, api_key=None, timeout=10):
self.base_url = base_url.rstrip("/")
self.api_key = api_key
self.timeout = timeout
self.session = requests.Session()
self.session.headers.update({
"Accept": "application/json",
"Content-Type": "application/json",
})
if api_key:
self.session.headers["Authorization"] = f"Bearer {api_key}"
def get(self, endpoint, params=None):
url = f"{self.base_url}/{endpoint.lstrip('/')}"
response = self.session.get(url, params=params, timeout=self.timeout)
response.raise_for_status()
return response.json()
def post(self, endpoint, data=None):
url = f"{self.base_url}/{endpoint.lstrip('/')}"
response = self.session.post(url, json=data, timeout=self.timeout)
response.raise_for_status()
return response.json()
client = APIClient("https://httpbin.org", api_key="demo-key")
data = client.get("/get", params={"test": "value"})
posted = client.post("/post", data={"key": "value"})
Pillow —— 图像处理库
安装
pip install Pillow
基本图像操作
from PIL import Image, ImageFilter, ImageEnhance
img = Image.open("example.jpg")
print(f"尺寸: {img.size}") # (宽, 高)
print(f"模式: {img.mode}") # RGB, L, P, RGBA 等
print(f"格式: {img.format}") # JPEG, PNG 等
thumb = img.resize((200, 200)) # 强制拉伸
tthumb_fitted = img.thumbnail((200, 200)) # 保持比例
box = (100, 100, 500, 500)
cropped = img.crop(box)
rotated = img.rotate(45, expand=True)
flipped_h = img.transpose(Image.FLIP_LEFT_RIGHT)
rgb_img = img.convert("RGB")
gray_img = img.convert("L")
img.save("output.jpg", quality=90)
img.save("output.png", optimize=True)
img.save("output.webp", quality=80)
滤镜与效果
from PIL import Image, ImageFilter, ImageEnhance, ImageDraw, ImageFont
img = Image.open("example.jpg").convert("RGB")
gaussian_blur = img.filter(ImageFilter.GaussianBlur(radius=3))
sharpened = img.filter(ImageFilter.SHARPEN)
edge_detected = img.filter(ImageFilter.FIND_EDGES)
emboss = img.filter(ImageFilter.EMBOSS)
enhancer = ImageEnhance.Color(img)
colorized = enhancer.enhance(1.5)
enhancer = ImageEnhance.Brightness(img)
bright = enhancer.enhance(1.2)
watermark = Image.new("RGBA", img.size, (0, 0, 0, 0))
draw = ImageDraw.Draw(watermark)
try:
font = ImageFont.truetype("simhei.ttf", 40)
except IOError:
font = ImageFont.load_default()
text = "© My Website 2026"
position = (img.size[0] - 200, img.size[1] - 50)
draw.text(position, text, fill=(255, 255, 255, 128), font=font)
final = Image.alpha_composite(img.convert("RGBA"), watermark)
final_rgb = final.convert("RGB")
final_rgb.save("watermarked.jpg", quality=95)
chardet —— 编码检测
安装与基本用法
pip install chardet
import chardet
# 检测字节串的编码
sample_bytes = b'\x89PNG\r\n\x1a\nHello World!'
result = chardet.detect(sample_bytes)
print(result)
# {'encoding': 'ascii', 'confidence': 1.0, 'language': ''}
chinese_text = "你好世界！这是一个中文测试。".encode("utf-8")
result = chardet.detect(chinese_text)
print(f"编码: {result['encoding']}, 置信度: {result['confidence']:.2f}")
批量扫描目录编码
import chardet
import os
def scan_directory_encoding(dir_path):
text_extensions = {".txt", ".csv", ".md", ".py", ".html"}
results = []
for root, _, files in os.walk(dir_path):
for filename in sorted(files):
if os.path.splitext(filename)[1].lower() in text_extensions:
filepath = os.path.join(root, filename)
with open(filepath, "rb") as f:
raw = f.read(10000)
result = chardet.detect(raw)
results.append({
"file": os.path.relpath(filepath, dir_path),
"encoding": result["encoding"],
"confidence": f"{result['confidence']:.0%}",
})
print(f"\n=== 编码检测报告 ===")
for r in results:
print(f"{r['file']}: {r['encoding']} ({r['confidence']})")
return results
psutil —— 系统监控
安装
pip install psutil
CPU 信息
import psutil
print(f"逻辑核心数: {psutil.cpu_count()}")
print(f"物理核心数: {psutil.cpu_count(logical=False)}")
usage = psutil.cpu_percent(interval=1)
print(f"CPU 总使用率: {usage}%")
per_core = psutil.cpu_percent(interval=1, percpu=True)
print(f"各核心: {per_core}")
内存信息
import psutil
vmem = psutil.virtual_memory()
print(f"总内存: {vmem.total / 1024**3:.1f} GB")
print(f"可用内存: {vmem.available / 1024**3:.1f} GB")
print(f"使用率: {vmem.percent}%")
swap = psutil.swap_memory()
print(f"交换空间: {swap.used / 1024**3:.1f}GB ({swap.percent}%)")
net_io = psutil.net_io_counters()
print(f"网络 IO:\n 接收: {net_io.bytes_recv / 1024**2:.2f} MB\n 发送: {net_io.bytes_sent / 1024**2:.2f} MB")
进程管理
import psutil
def list_processes(top_n=10):
processes = []
for proc in psutil.process_iter(["pid", "name", "cpu_percent", "memory_percent"]):
try:
info = proc.info
processes.append(info)
except (psutil.NoSuchProcess, psutil.AccessDenied):
pass
processes.sort(key=lambda x: x.get("cpu_percent", 0) or 0, reverse=True)
print(f"{'PID':<8} {'CPU%':>6} {'MEM%':>6} {'名称'}")
print("-" * 40)
for p in processes[:top_n]:
pid = p["pid"]
cpu = p["cpu_percent"] or 0
mem = p["memory_percent"] or 0
name = p["name"] or "unknown"
print(f"{pid:<8} {cpu:>6.1f} {mem:>6.1f} {name}")
list_processes(5)
实时系统监控仪表盘
import psutil
import time
def monitor_system(duration=10, interval=1):
print("=== 实时系统监控 ===")
prev_net = psutil.net_io_counters()
for i in range(duration // interval):
timestamp = time.strftime("%H:%M:%S")
cpu = psutil.cpu_percent(interval=interval)
mem = psutil.virtual_memory()
disk = psutil.disk_usage("/")
curr_net = psutil.net_io_counters()
down_mbps = (curr_net.bytes_recv - prev_net.bytes_recv) / 1024**2 / interval
up_mbps = (curr_net.bytes_sent - prev_net.bytes_sent) / 1024**2 / interval
prev_net = curr_net
print(f"{timestamp} CPU:{cpu:.0f}% MEM:{mem.percent:.0f}% DISK:{disk.percent:.0f}% ↓{down_mbps:.1f}MB/s ↑{up_mbps:.1f}MB/s")
time.sleep(interval)
# monitor_system(duration=10, interval=1)
总结
本文系统讲解了四大常用第三方模块：
- requests — 优雅简洁的 HTTP 请求库，支持 Session/重传/认证/文件上传
- Pillow — Python 标准图像处理库，resize/crop/filter/水印/验证码
- chardet — 自动检测文件编码，辅助批量扫描目录
- psutil — 跨平台系统监控（CPU/内存/磁盘/网络/进程管理）
本文是 Python 教程系列的第四十五篇。