Python生态系统提供了丰富的文件操作库，可以处理各种复杂的文件操作需求。本教程将介绍Python中最常用的文件操作库及其实际应用。

一、标准库核心模块

1.1 os模块 - 操作系统接口

主要功能：

• 文件和目录操作
• 路径操作
• 环境变量访问
• 进程管理

常用函数示例：

import os

# 文件操作
os.rename('old.txt', 'new.txt')  # 重命名
os.remove('file.txt')            # 删除文件
os.chmod('script.py', 0o755)     # 修改权限

# 目录操作
os.mkdir('new_dir')              # 创建目录
os.rmdir('empty_dir')            # 删除空目录
os.listdir('.')                  # 列出目录内容

# 路径操作
os.path.abspath('file.txt')      # 获取绝对路径
os.path.join('dir', 'file.txt')  # 路径拼接
os.path.getsize('file.txt')      # 获取文件大小

表1：os模块常用函数

1.2 shutil模块 - 高级文件操作

主要功能：

• 文件和目录的复制、移动
• 归档操作
• 磁盘空间管理

实用示例：

import shutil

# 文件复制
shutil.copy('src.txt', 'dst.txt')          # 复制文件内容
shutil.copy2('src.txt', 'dst.txt')         # 复制文件内容和元数据

# 目录复制
shutil.copytree('src_dir', 'dst_dir')      # 递归复制目录

# 文件/目录移动
shutil.move('src', 'dst')                  # 移动文件或目录

# 磁盘空间
total, used, free = shutil.disk_usage('.')  # 获取磁盘使用情况

shutil文件操作流程

二、文件格式处理库

2.1 csv模块 - CSV文件处理

读写CSV文件示例：

import csv

# 写入CSV文件
with open('data.csv', 'w', newline='') as f:
    writer = csv.writer(f)
    writer.writerow(['Name', 'Age', 'City'])
    writer.writerow(['Alice', 25, 'New York'])
    writer.writerows([
        ['Bob', 30, 'Chicago'],
        ['Charlie', 35, 'Los Angeles']
    ])

# 读取CSV文件
with open('data.csv', newline='') as f:
    reader = csv.reader(f)
    for row in reader:
        print(row)

# 字典方式读写
with open('data.csv', 'w', newline='') as f:
    fieldnames = ['Name', 'Age', 'City']
    writer = csv.DictWriter(f, fieldnames=fieldnames)
    writer.writeheader()
    writer.writerow({'Name': 'Alice', 'Age': 25, 'City': 'New York'})

CSV方言定制：

csv.register_dialect('unix', 
                    delimiter=',', 
                    quoting=csv.QUOTE_MINIMAL,
                    lineterminator='\n')

with open('data.csv', 'w', newline='') as f:
    writer = csv.writer(f, dialect='unix')
    # 使用自定义方言写入

2.2 json模块 - JSON数据处理

JSON文件操作示例：

import json

data = {
    "name": "Alice",
    "age": 25,
    "skills": ["Python", "Java"],
    "is_active": True
}

# 写入JSON文件
with open('data.json', 'w') as f:
    json.dump(data, f, indent=2)  # indent参数美化输出

# 读取JSON文件
with open('data.json') as f:
    loaded_data = json.load(f)
    print(loaded_data['name'])

# JSON字符串转换
json_str = json.dumps(data, sort_keys=True)
parsed_data = json.loads(json_str)

表2：json模块参数说明

三、高级文件处理库

3.1 pathlib模块 - 面向对象路径操作

Path对象使用示例：

from pathlib import Path

# 创建Path对象
p = Path('dir/subdir/file.txt')

# 常用操作
p.exists()      # 检查路径是否存在
p.is_file()     # 是否是文件
p.is_dir()      # 是否是目录
p.stat()        # 获取文件信息
p.rename('new.txt')  # 重命名

# 路径组合
new_p = p.parent / 'new_dir' / 'new_file.txt'

# 文件读写
content = p.read_text(encoding='utf-8')
p.write_text('Hello', encoding='utf-8')

# 通配符查找
for py_file in Path('.').glob('*.py'):
    print(py_file)

# 递归查找
for py_file in Path('.').rglob('*.py'):
    print(py_file)

实用案例：日志文件清理：

