企业官网建设流程全解析

Python 3.10+ 实战：用 pathlib 和 glob 递归筛选 5 种常见文件类型

在自动化文件处理任务中，递归遍历目录并筛选特定类型的文件是最基础也最高频的需求之一。传统方法往往需要组合多个模块和函数，而现代 Python（3.10+）的 pathlib 和 glob 模块提供了更优雅的解决方案。本文将带你开发一个工业级可用的文件筛选工具，不仅能处理图片、文档、代码等常见类型，还包含完整的错误处理和路径规范化机制。

1. 为什么选择 pathlib 和 glob？

在 Python 3.4 引入的 pathlib 模块彻底改变了文件路径的操作方式。相比传统的 os.path，它具有三大优势：

• 面向对象 API：路径不再是字符串，而是 Path 对象，方法链式调用更直观
• 跨平台一致性：自动处理 Windows 和 Unix 的路径分隔符差异
• 功能集成：一个模块替代了 os、os.path、glob 等多个模块的功能

以下是传统方法与 pathlib 的对比示例：

# 传统方式 import os.path dir_path = os.path.join('data', 'images') is_file = os.path.isfile(dir_path) # pathlib方式 from pathlib import Path dir_path = Path('data') / 'images' is_file = dir_path.is_file()

glob 模块则通过模式匹配简化了文件筛选。在 Python 3.11+ 中，glob 支持递归通配符**，使得深层目录遍历变得异常简单。

2. 构建核心筛选函数

我们首先实现一个支持递归遍历和扩展名筛选的基础函数：

from pathlib import Path from typing import Iterable, Union def filter_files_by_ext( root_dir: Union[str, Path], extensions: Iterable[str], recursive: bool = True ) -> list[Path]: """ 按扩展名筛选目录中的文件 :param root_dir: 搜索根目录 :param extensions: 目标扩展名集合(如 ['.jpg', '.png']) :param recursive: 是否递归搜索子目录 :return: 匹配的Path对象列表 """ path = Path(root_dir).resolve() if not path.exists(): raise FileNotFoundError(f"目录不存在: {path}") pattern = "**/*" if recursive else "*" all_files = path.glob(pattern) ext_set = {ext.lower() for ext in extensions} return [ f for f in all_files if f.is_file() and f.suffix.lower() in ext_set ]

这个函数已经可以处理基本需求：

# 查找所有Python和Markdown文件 files = filter_files_by_ext('.', ['.py', '.md'])

3. 支持5种常见文件类型

我们将文件分为五大类，每类定义典型扩展名：

扩展核心函数，增加预设类型支持：

from enum import Enum, auto class FileType(Enum): IMAGE = auto() DOCUMENT = auto() CODE = auto() ARCHIVE = auto() CONFIG = auto() FILE_TYPE_EXTENSIONS = { FileType.IMAGE: {'.jpg', '.jpeg', '.png', '.gif', '.webp', '.bmp'}, FileType.DOCUMENT: {'.pdf', '.docx', '.pptx', '.xlsx', '.doc', '.ppt', '.xls'}, FileType.CODE: {'.py', '.js', '.java', '.cpp', '.h', '.go', '.rs', '.ts'}, FileType.ARCHIVE: {'.zip', '.rar', '.7z', '.tar.gz', '.bz2'}, FileType.CONFIG: {'.json', '.yaml', '.yml', '.toml', '.ini', '.cfg'} } def filter_files_by_type( root_dir: Union[str, Path], file_types: Iterable[FileType], recursive: bool = True ) -> dict[FileType, list[Path]]: """ 按预定义文件类型筛选目录中的文件 :param root_dir: 搜索根目录 :param file_types: 目标文件类型集合 :param recursive: 是否递归搜索子目录 :return: 按类型分类的Path字典 """ path = Path(root_dir).resolve() if not path.exists(): raise FileNotFoundError(f"目录不存在: {path}") # 合并所有需要的扩展名 ext_set = set() for ft in file_types: ext_set.update(FILE_TYPE_EXTENSIONS[ft]) # 获取基础文件列表 matched_files = filter_files_by_ext(path, ext_set, recursive) # 按类型分类 result = {ft: [] for ft in file_types} for file in matched_files: for ft in file_types: if file.suffix.lower() in FILE_TYPE_EXTENSIONS[ft]: result[ft].append(file) break return result

