Day35 - 文件处理综合 (CSV/shutil/os/pathlib)

详细讲解

1. CSV 文件处理

1.1 CSV 模块基础

import csv

# 准备数据
data = [
    ['姓名', '年龄', '城市'],
    ['张三', '25', '北京'],
    ['李四', '30', '上海'],
    ['王五', '28', '广州']
]

# 写入 CSV 文件
with open('output.csv', 'w', newline='', encoding='utf-8') as f:
    writer = csv.writer(f)
    writer.writerows(data)  # 写入多行

# 读取 CSV 文件
with open('output.csv', 'r', encoding='utf-8') as f:
    reader = csv.reader(f)
    for row in reader:
        print(row)

注意：newline='' 在写入 CSV 时必须使用，否则在 Windows 上可能导致行距加倍。

1.2 字典方式操作 CSV

import csv

# 写入字典格式
fieldnames = ['name', 'age', 'city']
data = [
    {'name': '张三', 'age': 25, 'city': '北京'},
    {'name': '李四', 'age': 30, 'city': '上海'}
]

with open('dict_data.csv', 'w', newline='', encoding='utf-8') as f:
    writer = csv.DictWriter(f, fieldnames=fieldnames)
    writer.writeheader()  # 写入表头
    writer.writerows(data)

# 读取字典格式
with open('dict_data.csv', 'r', encoding='utf-8') as f:
    reader = csv.DictReader(f)
    for row in reader:
        print(f"{row['name']} - {row['age']} - {row['city']}")

1.3 CSV 高级选项

import csv

# 自定义分隔符
with open('tab_data.tsv', 'w', newline='', encoding='utf-8') as f:
    writer = csv.writer(f, delimiter='\t')
    writer.writerows(data)

# 引用处理
with open('quoted.csv', 'w', newline='', encoding='utf-8') as f:
    writer = csv.writer(f, quoting=csv.QUOTE_ALL)  # 所有字段加引号
    writer.writerows(data)

# quoting 选项：
# QUOTE_ALL      - 所有字段加引号
# QUOTE_MINIMAL  - 特殊字符字段加引号（默认）
# QUOTE_NONNUMERIC - 非数字字段加引号
# QUOTE_NONE     - 不加引号

1.4 处理大 CSV 文件

import csv

# 使用生成器处理大文件
def read_csv_chunked(filepath, chunk_size=1000):
    """分块读取大 CSV 文件"""
    with open(filepath, 'r', encoding='utf-8') as f:
        reader = csv.reader(f)
        header = next(reader)  # 读取表头
        
        chunk = []
        for row in reader:
            chunk.append(row)
            if len(chunk) >= chunk_size:
                yield header, chunk
                chunk = []
        
        if chunk:  # 处理最后一块
            yield header, chunk

# 使用
for header, chunk in read_csv_chunked('large_file.csv'):
    print(f"处理 {len(chunk)} 条记录")
    # 处理每块数据

2. shutil 模块 - 高级文件操作

2.1 复制操作

import shutil

# 复制文件
shutil.copy('source.txt', 'dest.txt')      # 复制文件内容
shutil.copy2('source.txt', 'dest.txt')    # 复制文件并保留元数据

# 复制目录
shutil.copytree('source_dir', 'dest_dir')  # 递归复制整个目录

# 复制到文件夹
shutil.copy('file.txt', 'target_dir/')     # 目标可以是目录

2.2 移动和重命名

import shutil

# 移动文件
shutil.move('source.txt', 'dest.txt')

# 移动目录
shutil.move('source_dir', 'dest_dir')

# 重命名（同一目录下移动即重命名）
shutil.move('old_name.txt', 'new_name.txt')

2.3 删除操作

import shutil

# 删除目录（必须是空目录）
import os
os.rmdir('empty_dir')

# 删除目录树（递归删除）
shutil.rmtree('dir_to_remove')

# 安全删除（移动到回收站，可安装 send2trash 库）
# pip install send2trash
try:
    import send2trash
    send2trash.send2trash('file_to_delete.txt')