def clean_old_logs(log_dir, days=30):
    """删除指定天数前的日志文件"""
    from datetime import datetime, timedelta
    cutoff = datetime.now() - timedelta(days=days)
    
    for log_file in Path(log_dir).glob('*.log'):
        mtime = datetime.fromtimestamp(log_file.stat().st_mtime)
        if mtime < cutoff:
            log_file.unlink()
            print(f"Deleted: {log_file}")

3.2 tempfile模块 - 临时文件处理

临时文件操作示例：

import tempfile

# 创建临时文件
with tempfile.NamedTemporaryFile(mode='w+', suffix='.tmp', delete=False) as tmp:
    tmp.write("临时数据")
    tmp_path = tmp.name  # 获取临时文件路径
    print(f"临时文件: {tmp_path}")

# 创建临时目录
with tempfile.TemporaryDirectory() as tmpdir:
    print(f"临时目录: {tmpdir}")
    # 在临时目录中工作
    temp_file = Path(tmpdir) / 'temp.txt'
    temp_file.write_text("临时数据")
    
# 退出with块后自动清理

安全临时文件模式：

def secure_temp_file(data):
    """安全创建临时文件并处理"""
    try:
        with tempfile.NamedTemporaryFile(
            mode='w',
            prefix='secure_',
            suffix='.tmp',
            dir='/secure/tmp',  # 指定安全目录
            delete=True
        ) as tmp:
            tmp.write(data)
            tmp.flush()  # 确保数据写入磁盘
            # 处理临时文件
            process_temp_file(tmp.name)
    except OSError as e:
        print(f"临时文件错误: {e}")

四、压缩文件处理

4.1 zipfile模块 - ZIP归档处理

ZIP文件操作示例：

import zipfile

# 创建ZIP文件
with zipfile.ZipFile('archive.zip', 'w') as zipf:
    zipf.write('file1.txt', arcname='data/file1.txt')  # 指定归档内路径
    zipf.write('file2.txt')

# 读取ZIP文件
with zipfile.ZipFile('archive.zip', 'r') as zipf:
    print(zipf.namelist())  # 列出归档内容
    zipf.extract('data/file1.txt', 'extracted')  # 提取特定文件
    zipf.extractall('extracted_all')  # 提取全部

# 密码保护的ZIP
with zipfile.ZipFile('secure.zip', 'w') as zipf:
    zipf.setpassword(b'secret')
    zipf.write('secret.txt')

# 检查ZIP文件有效性
if zipfile.is_zipfile('archive.zip'):
    print("有效的ZIP文件")

4.2 gzip/tarfile模块 - 压缩与归档

gzip压缩示例：

import gzip
import shutil

# 压缩文件
with open('large_file.txt', 'rb') as f_in:
    with gzip.open('large_file.txt.gz', 'wb') as f_out:
        shutil.copyfileobj(f_in, f_out)

# 解压文件
with gzip.open('large_file.txt.gz', 'rb') as f_in:
    with open('large_file.txt', 'wb') as f_out:
        shutil.copyfileobj(f_in, f_out)

tar归档示例：

import tarfile

# 创建tar归档
with tarfile.open('archive.tar.gz', 'w:gz') as tar:
    tar.add('dir_to_archive')  # 添加整个目录

# 读取tar归档
with tarfile.open('archive.tar.gz', 'r:gz') as tar:
    tar.extractall('extracted_dir')  # 解压全部
    # 或者逐个处理
    for member in tar.getmembers():
        if member.name.endswith('.txt'):
            tar.extract(member, 'extracted_txts')

五、特殊文件格式处理

5.1 configparser - 配置文件解析

INI文件处理示例：

from configparser import ConfigParser

config = ConfigParser()

# 读取INI文件
config.read('config.ini')

# 访问配置
db_host = config['Database']['host']
db_port = config.getint('Database', 'port')

# 修改配置
config['Database']['host'] = 'new.host.com'
with open('config.ini', 'w') as f:
    config.write(f)

# 创建新配置
config['Logging'] = {
    'level': 'DEBUG',
    'file': '/var/log/app.log'
}

5.2 pickle - Python对象序列化

对象序列化示例：

import pickle

data = {
    'name': 'Alice',
    'scores': [85, 92, 88],
    'active': True
}

# 序列化到文件
with open('data.pkl', 'wb') as f:
    pickle.dump(data, f)

