第三十八篇：操作文件和目录——os 与 shutil

前言

Python 提供了丰富的文件系统操作工具。os 模块提供基础的操作系统接口，shutil 提供更高级的文件和目录操作。本文将从基础到进阶，系统讲解如何高效地操作文件和目录。

os 模块——操作系统接口

获取当前工作目录与环境变量

import os

# 获取当前工作目录
cwd = os.getcwd()
print(f"当前目录: {cwd}")
# 例如: C:\Users\admin\.trae-cn\work

# 更改工作目录
os.chdir("/tmp")
print(os.getcwd())
os.chdir(cwd)  # 恢复

# 获取环境变量
home = os.environ.get("HOME") or os.environ.get("USERPROFILE")
print(f"用户目录: {home}")

# 遍历所有环境变量（前 5 个）
count = 0
for key, value in os.environ.items():
    print(f"{key}={value}")
    count += 1
    if count >= 5:
        break

os.path 路径处理

import os.path as path

base = "/home/user/documents/report_2026.txt"

# 路径分解
directory, filename = path.split(base)
name, ext = path.splitext(base)
print(directory)   # /home/user/documents
print(filename)    # report_2026.txt
print(name)        # /home/user/documents/report_2026
print(ext)         # .txt

# 路径拼接
full = path.join("/home/user", "documents", "file.txt")
print(full)  # /home/user/documents/file.txt

# 信息查询
stat_info = path.stat(base)
print(f"文件大小: {stat_info.st_size} bytes")
print(f"创建时间: {stat_info.st_ctime}")
print(f"修改时间: {stat_info.st_mtime}")

# 转换时间戳为可读格式
import datetime
mtime = datetime.datetime.fromtimestamp(stat_info.st_mtime)
print(f"最后修改: {mtime}")

# 相对路径与绝对路径
rel = path.relpath("/home/user/documents", "/home/user")
abs_from_rel = path.abspath(rel)
print(rel)       # documents
print(abs_from_rel)  # /home/user/documents

# 标准化路径
weird = "/home/user/../user/./documents/."
normalized = path.normpath(weird)
print(normalized)  # /home/user/documents

# 获取真实路径（解析符号链接）
real = path.realpath("/home/user")
print(real)

目录遍历

import os

def list_directory_structure(root, indent=0):
    """递归打印目录结构"""
    for item in sorted(os.listdir(root)):
        full_path = os.path.join(root, item)
        prefix = "  " * indent + ("📁 " if os.path.isdir(full_path) else "📄 ")
        print(f"{prefix}{item}")
        if os.path.isdir(full_path):
            list_directory_structure(full_path, indent + 1)

# 使用示例
list_directory_structure(".")

glob 通配符匹配文件

import glob

# 查找所有 Python 文件
py_files = glob.glob("**/*.py", recursive=True)
print(py_files)

# 查找所有 CSV 文件
csv_files = glob.glob("data/**/*.csv")
print(csv_files)

# 查找特定模式的文件
report_files = glob.glob("reports/**/report_*.pdf")
print(report_files)

shutil —— 高级文件操作

文件复制

import shutil

# 方法一：copy() — 复制内容（不保留元数据）
shutil.copy("source.txt", "destination.txt")

# 方法二：copy2() — 复制内容 + 元数据（ctime/mtime 等）
shutil.copy2("source.pdf", "backup/source_backup.pdf")

# 方法三：copyfile() — 仅复制内容（底层实现，更快）
shutil.copyfile("large.bin", "large_copy.bin")

# 验证复制结果
src_size = os.path.getsize("source.txt")
dst_size = os.path.getsize("destination.txt")
print(f"源文件: {src_size} bytes")
print(f"目标文件: {dst_size} bytes")
assert src_size == dst_size

目录复制

import shutil
import os

def copy_directory(src_dir, dest_dir, follow_symlinks=True):
    """完整复制目录树"""
    if not os.path.exists(dest_dir):
        os.makedirs(dest_dir)
    
    copied = 0
    skipped = 0
    
    for root, dirs, files in os.walk(src_dir):
        relative_root = os.path.relpath(root, src_dir)
        dest_root = os.path.join(dest_dir, relative_root)
        
        if not os.path.exists(dest_root):
            os.makedirs(dest_root)
        
        for file in files:
            src_file = os.path.join(root, file)
            dest_file = os.path.join(dest_root, file)
            
            try:
                shutil.copy2(src_file, dest_file)
                copied += 1
            except Exception as e:
                print(f"跳过 {src_file}: {e}")
                skipped += 1
    
    return {"copied": copied, "skipped": skipped}

