Python规则引擎驱动的批量文件管理工具实践

简介面向有Python基础的技术人员与学生这份资源提供了一套基于自定义规则的文件夹文件批量筛选方案。核心功能围绕filter_files函数展开可从文件名包含特定关键字、扩展名类型、最小文件大小限制、距最后修改时间不超过指定天数四个维度组合过滤适用于日常数据管理、清理冗余文档、快速定位目标文件等场景也适合作为学习Python I/O模块与日期时间处理的实战案例。资源包共1个文件文件类型为docx压缩包大小14KB携带方便内容为完整说明文档。目前已有72人学习下载。文档中详细拆解了每个筛选条件的判断流程并给出可直接运行的完整代码示例中通过conditions字典自由设定过滤规则按需省略对应键值对即可适配不同业务逻辑。读者可据此理解os.listdir、os.path.isfile、os.path.getsize、os.path.getmtime等系统调用与datetime模块的配合方式并在此基础上扩展更多个性化筛选条件。需要注意的是实际运行前建议先备份重要数据避免因误筛造成文件遗漏或误操作。1. Python自动化批量文件管理工具为什么过滤规则比一堆 if 靠谱先看一个最常翻车的场景你手工写了二十多个 if 去根据后缀清理旧文件结果日志目录里混着一个没有后缀的 pid 文件和一个带 .2024 的压缩包if 越加越乱最后连自己都不敢跑。Python自动化批量文件管理工具基于自定义规则过滤文件夹中特定文件要做的是把过滤条件从代码里抽出来变成一份可读、可改、可复现的规则文件——后缀、文件名模式、大小、修改时间按需组合匹配结果统一交给脚本去移动、归档或删除。它适合给下载目录、服务器日志目录、测试环境里的临时产物做定期整理。新手按下面步骤能跑通熟手也能把原来的多个 schedule 任务收敛成这一套规则引擎。2. 规则模型先把过滤条件从代码里抽出来2.1 为什么不用 if 堆规则很多人的第一个版本长这样if name.endswith(.tmp) and name.startswith(build_): os.remove(...)这种写法的问题不是“跑不通”而是“不敢改”。你今天加上一个.bak后缀判断明天又要排除.keep文件后天还要看修改时间if 嵌套越来越多每次上线都像在拆盲盒。把条件抽成规则之后逻辑变数据谁都能看懂这条规则在清理什么。常见做法是用 YAML 描述规则因为它的缩进结构比 JSON 适合人读也比 Python 源码更适合非开发角色去维护。规则的职责边界很清晰只负责“什么文件该被处理”至于“处理成什么样”由 action 字段决定。2.2 规则字段的语义与优先级先看一个最小可用的 config.yamlrecursive: true output: ./run.log rules: - name: clean_temp match: any include: extensions: - .tmp - .bak glob: - **/*_test_* exclude: extensions: - .keep action: move target_dir: ./trash这份规则表达的意思是递归扫描当前目录只要扩展名是 .tmp/.bak或者相对路径匹配**/*_test_*模式并且扩展名不是 .keep就移动到 ./trash。关键点在于 exclude 永远先执行命中 exclude 直接放弃不再走 include 判断。各字段语义整理如下字段类型说明recursivebool是否递归子目录默认 truerules[].namestr规则名日志里用来追溯rules[].matchstrinclude 内部多条件是 any 还是 allinclude.extensionslist[str]后缀匹配统一转小写后比较include.globlist[str]基于相对路径的 fnmatch 通配include.regexlist[str]对相对路径做正则搜索include.min_size / max_sizeint字节数过滤include.modified_before_daysfloatmtime 早于 N 天include.modified_after_daysfloatmtime 晚于 N 天excludedict结构与 include 一致命中即排除actionstrdry_run / move / deletetarget_dirstraction 为 move 时的目标目录这里的优先级是很多人第一次写会漏掉的同一文件可能同时命中多条规则默认只执行第一条命中的规则避免同一个文件被移动两次或者先移动后删除。要表达“多条规则独立处理不同文件”靠的是 rules 数组的顺序和 include 条件的差异化。2.3 规则预校验启动时 fail-fast规则写错比文件删错更常见。比如你把min_size写成了字符串把 action 写成了del如果程序不检查往往是处理到一半才炸那时候可能已经误操作了几个文件。我一般会在启动阶段就做一次完整校验def validate_config(cfg): errors [] rules cfg.get(rules, []) if not rules: errors.append(rules 不能为空) for idx, rule in enumerate(rules): if not rule.get(name): errors.append(frules[{idx}].name 是必填项) include rule.get(include) or {} if not include: errors.append(frules[{idx}].include 不能为空否则规则不生效) if rule.get(action) not in (dry_run, move, delete): errors.append(frules[{idx}].action 必须是 dry_run/move/delete) if errors: raise SystemExit(配置校验失败:\n \n.join(errors))这段代码的关键是“宁可启动时报错也不跑一半才出问题”。注意include为空的情况容易被忽略我刻意把它列为错误因为用户在写规则时如果漏了 include脚本默认行为会变得不可预测。action 的枚举校验也是同样道理typo 是最难排查的一类问题直接在配置加载阶段拦截掉最省事。3. 核心过滤实现扫描、匹配与日志对齐3.1 扫描用 scandir 代替 listdir isfile过滤的第一步是拿到所有文件这里有一个常见的性能误区和踩坑点。初学者习惯用os.listdir配合os.path.isfile判断但这样每个文件至少多一次系统调用。os.scandir返回的是 DirEntryis_file()和stat()结果会被缓存同一个文件在匹配多个规则时不必重复读磁盘。def collect_entries(root, recursive): stack [root] while stack: current stack.pop() with os.scandir(current) as it: for entry in it: if entry.is_file(follow_symlinksFalse): yield entry elif recursive and entry.is_dir(follow_symlinksFalse): stack.append(entry.path)这里用迭代栈而不是递归有两个实际好处一是避免目录层级很深时的递归栈溢出二是with os.scandir(...)能保证句柄及时释放。注意follow_symlinksFalse——如果目录里存在指向目录的符号链接默认的is_dir()会跟随它可能导致循环遍历把整个磁盘扫一遍。这是实际生产环境最容易踩的坑。3.2 匹配器把 YAML 编译成可复用对象规则匹配的核心是把 YAML 字段编译成一个可复用的类避免每个文件都重新做字符串拆分和正则编译。正则编译尤其贵对几万个小文件来说重复re.compile的开销能明显拉长整体耗时。class CompiledRule: def __init__(self, rule): self.name rule.get(name) self.action rule.get(action, dry_run) self.target_dir rule.get(target_dir) self.match rule.get(match, any) self.include self._build_conditions(rule.get(include, {})) self.exclude self._build_conditions(rule.get(exclude, {})) def _build_conditions(self, cond): return { exts: {e.lower() for e in cond.get(extensions, [])}, globs: cond.get(glob, []), regexes: [re.compile(p) for p in cond.get(regex, [])], min_size: cond.get(min_size), max_size: cond.get(max_size), before_days: cond.get(modified_before_days), after_days: cond.get(modified_after_days), } def _hit(self, cond, entry, rel_path): hits [] if cond[exts]: hits.append(entry.name.lower().endswith(tuple(cond[exts]))) for pattern in cond[globs]: hits.append(fnmatch.fnmatch(rel_path, pattern)) for regex in cond[regexes]: hits.append(bool(regex.search(rel_path))) stat entry.stat(follow_symlinksFalse) if cond[min_size] is not None: hits.append(stat.st_size cond[min_size]) if cond[max_size] is not None: hits.append(stat.st_size cond[max_size]) now time.time() if cond[before_days] is not None: hits.append(stat.st_mtime now - cond[before_days] * 86400) if cond[after_days] is not None: hits.append(stat.st_mtime now - cond[after_days] * 86400) if not hits: return False return all(hits) if self.match all else any(hits) def matches(self, entry, rel_path): if self.exclude and self._hit(self.exclude, entry, rel_path): return False if not self.include or not self._hit(self.include, entry, rel_path): return False return True_hit的设计有一点需要说明当某个类别比如 glob没有配置时它不会参与判断只有当所有类别都没配置时hits为空返回 False。这样match: any才能正确表达“扩展名或 glob 任一命中”——如果空类别返回 Falseany会把没配置的类别也算进去。这个细节不处理就会出现“只写扩展名却匹配不到任何文件”的玄学问题。min_size和max_size用的是字节1MB 就写1048576不要写成1MB字符串校验阶段不会帮你做单位换算。modified_before_days基于文件的 mtime不是 ctime如果你想表达“文件内容最后一次改动时间”mtime 是对的如果是“文件元数据变了也要算”那需要换成 ctime。3.3 命令行入口与 dry-run 输出把扫描和匹配组装起来再加一个命令行入口整个工具就成形了def match_entries(cfg, root): compiled [CompiledRule(r) for r in cfg[rules]] matched [] for entry in collect_entries(root, cfg.get(recursive, True)): rel_path os.path.relpath(entry.path, root).replace(os.sep, /) for rule in compiled: if rule.matches(entry, rel_path): matched.append((entry, rule, rel_path)) break return matched def parse_args(): parser argparse.ArgumentParser(descriptionfilter files by custom rules) parser.add_argument(--root, requiredTrue, help要扫描的根目录) parser.add_argument(--config, defaultconfig.yaml) parser.add_argument(--target-dir, defaultNone, help覆盖规则里的 target_dir) parser.add_argument(--dry-run, actionstore_true, help只打印不执行) parser.add_argument(--yes, actionstore_true, help跳过交互确认) parser.add_argument(--output, defaultfilter.log, help日志路径) return parser.parse_args()rel_path统一转成/分隔很重要否则 Windows 上写**/*_test_*永远匹配不到test_xxx.txt。命令行参数的意义在于测试时不改 YAML 就能快速验证不同目录--dry-run应该是你第一次运行时的默认操作只有确认输出符合预期才去掉。4. 动作执行与安全边界为什么删除前先过回收站4.1 三种动作dry_run、move、delete匹配到文件之后真正的处理逻辑并不复杂但安全边界必须想清楚。动作分为三档dry_run 只打印move 把文件挪到指定目录delete 直接删除。def run_action(entry, rule, args, log): action rule.action or dry_run if action dry_run: log.info(dry_run | rule%s | %s, rule.name, entry.path) return if action move: dst_dir rule.target_dir or args.target_dir if not dst_dir: raise ValueError(move 动作需要 target_dir) os.makedirs(dst_dir, exist_okTrue) dst unique_path(os.path.join(dst_dir, entry.name)) shutil.move(entry.path, dst) log.info(move | rule%s | %s - %s, rule.name, entry.path, dst) return if action delete: os.unlink(entry.path) log.info(delete | rule%s | %s, rule.name, entry.path)我一般把 delete 作为最后选项。更稳的套路是统一用 move 到一个.trash目录跑几天确认没问题再手动清空。这个习惯救过我好几次——有一次规则里的 glob 写宽了把本该保留的*.back也当成缓存文件处理因为有 trash 目录才没有造成不可逆损失。4.2 原子性与重名冲突move 动作里用到的unique_path是很多人会忽略的细节。如果目标目录已经存在同名文件直接shutil.move会覆盖这可能把两天前的整理结果冲掉。def unique_path(target): if not os.path.exists(target): return target stem, ext os.path.splitext(target) for i in range(1, 10000): candidate f{stem}_{i}{ext} if not os.path.exists(candidate): return candidate return f{stem}_{int(time.time() * 1000)}{ext}shutil.move在同一文件系统内走的是os.rename是原子操作但跨文件系统时它会先复制再删除源文件中途失败可能留下一个半成品。所以如果目标目录和源目录不在同一个磁盘建议把target_dir放在同一分区或者先 copy 再确认。这个边界不写出来生产环境第一次出问题就会很被动。4.3 失败重试与交互确认文件操作经常遇到瞬时原因导致的失败比如 Windows 上杀毒软件正在扫描文件、网络盘短暂抖动。一个简单的指数退避重试能吞掉大部分这类问题def with_retry(func, *args, retries3, base_delay0.5): last_err None for attempt in range(retries): try: return func(*args) except Exception as e: last_err e time.sleep(base_delay * (2 ** attempt)) raise last_err调用点最稳的形态是每个文件单独 try/except单文件失败不影响整批任务但必须记 warning 日志并计数最后汇总一条“成功 N / 失败 M”的结果。main 入口在真实执行前还会问一次 yes/no这个交互确认是防手滑的最后一道闸门def main(): args parse_args() cfg load_config(args.config) validate_config(cfg) root args.root or cfg.get(root, .) matched match_entries(cfg, root) if not args.dry_run and not args.yes: answer input(f将处理 {len(matched)} 个文件输入 yes 继续: ) if answer.strip() ! yes: return failed 0 for entry, rule, rel_path in matched: try: with_retry(run_action, entry, rule, args, log) except Exception: failed 1 log.warning(failed | rule%s | %s, rule.name, entry.path) log.info(done | total%d failed%d, len(matched), failed)定时任务里--yes是可以的但必须保证日志完整。手动调试时不要加--yes多一次确认不会让你慢多少。5. 常见问题与排查五条血泪记录5.1 正则里的括号被当成分组永远匹配不上现象规则里写了regex: [(\d{4})_(\d{2}).log]你明明有一个(2024)_01.log文件但就是匹配不到。原因圆括号在正则里是分组不是字面量你写的是“匹配括号内的内容”不是“匹配括号本身”。解决给括号加反斜杠转义python -c import re; print(bool(re.search(r\(\d{4}\)_\d{2}\.log, (2024)_01.log)))这里还有一个连环坑.log里的点在正则里表示任意字符必须写成\.。如果你只是想要简单的后缀通配别用正则用extensions或glob就够了正则留给真正需要模式匹配的场景。5.2 Windows 路径分隔符导致 glob 失效现象规则里写glob: [log/*.log]Windows 上报错或者匹配不到。原因fnmatch是对字符串做匹配Windows 的真实路径分隔符是反斜杠log/匹配不上log\。解决在算相对路径时统一转成/rel_path os.path.relpath(entry.path, root).replace(os.sep, /)这个修复不会影响后续shutil.move操作因为动作执行用的还是entry.path。不要在匹配阶段用Path对象字符串统一分隔符是最省事的方案。5.3 Linux 下后缀大小写不敏感实现现象规则配了.TMPLinux 上.tmp文件不匹配。原因Linux 文件系统区分大小写Windows/macOS 默认不区分。解决配置里统一小写匹配时也统一小写exts {e.lower() for e in config_extensions} entry.name.lower().endswith(tuple(exts))这个写法在 Windows 上也不会翻车因为lower()对已经是小写的单词没有副作用。如果你在建exts时漏了lower()在 Windows 上可能没事部署到 Linux 才炸属于“换环境才暴露”的典型问题。5.4 dry-run 结果和真实执行结果不一致现象dry-run 打印了 100 个文件真正执行只处理了 80 个。原因dry-run 到真实运行之间有延迟文件在这段时间内被改了 mtime或者被占用了。解决真实执行前重新 stat 一次关键属性变了就跳过并记 warning。before entry.stat(follow_symlinksFalse).st_mtime # ... 执行动作前再次确认 if abs(entry.stat(follow_symlinksFalse).st_mtime - before) 0.001: log.warning(文件已变化跳过: %s, entry.path)这个重验成本很低但能把“计划赶不上变化”的那部分风险挡在门口。5.5 文件被占用或权限不足结果静默跳过现象日志没有报错但文件没被清理你根本不知道它没被执行。原因Windows 上文件被 Excel 打开时os.unlink会抛 PermissionErrorLinux 上可能是目录无写权限。解决不要吞异常每个文件单独捕获并计数failed 0 try: os.unlink(p) except PermissionError: failed 1 log.warning(权限不足: %s, p)最后打印failed总数哪怕只有一个失败也要让你看见。6. 进阶把过滤升级成“热冷分离”的整理流水线基础版本跑通之后最常见的进阶方向是“不是删掉而是按热度归档”。把超过 30 天没动、体积大于 1MB 的文件挪到冷存储目录比直接删更安全也更符合“批量文件管理”的场景rules: - name: hot_to_cold match: all include: modified_before_days: 30 min_size: 1048576 exclude: gлоб: - active/** action: move target_dir: ./cold_storage/match: all表示两个条件都要满足exclude.glob保护 active 目录防止正在使用的文件被搬走。冷存储目录按月份分一层更利于回溯./cold_storage/2026/01/规则里写target_dir时可以用日期字段实现但日常手动维护的话直接写死目录也够用。另一个可以明显提速的优化点是正则预编译复用。用类级别的字典做缓存避免同一批规则在多轮扫描里反复编译class CompiledRule: _regex_cache {} def _compile(pattern): if pattern not in CompiledRule._regex_cache: CompiledRule._regex_cache[pattern] re.compile(pattern) return CompiledRule._regex_cache[pattern]对一万个文件来说正则编译开销可能只差几百毫秒但如果是十万级文件、每天定时跑这个差距会累积成肉眼可见的 CPU 消耗。最后说一个我自己栽过的跟头早期我图省事把下载目录、日志目录、构建产物各写了一套脚本等新文件类型出现要改的地方散在六个脚本里改一个漏一个最后误删了一个本该保留的配置备份。后来收敛成这一套规则配置新增一种文件类型只需要在 YAML 里加一行 glob 或 extension规则可复用、日志可追溯、意外可回滚。希望帮到你。本文还有配套的精品资源点击获取