使用示例：

# 查找所有图片和文档 files_by_type = filter_files_by_type( '/data', [FileType.IMAGE, FileType.DOCUMENT] )

4. 高级功能实现

4.1 并行化处理

对于大型目录，我们可以利用多线程加速文件遍历：

from concurrent.futures import ThreadPoolExecutor def parallel_filter_files_by_type( root_dir: Union[str, Path], file_types: Iterable[FileType], max_workers: int = 4 ) -> dict[FileType, list[Path]]: path = Path(root_dir).resolve() if not path.exists(): raise FileNotFoundError(f"目录不存在: {path}") # 获取所有子目录 all_dirs = [d for d in path.glob("**/*") if d.is_dir()] all_dirs.append(path) # 包含根目录 ext_set = set() for ft in file_types: ext_set.update(FILE_TYPE_EXTENSIONS[ft]) result = {ft: [] for ft in file_types} def process_dir(directory: Path): for f in directory.glob("*"): if not f.is_file(): continue for ft in file_types: if f.suffix.lower() in FILE_TYPE_EXTENSIONS[ft]: result[ft].append(f) break with ThreadPoolExecutor(max_workers=max_workers) as executor: executor.map(process_dir, all_dirs) return result

4.2 文件元信息收集

扩展函数以包含文件大小、修改时间等元数据：

from dataclasses import dataclass from datetime import datetime @dataclass class FileInfo: path: Path size: int # bytes modified: datetime file_type: FileType def get_file_info(file_path: Path) -> FileInfo: """获取文件的详细元信息""" stat = file_path.stat() for ft, exts in FILE_TYPE_EXTENSIONS.items(): if file_path.suffix.lower() in exts: return FileInfo( path=file_path, size=stat.st_size, modified=datetime.fromtimestamp(stat.st_mtime), file_type=ft ) raise ValueError(f"未知文件类型: {file_path.suffix}")

4.3 排除隐藏文件和系统文件

在 glob 模式中添加过滤条件：

def is_hidden(file_path: Path) -> bool: """检查是否为隐藏文件(跨平台)""" name = file_path.name return name.startswith('.') or name.startswith('~') def filter_files_with_exclusions( root_dir: Path, file_types: Iterable[FileType], exclude_hidden: bool = True ) -> dict[FileType, list[Path]]: result = filter_files_by_type(root_dir, file_types) if exclude_hidden: for ft in result: result[ft] = [f for f in result[ft] if not is_hidden(f)] return result

5. 完整工具实现

将上述功能整合为一个文件处理工具类：

class FileScanner: def __init__(self, root_dir: Union[str, Path]): self.root = Path(root_dir).resolve() if not self.root.exists(): raise FileNotFoundError(f"目录不存在: {self.root}") def scan( self, file_types: Iterable[FileType], recursive: bool = True, parallel: bool = False, exclude_hidden: bool = True ) -> dict[FileType, list[FileInfo]]: """ 扫描目录并返回分类文件信息 :param file_types: 要扫描的文件类型集合 :param recursive: 是否递归子目录 :param parallel: 是否使用多线程加速 :param exclude_hidden: 是否排除隐藏文件 :return: 按类型分类的文件信息字典 """ if parallel: files = parallel_filter_files_by_type( self.root, file_types ) else: files = filter_files_by_type( self.root, file_types, recursive ) if exclude_hidden: for ft in files: files[ft] = [f for f in files[ft] if not is_hidden(f)] # 转换为FileInfo对象 return { ft: [get_file_info(f) for f in files[ft]] for ft in files } def summary(self, file_infos: dict[FileType, list[FileInfo]]) -> str: """生成扫描结果摘要""" lines = [] total_files = 0 total_size = 0 for ft, infos in file_infos.items(): type_count = len(infos) type_size = sum(i.size for i in infos) lines.append( f"{ft.name:<10}: {type_count:>4} 文件, " f"{type_size/1024/1024:.2f} MB" ) total_files += type_count total_size += type_size lines.insert(0, f"扫描目录: {self.root}") lines.append("-" * 40) lines.append( f"{'总计':<10}: {total_files:>4} 文件, " f"{total_size/1024/1024:.2f} MB" ) return "\n".join(lines)

