python Python标准库-之十二
·
第12篇:Python标准库
目录
Python标准库概述
Python标准库是Python发行版中包含的一组模块和包,提供了丰富的功能,涵盖了从文本处理到网络编程的各种需求。标准库遵循"电池已包含"的哲学,使得Python能够处理各种常见的编程任务而无需安装额外的第三方库。
标准库的特点
- 丰富性:涵盖了各种常见编程需求
- 一致性:遵循统一的设计原则和接口规范
- 可靠性:经过充分测试,稳定可靠
- 跨平台:在不同操作系统上保持一致的行为
- 文档完善:有详细的官方文档支持
标准库的组织结构
# Python标准库按功能领域组织
# 文本处理、数据结构、文件操作、网络编程等
import sys
print("Python版本:", sys.version)
# 查看标准库模块
import pkgutil
import sys
print("部分标准库模块:")
for importer, modname, ispkg in pkgutil.iter_modules(sys.path):
if not ispkg and not modname.startswith('_'):
print(f" {modname}")
if len(modname) > 20: # 限制输出数量
break
文本处理模块
Python标准库提供了多个用于文本处理的模块。
string模块
import string
# 常用字符串常量
print("ASCII字母:", string.ascii_letters)
print("小写字母:", string.ascii_lowercase)
print("大写字母:", string.ascii_uppercase)
print("数字:", string.digits)
print("标点符号:", string.punctuation)
# 字符串模板
from string import Template
template = Template("Hello, $name! Welcome to $place.")
result = template.substitute(name="张三", place="Python世界")
print(result) # 输出:Hello, 张三! Welcome to Python世界.
# 安全替换(缺失的变量不会报错)
template = Template("Hello, $name! You are $age years old.")
result = template.safe_substitute(name="李四")
print(result) # 输出:Hello, 李四! You are $age years old.
re模块(正则表达式)
import re
# 基本匹配
text = "我的电话号码是13812345678,邮箱是example@email.com"
phone_pattern = r"1[3-9]\d{9}"
email_pattern = r"\w+@\w+\.\w+"
phones = re.findall(phone_pattern, text)
emails = re.findall(email_pattern, text)
print("电话号码:", phones)
print("邮箱:", emails)
# 分组和捕获
pattern = r"(\d{4})-(\d{2})-(\d{2})"
date_text = "今天是2023-12-25"
match = re.search(pattern, date_text)
if match:
year, month, day = match.groups()
print(f"年: {year}, 月: {month}, 日: {day}")
# 替换
text = "Python is great. Python is powerful."
new_text = re.sub(r"Python", "Java", text)
print(new_text) # 输出:Java is great. Java is powerful.
difflib模块
import difflib
# 文本差异比较
text1 = ["第一行", "第二行", "第三行"]
text2 = ["第一行", "第二行修改", "第三行", "新增行"]
diff = difflib.unified_diff(text1, text2, lineterm='')
print("文本差异:")
for line in diff:
print(line)
# 相似度比较
from difflib import SequenceMatcher
def similarity(a, b):
return SequenceMatcher(None, a, b).ratio()
text1 = "Hello, World!"
text2 = "Hello, Python!"
print(f"相似度: {similarity(text1, text2):.2f}")
# 最近匹配
from difflib import get_close_matches
words = ['apple', 'application', 'apply', 'appreciate', 'approach']
word = 'app'
matches = get_close_matches(word, words)
print(f"'{word}' 的近似匹配: {matches}")
数据结构模块
Python标准库提供了多种高级数据结构。
collections模块
from collections import namedtuple, deque, Counter, defaultdict, OrderedDict
# namedtuple - 命名元组
Point = namedtuple('Point', ['x', 'y'])
p = Point(1, 2)
print(f"点坐标: ({p.x}, {p.y})")
print(f"点坐标: {p}")
# deque - 双端队列
d = deque([1, 2, 3])
d.append(4) # 右端添加
d.appendleft(0) # 左端添加
print("双端队列:", d)
d.pop() # 右端删除
d.popleft() # 左端删除
print("操作后:", d)
# Counter - 计数器
text = "hello world"
counter = Counter(text)
print("字符计数:", counter)
print("最常见的字符:", counter.most_common(3))
# defaultdict - 默认字典
dd = defaultdict(list)
dd['fruits'].append('apple')
dd['fruits'].append('banana')
dd['vegetables'].append('carrot')
print("默认字典:", dict(dd))
# OrderedDict - 有序字典
od = OrderedDict()
od['first'] = 1
od['second'] = 2
od['third'] = 3
print("有序字典:", od)
heapq模块
import heapq
# 堆操作
data = [3, 1, 4, 1, 5, 9, 2, 6, 5, 3]
heapq.heapify(data) # 转换为堆
print("堆:", data)
# 压入元素
heapq.heappush(data, 0)
print("压入0后:", data)
# 弹出最小元素
min_val = heapq.heappop(data)
print(f"弹出最小值: {min_val}")
print("弹出后:", data)
# 获取最小的n个元素
smallest = heapq.nsmallest(3, [3, 1, 4, 1, 5, 9, 2, 6])
print("最小的3个数:", smallest)
# 获取最大的n个元素
largest = heapq.nlargest(3, [3, 1, 4, 1, 5, 9, 2, 6])
print("最大的3个数:", largest)
array模块
import array
# 创建数组
# 类型码: 'i'表示整数, 'f'表示浮点数, 'd'表示双精度浮点数
int_array = array.array('i', [1, 2, 3, 4, 5])
float_array = array.array('f', [1.1, 2.2, 3.3, 4.4, 5.5])
print("整数数组:", int_array)
print("浮点数组:", float_array)
# 数组操作
int_array.append(6)
int_array.extend([7, 8, 9])
print("扩展后:", int_array)
# 访问元素
print("第一个元素:", int_array[0])
print("最后三个元素:", int_array[-3:])
# 数组大小
print("数组长度:", len(int_array))
print("数组类型:", int_array.typecode)
文件和目录访问
Python提供了多种处理文件和目录的模块。
os.path模块
import os.path
# 路径操作
path = "/home/user/documents/file.txt"
print("目录:", os.path.dirname(path))
print("文件名:", os.path.basename(path))
print("文件名(无扩展名):", os.path.splitext(os.path.basename(path))[0])
print("扩展名:", os.path.splitext(path)[1])
# 路径组合
base_dir = "/home/user"
filename = "document.txt"
full_path = os.path.join(base_dir, filename)
print("完整路径:", full_path)
# 路径检查
print("路径存在吗?", os.path.exists(path))
print("是文件吗?", os.path.isfile(path))
print("是目录吗?", os.path.isdir(path))
print("是绝对路径吗?", os.path.isabs(path))
# 获取文件信息
if os.path.exists(path):
print("文件大小:", os.path.getsize(path))
print("创建时间:", os.path.getctime(path))
print("修改时间:", os.path.getmtime(path))
pathlib模块(推荐)
from pathlib import Path
# 创建路径对象
p = Path("/home/user/documents/file.txt")
# 路径属性
print("父目录:", p.parent)
print("文件名:", p.name)
print("文件 stem:", p.stem)
print("扩展名:", p.suffix)
# 路径操作
new_path = p.parent / "new_file.txt"
print("新路径:", new_path)
# 目录操作
current_dir = Path(".")
print("当前目录内容:")
for item in current_dir.iterdir():
if item.is_file():
print(f" 文件: {item}")
elif item.is_dir():
print(f" 目录: {item}")
# 通配符匹配
print("所有Python文件:")
for py_file in current_dir.glob("*.py"):
print(f" {py_file}")
# 递归匹配
print("所有子目录中的txt文件:")
for txt_file in current_dir.rglob("*.txt"):
print(f" {txt_file}")
glob模块
import glob
# 文件模式匹配
print("当前目录的所有Python文件:")
py_files = glob.glob("*.py")
for file in py_files:
print(f" {file}")
# 递归匹配
print("所有子目录中的Python文件:")
all_py_files = glob.glob("**/*.py", recursive=True)
for file in all_py_files:
print(f" {file}")
# 使用通配符
print("以数字开头的文件:")
number_files = glob.glob("[0-9]*.*")
for file in number_files:
print(f" {file}")
数据持久化
Python提供了多种数据持久化的方案。
pickle模块
import pickle
# 数据序列化
data = {
'name': '张三',
'age': 25,
'hobbies': ['读书', '游泳', '编程'],
'address': {
'city': '北京',
'district': '朝阳区'
}
}
# 序列化到文件
with open('data.pkl', 'wb') as f:
pickle.dump(data, f)
# 从文件反序列化
with open('data.pkl', 'rb') as f:
loaded_data = pickle.load(f)
print("加载的数据:", loaded_data)
# 序列化到字节串
serialized = pickle.dumps(data)
print("序列化后的长度:", len(serialized))
# 从字节串反序列化
deserialized = pickle.loads(serialized)
print("反序列化的数据:", deserialized)
json模块
import json
# Python对象转JSON
data = {
'name': '张三',
'age': 25,
'hobbies': ['读书', '游泳', '编程'],
'married': True,
'salary': None
}
# 转换为JSON字符串
json_string = json.dumps(data, ensure_ascii=False, indent=2)
print("JSON字符串:")
print(json_string)
# 保存到文件
with open('data.json', 'w', encoding='utf-8') as f:
json.dump(data, f, ensure_ascii=False, indent=2)
# 从JSON字符串解析
parsed_data = json.loads(json_string)
print("解析后的数据:", parsed_data)
# 从文件读取
with open('data.json', 'r', encoding='utf-8') as f:
file_data = json.load(f)
print("从文件读取的数据:", file_data)
# 自定义序列化
class Person:
def __init__(self, name, age):
self.name = name
self.age = age
def to_dict(self):
return {'name': self.name, 'age': self.age}
person = Person("李四", 30)
person_json = json.dumps(person.to_dict(), ensure_ascii=False)
print("自定义对象JSON:", person_json)
shelve模块
import shelve
# 使用shelve创建持久化字典
with shelve.open('shelf_data') as shelf:
# 存储数据
shelf['name'] = '张三'
shelf['age'] = 25
shelf['hobbies'] = ['读书', '游泳', '编程']
# 读取数据
print("姓名:", shelf['name'])
print("年龄:", shelf['age'])
print("爱好:", shelf['hobbies'])
# 遍历所有键
print("所有键:", list(shelf.keys()))
# shelve会自动创建三个文件:shelf_data.dat, shelf_data.bak, shelf_data.dir
数据压缩和归档
Python标准库提供了多种压缩和归档功能。
zlib模块
import zlib
# 数据压缩
data = b"This is some data that we want to compress. " * 10
print("原始数据长度:", len(data))
# 压缩数据
compressed = zlib.compress(data)
print("压缩后长度:", len(compressed))
print("压缩率:", f"{(1 - len(compressed)/len(data)) * 100:.2f}%")
# 解压缩数据
decompressed = zlib.decompress(compressed)
print("解压缩后长度:", len(decompressed))
print("数据一致吗?", data == decompressed)
# 使用不同的压缩级别
compressed_fast = zlib.compress(data, 1) # 快速压缩
compressed_best = zlib.compress(data, 9) # 最佳压缩
print("快速压缩长度:", len(compressed_fast))
print("最佳压缩长度:", len(compressed_best))
zipfile模块
import zipfile
import os
# 创建ZIP文件
with zipfile.ZipFile('example.zip', 'w', zipfile.ZIP_DEFLATED) as zipf:
# 添加文件到ZIP
zipf.write('data.json') # 添加现有文件
zipf.writestr('info.txt', '这是ZIP中的文本文件') # 直接添加文本
# 添加目录中的所有文件
if os.path.exists('utils'):
for root, dirs, files in os.walk('utils'):
for file in files:
file_path = os.path.join(root, file)
arc_path = os.path.relpath(file_path, '.')
zipf.write(file_path, arc_path)
# 读取ZIP文件
with zipfile.ZipFile('example.zip', 'r') as zipf:
# 列出ZIP中的文件
print("ZIP文件内容:")
for info in zipf.infolist():
print(f" {info.filename} ({info.file_size} bytes)")
# 读取特定文件
if 'info.txt' in zipf.namelist():
content = zipf.read('info.txt')
print("info.txt内容:", content.decode('utf-8'))
# 解压所有文件
# zipf.extractall('extracted/')
tarfile模块
import tarfile
# 创建TAR文件
with tarfile.open('example.tar.gz', 'w:gz') as tar:
# 添加文件
if os.path.exists('data.json'):
tar.add('data.json')
tar.add('info.txt')
# 读取TAR文件
with tarfile.open('example.tar.gz', 'r:gz') as tar:
# 列出TAR中的文件
print("TAR文件内容:")
for member in tar.getmembers():
print(f" {member.name} ({member.size} bytes)")
# 提取特定文件
# tar.extract('data.json', 'extracted/')
# 提取所有文件
# tar.extractall('extracted/')
加密服务
Python标准库提供了基本的加密功能。
hashlib模块
import hashlib
# MD5哈希
text = "Hello, World!"
md5_hash = hashlib.md5(text.encode('utf-8')).hexdigest()
print(f"MD5: {md5_hash}")
# SHA1哈希
sha1_hash = hashlib.sha1(text.encode('utf-8')).hexdigest()
print(f"SHA1: {sha1_hash}")
# SHA256哈希
sha256_hash = hashlib.sha256(text.encode('utf-8')).hexdigest()
print(f"SHA256: {sha256_hash}")
# 更新哈希对象
hasher = hashlib.sha256()
hasher.update(b"Hello, ")
hasher.update(b"World!")
final_hash = hasher.hexdigest()
print(f"分步哈希: {final_hash}")
# 文件哈希
def file_hash(filename, algorithm='sha256'):
"""计算文件的哈希值"""
hasher = hashlib.new(algorithm)
with open(filename, 'rb') as f:
for chunk in iter(lambda: f.read(4096), b""):
hasher.update(chunk)
return hasher.hexdigest()
# if os.path.exists('data.json'):
# file_hash_value = file_hash('data.json')
# print(f"文件哈希: {file_hash_value}")
secrets模块(推荐用于安全用途)
import secrets
# 生成安全的随机数
secure_random = secrets.randbelow(100)
print(f"安全随机数: {secure_random}")
# 生成随机选择
choices = ['red', 'green', 'blue', 'yellow']
secure_choice = secrets.choice(choices)
print(f"安全随机选择: {secure_choice}")
# 生成令牌
token = secrets.token_hex(16)
print(f"十六进制令牌: {token}")
url_token = secrets.token_urlsafe(16)
print(f"URL安全令牌: {url_token}")
# 生成密码
alphabet = "abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789!@#$%^&*"
password = ''.join(secrets.choice(alphabet) for _ in range(12))
print(f"生成密码: {password}")
通用操作系统服务
Python提供了与操作系统交互的模块。
os模块
import os
# 环境变量
print("当前工作目录:", os.getcwd())
print("用户主目录:", os.path.expanduser("~"))
print("PATH环境变量:", os.environ.get('PATH', '未设置')[:50] + "...")
# 进程管理
print("当前进程ID:", os.getpid())
print("当前用户ID:", os.getuid()) # Unix/Linux系统
print("当前用户名:", os.getlogin())
# 目录操作
# os.mkdir("new_directory") # 创建目录
# os.makedirs("parent/child/grandchild") # 创建多级目录
# os.chdir("new_directory") # 切换目录
# os.rmdir("new_directory") # 删除空目录
# 文件操作
# os.remove("file.txt") # 删除文件
# os.rename("old_name.txt", "new_name.txt") # 重命名文件
# 系统信息
print("操作系统:", os.name)
print("系统平台:", os.platform) # 需要import platform
platform模块
import platform
# 系统信息
print("系统:", platform.system())
print("节点名称:", platform.node())
print("发布版本:", platform.release())
print("版本:", platform.version())
print("机器:", platform.machine())
print("处理器:", platform.processor())
print("架构:", platform.architecture())
print("平台:", platform.platform())
# Python信息
print("Python实现:", platform.python_implementation())
print("Python版本:", platform.python_version())
print("Python编译器:", platform.python_compiler())
subprocess模块
import subprocess
# 执行系统命令
try:
# 简单命令执行
result = subprocess.run(['echo', 'Hello, World!'],
capture_output=True, text=True, check=True)
print("命令输出:", result.stdout.strip())
# 执行shell命令
result = subprocess.run('echo "Hello from shell"',
shell=True, capture_output=True, text=True)
print("Shell输出:", result.stdout.strip())
# 获取系统信息
result = subprocess.run(['python', '--version'],
capture_output=True, text=True)
print("Python版本:", result.stderr.strip())
except subprocess.CalledProcessError as e:
print(f"命令执行失败: {e}")
except FileNotFoundError:
print("命令未找到")
并发执行
Python提供了多种并发执行的机制。
threading模块
import threading
import time
# 线程函数
def worker(name, delay):
"""工作线程函数"""
for i in range(5):
print(f"线程 {name}: {i}")
time.sleep(delay)
print(f"线程 {name} 完成")
# 创建和启动线程
thread1 = threading.Thread(target=worker, args=("A", 1))
thread2 = threading.Thread(target=worker, args=("B", 1.5))
thread1.start()
thread2.start()
# 等待线程完成
thread1.join()
thread2.join()
print("所有线程完成")
multiprocessing模块
import multiprocessing
import time
# 进程函数
def worker(name, delay):
"""工作进程函数"""
for i in range(3):
print(f"进程 {name}: {i}")
time.sleep(delay)
print(f"进程 {name} 完成")
# 创建和启动进程
if __name__ == '__main__':
process1 = multiprocessing.Process(target=worker, args=("P1", 1))
process2 = multiprocessing.Process(target=worker, args=("P2", 1.5))
process1.start()
process2.start()
# 等待进程完成
process1.join()
process2.join()
print("所有进程完成")
concurrent.futures模块
from concurrent.futures import ThreadPoolExecutor, ProcessPoolExecutor
import time
# 工作函数
def compute_square(n):
"""计算平方"""
time.sleep(1) # 模拟耗时操作
return n * n
# 使用线程池
with ThreadPoolExecutor(max_workers=3) as executor:
# 提交任务
futures = [executor.submit(compute_square, i) for i in range(5)]
# 获取结果
results = [future.result() for future in futures]
print("线程池结果:", results)
# 使用进程池
if __name__ == '__main__':
with ProcessPoolExecutor(max_workers=3) as executor:
# 提交任务
futures = [executor.submit(compute_square, i) for i in range(5)]
# 获取结果
results = [future.result() for future in futures]
print("进程池结果:", results)
网络和互联网
Python标准库提供了丰富的网络编程支持。
urllib模块
from urllib import request, parse
import json
# 发送HTTP请求
try:
# GET请求
response = request.urlopen('https://httpbin.org/get')
data = response.read().decode('utf-8')
print("GET响应:", data[:100] + "...")
# POST请求
data = {'key1': 'value1', 'key2': 'value2'}
data_bytes = parse.urlencode(data).encode('utf-8')
req = request.Request('https://httpbin.org/post', data=data_bytes)
response = request.urlopen(req)
result = response.read().decode('utf-8')
print("POST响应:", result[:100] + "...")
except Exception as e:
print(f"请求失败: {e}")
socket模块
import socket
# 创建TCP客户端
def tcp_client():
"""TCP客户端示例"""
try:
# 创建socket对象
client_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
# 连接服务器
client_socket.connect(('httpbin.org', 80))
# 发送HTTP请求
request = b"GET /get HTTP/1.1\r\nHost: httpbin.org\r\n\r\n"
client_socket.send(request)
# 接收响应
response = client_socket.recv(1024)
print("服务器响应:", response.decode('utf-8')[:200] + "...")
# 关闭连接
client_socket.close()
except Exception as e:
print(f"客户端错误: {e}")
# 简单的TCP服务器(示例)
def tcp_server():
"""TCP服务器示例"""
try:
# 创建socket对象
server_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
# 绑定地址和端口
server_socket.bind(('localhost', 8888))
# 监听连接
server_socket.listen(5)
print("服务器启动,监听端口 8888...")
# 接受连接
client_socket, addr = server_socket.accept()
print(f"连接来自: {addr}")
# 接收数据
data = client_socket.recv(1024)
print("接收到数据:", data.decode('utf-8'))
# 发送响应
response = b"HTTP/1.1 200 OK\r\nContent-Type: text/plain\r\n\r\nHello, Client!"
client_socket.send(response)
# 关闭连接
client_socket.close()
server_socket.close()
except Exception as e:
print(f"服务器错误: {e}")
# tcp_client() # 取消注释以运行客户端示例
互联网数据处理
Python标准库支持多种互联网数据格式。
email模块
from email.mime.text import MIMEText
from email.mime.multipart import MIMEMultipart
import smtplib
# 创建邮件消息
def create_email():
"""创建邮件示例"""
# 创建多部分邮件
msg = MIMEMultipart()
msg['From'] = 'sender@example.com'
msg['To'] = 'receiver@example.com'
msg['Subject'] = 'Python邮件示例'
# 邮件正文
body = "这是使用Python发送的邮件示例。"
msg.attach(MIMEText(body, 'plain', 'utf-8'))
return msg
# 邮件消息示例
email_msg = create_email()
print("邮件内容:")
print(email_msg.as_string()[:200] + "...")
base64模块
import base64
# 编码和解码
text = "Hello, World!"
text_bytes = text.encode('utf-8')
# Base64编码
encoded = base64.b64encode(text_bytes)
print("Base64编码:", encoded.decode('ascii'))
# Base64解码
decoded = base64.b64decode(encoded)
print("Base64解码:", decoded.decode('utf-8'))
# URL安全的Base64编码
url_safe_encoded = base64.urlsafe_b64encode(text_bytes)
print("URL安全编码:", url_safe_encoded.decode('ascii'))
html模块
import html
# HTML转义和反转义
text = "Hello <script>alert('XSS')</script> World & Python"
escaped = html.escape(text)
print("转义后:", escaped)
unescaped = html.unescape(escaped)
print("反转义:", unescaped)
# 格式化HTML
from html import escape
def format_user_info(name, email):
"""格式化用户信息为HTML"""
return f"<div><strong>姓名:</strong> {escape(name)}<br><strong>邮箱:</strong> {escape(email)}</div>"
user_html = format_user_info("张三", "zhangsan@example.com")
print("用户信息HTML:", user_html)
结构化标记处理工具
Python标准库提供了处理结构化标记的工具。
xml模块
import xml.etree.ElementTree as ET
# 创建XML
def create_xml():
"""创建XML示例"""
# 创建根元素
root = ET.Element("students")
# 添加学生元素
student1 = ET.SubElement(root, "student", id="1")
name1 = ET.SubElement(student1, "name")
name1.text = "张三"
age1 = ET.SubElement(student1, "age")
age1.text = "20"
student2 = ET.SubElement(root, "student", id="2")
name2 = ET.SubElement(student2, "name")
name2.text = "李四"
age2 = ET.SubElement(student2, "age")
age2.text = "21"
# 创建ElementTree对象
tree = ET.ElementTree(root)
return tree
# 保存XML到文件
xml_tree = create_xml()
xml_tree.write("students.xml", encoding="utf-8", xml_declaration=True)
# 读取和解析XML
try:
tree = ET.parse("students.xml")
root = tree.getroot()
print("XML内容:")
for student in root.findall("student"):
student_id = student.get("id")
name = student.find("name").text
age = student.find("age").text
print(f" 学生ID: {student_id}, 姓名: {name}, 年龄: {age}")
except FileNotFoundError:
print("XML文件未找到")
except ET.ParseError as e:
print(f"XML解析错误: {e}")
csv模块
import csv
# 写入CSV文件
data = [
['姓名', '年龄', '城市'],
['张三', '25', '北京'],
['李四', '30', '上海'],
['王五', '35', '广州']
]
with open('users.csv', 'w', newline='', encoding='utf-8') as csvfile:
writer = csv.writer(csvfile)
writer.writerows(data)
# 读取CSV文件
with open('users.csv', 'r', encoding='utf-8') as csvfile:
reader = csv.reader(csvfile)
print("CSV内容:")
for row in reader:
print(" ", row)
# 使用字典读写CSV
with open('users_dict.csv', 'w', newline='', encoding='utf-8') as csvfile:
fieldnames = ['姓名', '年龄', '城市']
writer = csv.DictWriter(csvfile, fieldnames=fieldnames)
writer.writeheader()
writer.writerow({'姓名': '张三', '年龄': '25', '城市': '北京'})
writer.writerow({'姓名': '李四', '年龄': '30', '城市': '上海'})
# 读取字典格式的CSV
with open('users_dict.csv', 'r', encoding='utf-8') as csvfile:
reader = csv.DictReader(csvfile)
print("\n字典格式CSV内容:")
for row in reader:
print(" ", row)
实际应用示例
# 综合应用示例:文件备份工具
import os
import shutil
import hashlib
import json
from datetime import datetime
from pathlib import Path
class BackupManager:
"""备份管理器"""
def __init__(self, backup_dir="backups"):
self.backup_dir = Path(backup_dir)
self.backup_dir.mkdir(exist_ok=True)
self.backup_info_file = self.backup_dir / "backup_info.json"
self.load_backup_info()
def load_backup_info(self):
"""加载备份信息"""
if self.backup_info_file.exists():
with open(self.backup_info_file, 'r', encoding='utf-8') as f:
self.backup_info = json.load(f)
else:
self.backup_info = {}
def save_backup_info(self):
"""保存备份信息"""
with open(self.backup_info_file, 'w', encoding='utf-8') as f:
json.dump(self.backup_info, f, ensure_ascii=False, indent=2)
def calculate_file_hash(self, filepath):
"""计算文件哈希值"""
hasher = hashlib.md5()
with open(filepath, 'rb') as f:
for chunk in iter(lambda: f.read(4096), b""):
hasher.update(chunk)
return hasher.hexdigest()
def backup_file(self, source_file):
"""备份单个文件"""
source_path = Path(source_file)
if not source_path.exists():
print(f"源文件不存在: {source_file}")
return False
# 计算文件哈希
file_hash = self.calculate_file_hash(source_path)
# 检查是否已备份
if str(source_path) in self.backup_info:
if self.backup_info[str(source_path)]['hash'] == file_hash:
print(f"文件未修改,跳过备份: {source_file}")
return True
# 创建备份文件名
timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
backup_filename = f"{source_path.stem}_{timestamp}{source_path.suffix}"
backup_path = self.backup_dir / backup_filename
# 复制文件
try:
shutil.copy2(source_path, backup_path)
print(f"备份成功: {source_file} -> {backup_path}")
# 更新备份信息
self.backup_info[str(source_path)] = {
'backup_path': str(backup_path),
'hash': file_hash,
'backup_time': timestamp
}
self.save_backup_info()
return True
except Exception as e:
print(f"备份失败: {e}")
return False
def backup_directory(self, source_dir):
"""备份目录"""
source_path = Path(source_dir)
if not source_path.exists() or not source_path.is_dir():
print(f"源目录不存在或不是目录: {source_dir}")
return
print(f"开始备份目录: {source_dir}")
for file_path in source_path.rglob("*"):
if file_path.is_file():
self.backup_file(file_path)
print("目录备份完成")
def list_backups(self):
"""列出所有备份"""
print("备份信息:")
for source, info in self.backup_info.items():
print(f" 源文件: {source}")
print(f" 备份文件: {info['backup_path']}")
print(f" 备份时间: {info['backup_time']}")
print()
# 配置文件管理器
class ConfigManager:
"""配置文件管理器"""
def __init__(self, config_file="config.json"):
self.config_file = Path(config_file)
self.config = self.load_config()
def load_config(self):
"""加载配置"""
if self.config_file.exists():
try:
with open(self.config_file, 'r', encoding='utf-8') as f:
return json.load(f)
except json.JSONDecodeError:
print("配置文件格式错误,使用默认配置")
return self.get_default_config()
def save_config(self):
"""保存配置"""
with open(self.config_file, 'w', encoding='utf-8') as f:
json.dump(self.config, f, ensure_ascii=False, indent=2)
def get(self, key, default=None):
"""获取配置值"""
return self.config.get(key, default)
def set(self, key, value):
"""设置配置值"""
self.config[key] = value
self.save_config()
def get_default_config(self):
"""获取默认配置"""
return {
"backup_dirs": ["./documents", "./projects"],
"backup_interval": 86400, # 24小时(秒)
"max_backups": 10,
"exclude_patterns": [".tmp", ".log"]
}
# 主程序
def main():
"""主程序"""
print("=== Python标准库综合应用示例 ===\n")
# 配置管理
config = ConfigManager()
print("当前配置:")
for key, value in config.config.items():
print(f" {key}: {value}")
# 文件备份
backup_manager = BackupManager()
# 创建示例文件
sample_dir = Path("sample_files")
sample_dir.mkdir(exist_ok=True)
# 创建示例文件
sample_files = [
("test1.txt", "这是测试文件1的内容"),
("test2.txt", "这是测试文件2的内容"),
("data.json", '{"name": "张三", "age": 25}')
]
for filename, content in sample_files:
with open(sample_dir / filename, 'w', encoding='utf-8') as f:
f.write(content)
# 备份示例文件
backup_manager.backup_directory(sample_dir)
# 显示备份信息
backup_manager.list_backups()
# 清理示例文件
shutil.rmtree(sample_dir, ignore_errors=True)
if __name__ == "__main__":
main()
总结
本篇教程详细介绍了Python标准库的各个方面,包括文本处理、数据结构、文件操作、数据持久化、压缩归档、加密服务、操作系统服务、并发执行、网络编程等模块。
Python标准库是Python强大功能的重要组成部分,掌握这些模块的使用对于提高编程效率和代码质量至关重要。标准库提供了大量经过优化和测试的功能,可以直接使用而无需安装第三方库。
在下一章中,我们将学习虚拟环境和包管理,了解如何管理Python项目的依赖关系。
更多推荐
所有评论(0)