result = copy_directory("project_src", "project_backup")
print(f"已复制 {result['copied']} 个文件，跳过 {result['skipped']} 个")

移动与删除

import shutil
import os

# 移动文件或目录
shutil.move("old_location/file.txt", "new_location/file.txt")

# 重命名（本质上是同目录下的移动）
shutil.move("old_name.txt", "new_name.txt")

# 安全删除整个目录树
def safe_delete(directory):
    """安全删除目录及所有内容"""
    if os.path.exists(directory):
        try:
            shutil.rmtree(directory)
            print(f"已删除: {directory}")
        except PermissionError:
            print(f"权限不足: {directory}")
        except OSError as e:
            print(f"删除失败: {e}")

safe_delete("temp_files")

# 安全删除单个文件
def safe_remove(filepath):
    if os.path.isfile(filepath):
        os.remove(filepath)
        print(f"已删除: {filepath}")

safe_remove("unused_config.json")

压缩与解压

import shutil
import os

def compress_to_tar_gz(source_dir, output_filename="archive.tar.gz"):
    """将目录压缩为 tar.gz 归档"""
    if os.path.exists(output_filename):
        os.remove(output_filename)
    
    shutil.make_archive(
        base_name=output_filename.replace(".tar.gz", ""),
        format="gztar",
        root_dir=os.path.dirname(source_dir) or ".",
        base_dir=os.path.basename(source_dir),
    )
    
    size_mb = os.path.getsize(output_filename) / (1024 * 1024)
    print(f"已压缩: {output_filename} ({size_mb:.2f} MB)")
    return output_filename

def extract_archive(archive_path, dest_dir="."):
    """解压缩 tar/gz/zip 文件"""
    shutil.unpack_archive(archive_path, dest_dir)
    print(f"已解压: {archive_path} → {dest_dir}")

# 使用示例
compress_to_tar_gz("my_project")
# extract_archive("my_project.tar.gz")

实战：目录整理工具

import os
import shutil
from collections import defaultdict

def organize_directory(target_dir, dry_run=True):
    """
    整理目录下文件到子文件夹。
    根据文件后缀分类存放。
    
    Args:
        target_dir: 目标目录路径
        dry_run: True 时只模拟不执行
    
    Returns:
        整理结果统计
    """
    # 定义文件扩展名映射
    extensions_map = {
        "图片": [".jpg", ".jpeg", ".png", ".gif", ".bmp", ".svg", ".webp"],
        "文档": [".doc", ".docx", ".pdf", ".txt", ".md", ".rtf"],
        "表格": [".xls", ".xlsx", ".csv"],
        "压缩包": [".zip", ".rar", ".7z", ".tar", ".gz", ".bz2"],
        "音频": [".mp3", ".wav", ".flac", ".aac", ".ogg"],
        "视频": [".mp4", ".avi", ".mkv", ".mov", ".wmv"],
        "代码": [".py", ".js", ".html", ".css", ".java", ".c", ".cpp", ".go", ".rs"],
        "其他": [],
    }
    
    # 构建反向映射
    ext_to_category = {}
    for category, exts in extensions_map.items():
        for ext in exts:
            ext_to_category[ext.lower()] = category
    
    result = defaultdict(lambda: {"files": [], "errors": []})
    
    # 扫描目录
    items = os.listdir(target_dir)
    
    for item in items:
        filepath = os.path.join(target_dir, item)
        
        if not os.path.isfile(filepath):
            continue
        
        _, ext = os.path.splitext(item)
        ext_lower = ext.lower()
        
        category = ext_to_category.get(ext_lower, "其他")
        category_dir = os.path.join(target_dir, category)
        dest = os.path.join(category_dir, item)
        
        try:
            if dry_run:
                print(f"[模拟] {item} → {category}/{item}")
            else:
                if not os.path.exists(category_dir):
                    os.makedirs(category_dir)
                
                if os.path.exists(dest):
                    print(f"[跳过] {item} 已存在")
                    continue
                
                shutil.move(filepath, dest)
                print(f"[移动] {item} → {category}/{item}")
            
            result[category]["files"].append(item)
        
        except Exception as e:
            result["错误"]["errors"].append(f"{item}: {e}")
            print(f"[错误] {item}: {e}")
    
    if dry_run:
        print("\n--- 整理预览 ---")
    
    # 统计
    stats = []
    for category, info in result.items():
        count = len(info["files"])
        if count > 0:
            stats.append(f"  {category}: {count} 个文件")
        for error in info.get("errors", []):
            stats.append(f"  ⚠️ 错误: {error}")
    
    return dict(result)