except ImportError:
    print("请安装: pip install send2trash")

2.4 归档和压缩

import shutil

# 创建归档（zip, tar, gztar, bztar, xztar）
shutil.make_archive('backup', 'zip', 'directory_to_backup')

# 解压归档
shutil.unpack_archive('backup.zip', 'extract_here')

# 列出支持的格式
print(shutil.get_archive_formats())
print(shutil.get_unpack_formats())

2.5 其他有用功能

import shutil

# 获取磁盘使用情况
total, used, free = shutil.disk_usage('/')
print(f"总容量: {total // (2**30)} GB")
print(f"已使用: {used // (2**30)} GB")
print(f"剩余: {free // (2**30)} GB")

# 复制权限（仅 Unix）
# shutil.copystat(src, dst)

# 获取临时目录
print(shutil.gettempdir())

# 命令行工具包装
# shutil.which('python')  # 查找命令路径

3. os 模块 - 系统交互

3.1 路径操作

import os

# 获取当前工作目录
print(os.getcwd())

# 改变当前目录
os.chdir('/path/to/directory')

# 返回绝对路径
os.path.abspath('relative/path')

# 判断路径类型
os.path.isfile('file.txt')     # 是否为文件
os.path.isdir('directory')     # 是否为目录
os.path.islink('link')         # 是否为符号链接
os.path.exists('path')         # 路径是否存在

# 路径连接
path = os.path.join('dir', 'subdir', 'file.txt')
# 输出: dir/subdir/file.txt (Unix) 或 dir\subdir\file.txt (Windows)

# 分割路径
dirname, basename = os.path.split('/path/to/file.txt')
# dirname: /path/to, basename: file.txt

# 分割扩展名
name, ext = os.path.splitext('file.txt')
# name: file, ext: .txt

3.2 目录操作

import os

# 创建目录
os.mkdir('new_dir')           # 创建单级目录（父目录必须存在）
os.makedirs('path/to/new/dir')  # 递归创建目录

# 列出目录内容
os.listdir('.')               # 返回列表

# 遍历目录树
for dirpath, dirnames, filenames in os.walk('.'):
    print(f"目录: {dirpath}")
    print(f"子目录: {dirnames}")
    print(f"文件: {filenames}")

# 删除
os.remove('file.txt')         # 删除文件
os.rmdir('empty_dir')         # 删除空目录

3.3 文件信息

import os
import datetime

# 获取文件大小（字节）
size = os.path.getsize('file.txt')

# 获取文件修改时间
mtime = os.path.getmtime('file.txt')
print(datetime.datetime.fromtimestamp(mtime))

# 获取创建时间
ctime = os.path.getctime('file.txt')

# 获取文件详细信息
stat_info = os.stat('file.txt')
print(f"大小: {stat_info.st_size} 字节")
print(f"修改: {datetime.datetime.fromtimestamp(stat_info.st_mtime)}")

3.4 环境变量

import os

# 获取环境变量
home = os.environ.get('HOME')           # Unix
# home = os.environ.get('USERPROFILE') # Windows
path = os.environ.get('PATH', '')

# 设置环境变量
os.environ['MY_VAR'] = 'value'

# 获取所有环境变量
for key, value in os.environ.items():
    print(f"{key}={value}")

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

Python 3.4+ 推荐使用 pathlib，比 os.path 更直观。

4.1 基本使用

from pathlib import Path

# 当前目录
p = Path('.')
print(p.absolute())

# 路径拼接
p = Path('dir') / 'subdir' / 'file.txt'
print(p)

# 获取路径各部分
p = Path('/home/user/documents/file.txt')
print(p.name)      # file.txt
print(p.stem)      # file
print(p.suffix)    # .txt
print(p.parent)    # /home/user/documents
print(p.parents)   # 父路径迭代器

# 路径判断
p = Path('file.txt')
print(p.is_file())    # True
print(p.is_dir())     # False
print(p.exists())     # True