# 从文件反序列化
with open('data.pkl', 'rb') as f:
    loaded_data = pickle.load(f)

# 安全提醒：不要反序列化不受信任的来源

安全替代方案：

# 对于简单数据，可以考虑使用json
import json

with open('data.json', 'w') as f:
    json.dump(data, f)

with open('data.json') as f:
    loaded_data = json.load(f)

六、应用案例

6.1 文件监控工具

import time
from pathlib import Path
import hashlib

def file_monitor(path, interval=5):
    """监控文件变化"""
    path = Path(path)
    if not path.exists():
        raise FileNotFoundError(f"文件不存在: {path}")
    
    last_mtime = path.stat().st_mtime
    last_hash = file_hash(path)
    
    while True:
        try:
            current_mtime = path.stat().st_mtime
            if current_mtime != last_mtime:
                current_hash = file_hash(path)
                if current_hash != last_hash:
                    print(f"文件 {path} 内容已更改")
                    last_hash = current_hash
                last_mtime = current_mtime
            time.sleep(interval)
        except KeyboardInterrupt:
            print("监控停止")
            break

def file_hash(filepath):
    """计算文件哈希"""
    hash_md5 = hashlib.md5()
    with open(filepath, "rb") as f:
        for chunk in iter(lambda: f.read(4096), b""):
            hash_md5.update(chunk)
    return hash_md5.hexdigest()

6.2 目录同步工具

import filecmp
import shutil
from pathlib import Path

def sync_dirs(src, dst):
    """同步两个目录"""
    src, dst = Path(src), Path(dst)
    comparison = filecmp.dircmp(src, dst)
    
    # 复制左侧独有的文件
    for item in comparison.left_only:
        src_path = src / item
        if src_path.is_file():
            shutil.copy2(src_path, dst)
        else:
            shutil.copytree(src_path, dst / item)
    
    # 复制更新的文件
    for item in comparison.diff_files:
        shutil.copy2(src / item, dst / item)
    
    # 递归处理子目录
    for subdir in comparison.common_dirs:
        sync_dirs(src / subdir, dst / subdir)

6.3 文件搜索工具

from pathlib import Path
import re

def find_files(pattern, root='.', recursive=True):
    """查找匹配模式的文件"""
    root = Path(root)
    if recursive:
        gen = root.rglob('*')
    else:
        gen = root.glob('*')
    
    regex = re.compile(pattern)
    for item in gen:
        if item.is_file() and regex.search(item.name):
            yield item

# 使用示例
for file in find_files(r'\.(py|txt)#39;, 'src'):
    print(file)

七、其他

1. 批量操作：减少IO操作次数，批量读写数据
2. 缓冲区调整：对大文件调整缓冲区大小
3. 内存映射：对大文件使用mmap
4. 并行处理：对大量文件使用多线程/多进程
5. 资源清理：确保文件描述符正确关闭

高效文件复制示例：

def efficient_copy(src, dst, buffer_size=16*1024):
    """高效文件复制函数"""
    with open(src, 'rb') as f_src:
        with open(dst, 'wb') as f_dst:
            while True:
                chunk = f_src.read(buffer_size)
                if not chunk:
                    break
                f_dst.write(chunk)

内存映射示例：

import mmap

def mmap_example(filepath):
    """使用内存映射处理大文件"""
    with open(filepath, 'r+') as f:
        with mmap.mmap(f.fileno(), 0) as mm:
            # 像操作字符串一样操作文件
            if b'search_term' in mm:
                pos = mm.find(b'search_term')
                mm[pos:pos+12] = b'replacement'

八、总结

Python提供了丰富的文件操作库，可以满足从基础到高级的各种需求：

1. 标准库基础：os, io, shutil提供核心功能
2. 路径处理：pathlib提供现代化面向对象接口
3. 文件格式：csv, json, configparser处理特定格式
4. 压缩归档：zipfile, tarfile, gzip处理压缩文件
5. 高级功能：tempfile, mmap提供特殊场景支持

选择库时应考虑：

• 项目需求（基础操作还是特定格式）
• Python版本兼容性
• 性能要求
• 代码可维护性

掌握这些库的使用方法，可以让我们在Python中高效地处理各种文件操作任务。

持续更新Python编程学习日志与技巧，敬请关注！

#编程# #python# #在头条记录我的2025# #夏日生活打卡季#