使用示例：

scanner = FileScanner("~/projects") results = scanner.scan([ FileType.CODE, FileType.DOCUMENT ]) print(scanner.summary(results))

输出示例：

扫描目录: /home/user/projects CODE : 42 文件, 1.75 MB DOCUMENT : 8 文件, 4.32 MB ---------------------------------------- 总计 : 50 文件, 6.07 MB

6. 性能优化技巧

1. 缓存扩展名集合：将FILE_TYPE_EXTENSIONS转换为frozenset提升查找速度
2. 提前终止遍历：对大型目录可设置最大文件数限制
3. 惰性求值：使用生成器替代列表保存中间结果
4. 索引加速：对重复扫描可建立文件索引数据库

优化后的 glob 模式示例：

# 使用rglob进行递归遍历（Python 3.11+优化） def fast_glob(path: Path, pattern: str): if "**" in pattern: return path.rglob(pattern.replace("**/*", "")) return path.glob(pattern)

7. 错误处理与边缘情况

健壮的文件处理需要考虑以下异常情况：

def safe_scan( root_dir: Union[str, Path], file_types: Iterable[FileType] ) -> dict[FileType, list[FileInfo]]: try: scanner = FileScanner(root_dir) return scanner.scan(file_types) except FileNotFoundError as e: print(f"错误: {e}") return {} except PermissionError: print(f"无权限访问目录: {root_dir}") return {} except Exception as e: print(f"未知错误: {e}") return {}

特殊场景处理建议：

• 符号链接：使用Path.resolve()解析真实路径
• 非法文件名：捕获UnicodeEncodeError
• 网络驱动器：添加超时机制
• 内存限制：分批处理结果

8. 实际应用案例

8.1 项目文档自动化收集

def collect_project_docs(project_root: Path): scanner = FileScanner(project_root) docs = scanner.scan([ FileType.DOCUMENT, FileType.CONFIG ]) # 按修改时间排序 for ft in docs: docs[ft].sort(key=lambda x: x.modified, reverse=True) # 生成README.md with open(project_root / "README.md", "w") as f: f.write("# 项目文档索引\n\n") for ft, infos in docs.items(): f.write(f"## {ft.name}\n") for info in infos: rel_path = info.path.relative_to(project_root) f.write(f"- [{rel_path}]({rel_path}) ") f.write(f"({info.modified.date()}, {info.size/1024:.1f} KB)\n") f.write("\n")

8.2 图片资源批量处理

from PIL import Image def process_images(source_dir: Path, target_dir: Path): scanner = FileScanner(source_dir) images = scanner.scan([FileType.IMAGE]) target_dir.mkdir(exist_ok=True) for img_info in images[FileType.IMAGE]: try: with Image.open(img_info.path) as img: # 转换为WebP格式并调整大小 new_path = target_dir / f"{img_info.path.stem}.webp" img.resize((1024, 1024)).save(new_path, "WEBP") except Exception as e: print(f"处理失败 {img_info.path}: {e}")

8.3 代码库统计分析

def analyze_codebase(project_root: Path): scanner = FileScanner(project_root) code_files = scanner.scan([FileType.CODE]) loc_stats = {} for lang, exts in { "Python": {".py"}, "JavaScript": {".js", ".ts"}, "C++": {".cpp", ".hpp", ".h"} }.items(): files = [ f for f in code_files[FileType.CODE] if f.path.suffix in exts ] total_lines = 0 for f in files: try: with open(f.path, "r") as src: total_lines += sum(1 for _ in src) except Exception: continue loc_stats[lang] = total_lines print("代码行数统计:") for lang, lines in loc_stats.items(): print(f"{lang}: {lines}")