4.2 目录操作

from pathlib import Path

# 创建目录
Path('new_dir').mkdir()              # 创建单级目录
Path('a/b/c').mkdir(parents=True)   # 递归创建

# 创建目录（如果不存在）
Path('dir').mkdir(exist_ok=True)

# 删除空目录
Path('empty_dir').rmdir()

# 列出目录内容
p = Path('.')
for item in p.iterdir():  # 只列出直接子项
    print(item.name)

# glob 模式匹配
for py_file in p.glob('*.py'):  # 所有 .py 文件
    print(py_file)

for py_file in p.rglob('*.py'):  # 递归所有 .py 文件（包括子目录）
    print(py_file)

4.3 文件读写

from pathlib import Path

# 读取文本
content = Path('file.txt').read_text(encoding='utf-8')

# 写入文本
Path('output.txt').write_text('Hello, World!', encoding='utf-8')

# 读取字节
content_bytes = Path('file.txt').read_bytes()

# 写入字节
Path('output.bin').write_bytes(b'\x00\x01\x02')

# 追加写入（需要用 open）
p = Path('log.txt')
p.write_text('新内容\n', encoding='utf-8')  # 这会覆盖

# 正确追加方式
with p.open('a', encoding='utf-8') as f:
    f.write('追加内容\n')

4.4 Path 与 os.path 互转

from pathlib import Path

# Path 转字符串
str_path = str(Path('dir/file.txt'))
print(str_path)  # dir/file.txt

# os.path 函数中使用 Path
import os
p = Path('file.txt')
print(os.path.exists(p))  # Path 对象可直接用于 os.path 函数

# 获取路径字符串用于 subprocess 等
import subprocess
subprocess.run(['ls', '-la', str(Path('.'))])

5. 实战案例：文件处理工具

from pathlib import Path
import csv
import shutil
from datetime import datetime

class FileManager:
    """文件管理工具类"""
    
    def __init__(self, base_dir='.'):
        self.base_dir = Path(base_dir)
    
    def create_directory(self, dir_name):
        """创建目录"""
        target = self.base_dir / dir_name
        target.mkdir(parents=True, exist_ok=True)
        print(f"已创建目录: {target}")
        return target
    
    def list_files(self, pattern='*'):
        """列出文件"""
        return list(self.base_dir.glob(pattern))
    
    def find_large_files(self, min_size_mb=10):
        """查找大于指定大小的文件"""
        large_files = []
        for file in self.base_dir.rglob('*'):
            if file.is_file():
                size_mb = file.stat().st_size / (1024 * 1024)
                if size_mb >= min_size_mb:
                    large_files.append((file, size_mb))
        return sorted(large_files, key=lambda x: x[1], reverse=True)
    
    def backup_file(self, filename, backup_dir='backups'):
        """备份文件"""
        src = self.base_dir / filename
        if not src.exists():
            raise FileNotFoundError(f"文件不存在: {src}")
        
        # 创建备份目录
        backup_path = self.base_dir / backup_dir
        backup_path.mkdir(exist_ok=True)
        
        # 生成带时间戳的备份文件名
        timestamp = datetime.now().strftime('%Y%m%d_%H%M%S')
        backup_name = f"{src.stem}_{timestamp}{src.suffix}"
        dst = backup_path / backup_name
        
        shutil.copy2(src, dst)
        print(f"已备份到: {dst}")
        return dst
    
    def read_csv_as_dict(self, csv_file):
        """读取 CSV 为字典列表"""
        with open(csv_file, 'r', encoding='utf-8') as f:
            reader = csv.DictReader(f)
            return list(reader)
    
    def write_dict_to_csv(self, data, csv_file, fieldnames=None):
        """写入字典列表到 CSV"""
        if not data:
            return
        
        if fieldnames is None:
            fieldnames = list(data[0].keys())
        
        with open(csv_file, 'w', newline='', encoding='utf-8') as f:
            writer = csv.DictWriter(f, fieldnames=fieldnames)
            writer.writeheader()
            writer.writerows(data)
        print(f"已写入 CSV: {csv_file}")
    
    def clean_old_files(self, days=30):
        """清理指定天数前的文件"""
        from datetime import timedelta
        cutoff = datetime.now() - timedelta(days=days)
        removed = []
        
        for file in self.base_dir.rglob('*'):
            if file.is_file():
                mtime = datetime.fromtimestamp(file.stat().st_mtime)
                if mtime < cutoff:
                    file.unlink()
                    removed.append(file)
        
        print(f"已清理 {len(removed)} 个文件")
        return removed
    
    def get_disk_usage(self):
        """获取磁盘使用情况"""
        import shutil as sh
        usage = sh.disk_usage(self.base_dir)
        return {
            'total_gb': usage.total / (1024**3),
            'used_gb': usage.used / (1024**3),
            'free_gb': usage.free / (1024**3),
            'percent': (usage.used / usage.total) * 100
        }


