第12篇:Python标准库

目录

Python标准库概述

Python标准库是Python发行版中包含的一组模块和包,提供了丰富的功能,涵盖了从文本处理到网络编程的各种需求。标准库遵循"电池已包含"的哲学,使得Python能够处理各种常见的编程任务而无需安装额外的第三方库。

标准库的特点

  1. 丰富性:涵盖了各种常见编程需求
  2. 一致性:遵循统一的设计原则和接口规范
  3. 可靠性:经过充分测试,稳定可靠
  4. 跨平台:在不同操作系统上保持一致的行为
  5. 文档完善:有详细的官方文档支持

标准库的组织结构

# 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项目的依赖关系。


Logo

Agent 垂直技术社区,欢迎活跃、内容共建。

更多推荐