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

Python 3.10 实战用 pathlib 和 glob 递归筛选 5 种常见文件类型在自动化文件处理任务中递归遍历目录并筛选特定类型的文件是最基础也最高频的需求之一。传统方法往往需要组合多个模块和函数而现代 Python3.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种常见文件类型我们将文件分为五大类每类定义典型扩展名文件类型常见扩展名图片.jpg, .png, .gif, .webp文档.pdf, .docx, .pptx, .xlsx代码.py, .js, .java, .cpp, .go压缩包.zip, .rar, .7z, .tar.gz配置文件.json, .yaml, .toml, .ini扩展核心函数增加预设类型支持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_workersmax_workers) as executor: executor.map(process_dir, all_dirs) return result4.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( pathfile_path, sizestat.st_size, modifieddatetime.fromtimestamp(stat.st_mtime), file_typeft ) 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 result5. 完整工具实现将上述功能整合为一个文件处理工具类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 MB6. 性能优化技巧缓存扩展名集合将FILE_TYPE_EXTENSIONS转换为frozenset提升查找速度提前终止遍历对大型目录可设置最大文件数限制惰性求值使用生成器替代列表保存中间结果索引加速对重复扫描可建立文件索引数据库优化后的 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(keylambda x: x.modified, reverseTrue) # 生成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_okTrue) 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})