# 使用示例
if __name__ == '__main__':
    fm = FileManager('/tmp/test')
    
    # 创建测试目录
    fm.create_directory('data')
    
    # 创建测试文件
    (fm.base_dir / 'data' / 'test.txt').write_text('测试内容')
    
    # 列出所有文件
    print("\n所有文件:")
    for f in fm.list_files('**/*'):
        print(f"  {f}")
    
    # 读取磁盘使用情况
    usage = fm.get_disk_usage()
    print(f"\n磁盘使用: {usage['percent']:.1f}%")

背诵版

模块速查

┌─────────────────────────────────────────────────────────────┐ │ 文件处理模块对比 │ ├─────────────────────────────────────────────────────────────┤ │ os │ 底层系统调用，传统 API │ │ shutil │ 高级文件操作（复制/移动/压缩） │ │ pathlib │ 面向对象路径操作（Python 3.4+，推荐） │ │ csv │ CSV 文件读写 │ └─────────────────────────────────────────────────────────────┘

pathlib Path 常用方法

方法 说明
Path.mkdir() 创建目录
Path.unlink() 删除文件
Path.rename() 重命名
Path.glob() 模式匹配
Path.iterdir() 遍历子项
Path.read_text() 读取文本
Path.write_text() 写入文本
Path.is_file() 是否为文件
Path.is_dir() 是否为目录
Path.exists() 是否存在
Path.stat() 获取信息
Path.resolve() 转绝对路径

shutil 常用函数

函数 说明
shutil.copy() 复制文件
shutil.copy2() 复制保留元数据
shutil.copytree() 复制目录树
shutil.move() 移动/重命名
shutil.rmtree() 删除目录树
shutil.make_archive() 创建归档
shutil.unpack_archive() 解压归档
shutil.disk_usage() 磁盘使用情况

考前记忆

面试重点

  1. pathlib vs os.path

    • pathlib 面向对象，API 更直观
    • Python 3.4+ 推荐使用 pathlib
    • Path 对象可与 os.path 函数混用

  2. shutil.rmtree() vs os.remove()

    • rmtree() 删除目录树
    • remove() 删除单个文件

  3. CSV 读取注意

    • newline='' 避免 Windows 换行问题
    • DictReader/DictWriter 更方便

  4. 路径拼接

    • 使用 / 运算符：Path('dir') / 'file.txt'
    • 避免使用字符串拼接

  5. glob vs rglob

    • glob() 只匹配当前目录
    • rglob() 递归匹配

记忆口诀

pathlib面向对象， 路径操作最轻松。 shutil管复制移动， csv处理表格中。 os底层系统调， 三者配合显神通。

测试题

选择题

1. 哪个模块是 Python 3.4+ 推荐使用的路径操作方式？

# A. os.path
# B. pathlib
# C. shutil
# D. glob

答案：B

2. 如何递归列出目录下所有 .txt 文件？

# A.
for f in Path('.').glob('*.txt'):
    print(f)