# 测试运行（dry_run=True 只预览不执行）
target = "test_folder"
if not os.path.exists(target):
    os.makedirs(target)
    # 创建一些测试文件
    for name in ["photo.jpg", "readme.md", "data.xlsx", "archive.zip", 
                 "song.mp3", "video.mp4", "script.py", "notes.txt"]:
        with open(os.path.join(target, name), "w") as f:
            f.write("test content\n")

result = organize_directory(target, dry_run=True)

实战：日志轮转管理

import os
import shutil
import glob
import time
from datetime import datetime

class LogRotator:
    """日志文件自动轮转管理器"""
    
    def __init__(self, log_dir="logs", max_days=30, max_size_mb=50):
        self.log_dir = log_dir
        self.max_days = max_days
        self.max_size_mb = max_size_mb
        os.makedirs(log_dir, exist_ok=True)
    
    def get_log_files(self):
        """获取所有日志文件"""
        return glob.glob(os.path.join(self.log_dir, "*.log"))
    
    def should_rotate(self, filepath):
        """检查日志文件是否需要轮转"""
        stat = os.stat(filepath)
        size_mb = stat.st_size / (1024 * 1024)
        
        if size_mb >= self.max_size_mb:
            return True, f"大小 {size_mb:.1f}MB >= {self.max_size_mb}MB"
        
        age_days = (time.time() - stat.st_mtime) / (24 * 3600)
        if age_days >= self.max_days:
            return True, f"年龄 {age_days:.0f}天 >= {self.max_days}天"
        
        return False, None
    
    def rotate_log(self, filepath):
        """轮转单个日志文件"""
        timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
        base_name = os.path.splitext(filepath)[0]
        rotated_path = f"{base_name}_{timestamp}.log"
        
        shutil.copy2(filepath, rotated_path)
        
        # 清空原文件
        with open(filepath, "w") as f:
            f.truncate(0)
        
        print(f"[轮转] {os.path.basename(filepath)} → {os.path.basename(rotated_path)}")
        return rotated_path
    
    def clean_old_logs(self):
        """清理超过保留期的旧日志"""
        deleted = 0
        cutoff_time = time.time() - self.max_days * 24 * 3600
        
        old_logs = glob.glob(os.path.join(self.log_dir, "*.log.*"))
        for log_file in old_logs:
            if os.stat(log_file).st_mtime < cutoff_time:
                os.remove(log_file)
                deleted += 1
        
        if deleted > 0:
            print(f"[清理] 删除了 {deleted} 个过期日志文件")
        
        return deleted
    
    def manage(self, dry_run=True):
        """统一管理入口"""
        print(f"\n开始日志管理: {self.log_dir}")
        print(f"规则: 最大 {self.max_size_mb}MB / 保留 {self.max_days} 天\n")
        
        rotated = []
        log_files = self.get_log_files()
        
        for filepath in log_files:
            need_rotation, reason = self.should_rotate(filepath)
            if need_rotation:
                print(f"[需要轮转] {os.path.basename(filepath)}: {reason}")
                if not dry_run:
                    rotated.append(self.rotate_log(filepath))
            else:
                print(f"[正常] {os.path.basename(filepath)}")
        
        if not dry_run:
            self.clean_old_logs()
        
        print(f"\n完成! 轮转了 {len(rotated)} 个文件\n")
        return rotated

# 使用示例
rotator = LogRotator("logs", max_days=30, max_size_mb=10)
rotator.manage(dry_run=True)

实战：项目文件模板生成器

import os
import shutil