# B.
for f in Path('.').rglob('*.txt'):
    print(f)

# C.
for f in os.listdir('.'):
    if f.endswith('.txt'):
        print(f)

# D.
以上都不对

答案：B

3. 写入 CSV 文件时，为什么要使用 newline=''？

# A. 加快写入速度
# B. 避免 Windows 上行距加倍
# C. 兼容 Linux
# D. 必须这样做

答案：B

4. 如何复制文件并保留文件的元数据（修改时间等）？

# A. shutil.copy()
# B. shutil.copytree()
# C. shutil.copy2()
# D. shutil.move()

答案：C

5. 删除非空目录应该使用哪个函数？

# A. os.rmdir()
# B. shutil.rmtree()
# C. os.remove()
# D. Path.rmdir()

答案：B

编程题

1. 实现文件搜索工具：

from pathlib import Path
from datetime import datetime

def search_files(directory, pattern='*', min_size=0, max_size=float('inf')):
    """搜索文件"""
    results = []
    dir_path = Path(directory)
    
    for file in dir_path.rglob(pattern):
        if file.is_file():
            try:
                size = file.stat().st_size
                if min_size <= size <= max_size:
                    results.append({
                        'name': file.name,
                        'path': str(file),
                        'size': size,
                        'modified': datetime.fromtimestamp(
                            file.stat().st_mtime
                        ).strftime('%Y-%m-%d %H:%M')
                    })
            except (PermissionError, FileNotFoundError):
                continue
    
    return results

# 使用
results = search_files('.', pattern='*.py', min_size=1000)
for r in results:
    print(f"{r['name']}: {r['size']} 字节, 修改于 {r['modified']}")

2. 实现目录备份脚本：

from pathlib import Path
import shutil
from datetime import datetime

def backup_directory(source, destination):
    """备份目录"""
    src = Path(source)
    dst = Path(destination)
    
    if not src.exists():
        raise FileNotFoundError(f"源目录不存在: {src}")
    
    # 创建带时间戳的备份目录
    timestamp = datetime.now().strftime('%Y%m%d_%H%M%S')
    backup_name = f"{src.name}_backup_{timestamp}"
    backup_path = dst / backup_name
    
    # 复制目录
    shutil.copytree(src, backup_path)
    
    print(f"备份完成: {backup_path}")
    return backup_path

# 使用
backup_directory('/path/to/source', '/path/to/backups')

3. CSV 数据处理工具：

import csv
from pathlib import Path

def process_csv(input_file, output_file, transform_fn):
    """处理 CSV 数据"""
    with open(input_file, 'r', encoding='utf-8') as inf, \
         open(output_file, 'w', newline='', encoding='utf-8') as outf:
        
        reader = csv.DictReader(inf)
        fieldnames = reader.fieldnames
        
        writer = csv.DictWriter(outf, fieldnames=fieldnames)
        writer.writeheader()
        
        for row in reader:
            transformed = transform_fn(row)
            if transformed:  # 如果返回 None 则跳过
                writer.writerow(transformed)

# 使用示例：过滤年龄大于 25 的记录，并添加一列
def transform(row):
    if int(row['age']) > 25:
        row['status'] = 'adult'
        return row
    return None

process_csv('input.csv', 'output.csv', transform)

问答题

Q1: pathlib 相比 os.path 有哪些优势？

  1. 面向对象：Path 是类，有丰富的方法
  2. 链式调用：Path('a') / 'b' / 'c' 更直观
  3. 统一接口：Windows 和 Unix 路径行为一致
  4. 功能丰富：glob, iterdir, read_text 等方法
  5. 可与其他模块混用：可直接传给 os.path 函数

Q2: shutil 模块的主要功能有哪些？

  1. 复制：copy, copy2, copytree
  2. 移动：move
  3. 删除：rmtree
  4. 归档：make_archive, unpack_archive
  5. 磁盘信息：disk_usage
  6. 权限：copystat, chmod

参考资料