def create_project_template(project_name, output_dir="."):
    """创建标准的项目目录结构和模板文件"""
    project_path = os.path.join(output_dir, project_name)
    
    # 定义目录结构
    directories = [
        "src",
        "tests",
        "docs",
        "examples",
        ".github/workflows",
    ]
    
    # 定义模板文件
    template_files = {
        "README.md": """# {{project_name}}

## 简介
{{project_name}} 是一个 Python 项目。

## 安装
~~~bash
pip install -r requirements.txt
~~~

## 使用方法
~~~bash
python -m src.main
~~~

## 运行测试
~~~bash
python -m pytest tests/ -v
~~~
""",
        "requirements.txt": "# Python 依赖包\npytest>=7.0\nblack>=22.0\nflake8>=5.0\n",
        "setup.py": """from setuptools import setup, find_packages

setup(
    name="{{project_name}}",
    version="0.1.0",
    packages=find_packages(where="src"),
    package_dir={{"": "src"}},
    python_requires=">=3.8",
)
""",
        ".gitignore": """__pycache__/
*.py[cod]
*.egg-info/
dist/
build/
.env
*.log
.DS_Store
""",
    }
    
    # 创建目录
    for dirpath in directories:
        full_path = os.path.join(project_path, dirpath)
        os.makedirs(full_path, exist_ok=True)
        print(f"[目录] {dirpath}/")
    
    # 创建模板文件
    for filename, content in template_files.items():
        filepath = os.path.join(project_path, filename)
        with open(filepath, "w", encoding="utf-8") as f:
            f.write(content.format(project_name=project_name))
        print(f"[文件] {filename}")
    
    # 创建 Python 初始化文件
    init_content = '__version__ = "0.1.0"'
    with open(os.path.join(project_path, "src", "__init__.py"), "w") as f:
        f.write(init_content)
    print("[文件] src/__init__.py")
    
    test_init = """import unittest
from src import __version__

class TestProject(unittest.TestCase):
    def test_version(self):
        self.assertIsInstance(__version__, str)

if __name__ == "__main__":
    unittest.main()
"""
    with open(os.path.join(project_path, "tests", "__init__.py"), "w") as f:
        f.write(test_init)
    print("[文件] tests/__init__.py")
    
    print(f"\n✅ 项目 '{project_name}' 创建完成!")
    print(f"📂 路径: {project_path}\n")

# 使用
create_project_template("my_app")

os vs pathlib 选择指南

场景	os.path	pathlib
基础兼容	✅ 适用所有 Python 版本	⚠️ Python 3.4+
简单路径拼接	✅ os.path.join()	✅ Path() / Path()
路径操作链式调用	❌ 繁琐	✅ .parent, .suffix
跨平台兼容性	✅ 良好	✅ 更好
现代项目推荐	—	✅ 首选

import os
from pathlib import Path

# os.path 风格
src_dir = os.path.join("/home", "user", "project")
files = os.listdir(src_dir)

# pathlib 风格（更直观）
src_dir = Path("/home/user/project")
files = list(src_dir.iterdir())

# 两者对比
# os.path:
print(os.path.join("a", "b", "c"))      # a/b/c
print(os.path.splitext("file.txt"))      # ('file', '.txt')
print(os.path.basename("a/b/c.txt"))     # c.txt

# pathlib:
print(Path("a") / "b" / "c")             # a/b/c
print(Path("file.txt").suffix)           # .txt
print(Path("a/b/c.txt").name)            # c.txt

最佳实践总结

✅ 推荐做法

1. 优先使用 pathlib — 现代、直观、跨平台
2. 使用 shutil.copy2 保留元数据 — 比 copy 更安全
3. 大目录用 os.walk 惰性遍历 — 避免一次性加载所有文件列表
4. 删除前做 dry-run — 避免误删重要文件
5. 捕获异常确保数据安全 — 区分 FileNotFoundError、PermissionError 等

❌ 避免做法

# ❌ 直接递归删除可能出错
import os
def delete_all(path):
    for item in os.listdir(path):
        p = os.path.join(path, item)
        if os.path.isdir(p):
            delete_all(p)  # 深层递归可能导致栈溢出
        else:
            os.remove(p)
    os.rmdir(path)

# ✅ 使用 shutil.rmtree
import shutil
shutil.rmtree(path)

# ❌ 忽略异常直接操作
shutil.copy("important.doc", "backup/")  # 如果目标不存在会报错

# ✅ 先检查再操作
backup_dir = "backup"
if not os.path.exists(backup_dir):
    os.makedirs(backup_dir)
shutil.copy("important.doc", backup_dir)

总结

本文系统讲解了 Python 文件系统操作的核心知识：

1. os 模块 — 工作目录、环境变量、路径处理
2. os.path — split/join/extinfo/stat 等常用函数
3. 目录遍历 — listdir/os.walk/glob 通配符
4. shutil — 文件复制(move/copy2)、目录复制、压缩解压
5. 实战工具 — 目录整理工具、日志轮转管理器、项目模板生成器
6. pathlib vs os.path — 现代项目推荐使用 pathlib
7. 最佳实践 — 安全删除、异常处理、跨平台兼容

下一篇文章将深入讲解序列化：pickle 与 JSON。

---

本文是 Python 教程系列的第三十八篇。