尧图网页设计 
如果项目的清理工具模块叫“赵光义”第一眼看到这个名字很多人都会有点出戏。不过它后面跟的另一句话很有价值清理工具“总是分外用心”。这句话听起来不像技术指标却是线上清理工具最容易踩坑的关键点。清理工具做得好不是看它删除速度有多快而是看它能不能做到少误删、可恢复、有记录。本文不写任何复杂的底层框架只讲一类朴素但实用的场景用 Python 实现一个面向日志文件、临时文件的清理工具。工具会包含配置化扫描、预览模式、隔离回收、审计日志这几个能力。代码可以直接复制到本地项目验证也可以按团队实际情况改造成运维小工具。无论你是在做脚本工具还是给业务系统写维护模块这套思路都适用。1. 背景与核心概念1.1 清理工具到底在清理什么清理工具的目标很直接把不再需要的临时文件、过期日志、构建产物、缓存文件从磁盘上释放出来。注意这里说的是“不再需要”而不是“我看着不顺眼”。如果一个工具只维护一份文件后缀列表然后前缀式地删除它就不是清理工具而是“高风险事故制造器”。我在实际项目里见过不少类似需求定时清理应用日志目录里超过 N 天的.log文件清理测试环境中的.tmp、.bak、.old文件限制单个日志文件超过多少 MB 后归档或移除报表服务生成的多余临时文件在任务结束后统一回收。这些需求看似简单实际实现时却会涉及几个共同问题要清理哪些目录、按什么规则判断、哪些文件需要保留、删除前是否需要审计、误删后能不能找回。如果只靠一条rm -rf或os.remove生产环境很容易出问题。1.2 清理工具最容易踩的风险清理工具首先面对的是“误删风险”。文件一旦被物理删除很多场景下是找不回来的尤其是没有版本管理、没有备份、没有审计日志的服务器目录。一个配置项写错、一个通配符写宽就可能导致整批业务文件被清掉。第二个风险是“规则误判”。比如有的项目会在日志目录里放模板文件只是文件名中带着log字样有的缓存文件虽然没有近期修改时间但业务启动时会重新读取。如果清理规则只看文件后缀就会把正在使用的文件移走。第三个风险是“配置影响面过大”。本地开发时目录是相对路径一旦上线后把 target 指向根目录或网络挂载目录扫描范围会成倍放大。实际处理时必须给扫描动作加上边界限制而不是盲目递归。第四个风险是“永久删除”。线上清理解析日志、归档文件后即便文件被删除也应该有一个兜底机制。比较好的做法是「先隔离后清理」先把文件移动到回收区确认无误之后再过期删除。整个过程还要有日志否则出现问题后连排查线索都没有。1.3 本文能帮你掌握什么本文将以一个名为cleaner.py的 Python 清理工具为例带大家实现如下能力通过 YAML 配置维护目标目录和清理规则使用dry-run先预览不实际改动文件使用隔离目录代替直接删除文件自动记录清理前后文件信息与审计日志限制扫描深度、跳过符号链接、避免误删关键目录通过命令行参数控制执行模式。代码重点在思路和可扩展性不依赖数据库、消息队列等重型组件适合中小项目和运维场景快速落地。2. 安全清理工具设计思路写清理工具不能一上来先写unlink方法要先想清楚清理流程里的几个核心概念。2.1 “清理”应该分成几个动作很多清理工具把“删除文件”当作唯一动作这是设计上的简化。更安全的设计应该把清理拆成三个层级第一层是“预览”。扫描文件后只输出哪些文件会命中规则文件大小、修改时间、所在目录是什么但不对文件做任何变更。这个动作适合在深夜任务执行前或者人工运维时先验证配置。第二层是“隔离”。把命中的文件移动到回收目录例如.rescue/20250101。这样文件虽然从原始目录消失了但仍然保留在磁盘上。如果业务发现文件还需要运维可以快速恢复。第三层是“过期清理”。隔离区中的文件在下一次巡检时如果超过保留期限再进行物理删除。这一步应该默认不开启由人工或定时任务显式触发。安全的清理工具默认动作应该是预览或隔离而不是永久删除。本文示例代码会把preview作为默认模式只有手动传入--mode delete时才会执行真实删除。2.2 配置与规则要分离清理规则的维护者不一定是开发清理工具的人。运维人员可能希望只修改 YAML而不是去改 Python 源码。所以一个规范的清理工具至少应该包含两部分内容一是“清理范围”也就是targets。它描述工具允许扫描哪些目录不能扫描哪些目录。二是“清理规则”也就是rules。它描述文件需要满足什么条件例如后缀必须匹配哪些类型、修改时间是否超过 N 天、文件大小是否超过 N MB。把配置和代码分离后新增一条规则只需要修改配置文件不用改动程序逻辑也更容易走配置评审流程。2.3 审计日志必须独立保存清理工具一旦在真实环境执行审计日志和清理动作本身同等重要。审计日志最好包含以下信息任务执行时间匹配到的清理规则名称文件绝对路径文件大小文件修改时间执行动作是预览、移动还是删除移动后的隔离路径任务结果。注意审计日志文件不能放在待清理目录中。如果日志目录本身命中规则可能被清理脚本删掉导致没有日志可查。通常建议把审计日志输出到独立的./audit或服务器集中日志目录并使用 UTF-8 编码保存。3. 环境准备与项目结构3.1 运行环境本文示例代码使用 Python 编写依赖很少Python 3.10 及以上版本PyYAML用于读取 YAML 配置文件操作系统Windows、Linux、macOS 均可不需要额外数据库不需要消息队列。Python 版本可以根据你的实际情况调整。如果你不想安装 PyYAML也可以把配置改成 JSON使用标准库json就行。但 YAML 注释能力更好所以下文示例使用 YAML 配置。安装依赖的命令如下pip install pyyaml如果使用的是 Python 3.11也可以通过虚拟环境管理依赖避免污染全局环境python -m venv .venv source .venv/bin/activate # Windows 系统使用 .venv\Scripts\activate pip install pyyaml3.2 项目目录约定为了便于理解我们创建一个file-cleaner-demo目录结构如下file-cleaner-demo/ ├── cleaner.py # 清理工具入口 ├── config.yaml # 清理规则配置 ├── workspace/ │ ├── clean-demo/ │ │ ├── app.log │ │ ├── old.tmp │ │ └── keep.txt其中workspace/clean-demo是测试目录用来模拟真实业务目录。这样设计的好处是我们可以在本地随意验证清理逻辑不需要直接操作系统目录。3.3 配置示例config.yaml内容如下# 清理工具配置文件 targets: - ./workspace/clean-demo rescue_dir: ./workspace/.rescue rules: - name: clear stale tmp files descriptions: 清理超过1小时的临时文件 glob: - *.tmp older_than_days: 0.1 - name: clear old log files descriptions: 清理超过2天的日志文件 glob: - *.log older_than_days: 2 - name: clear oversized log files descriptions: 清理超过5MB的日志文件 glob: - *.log larger_than_mb: 5这里需要注意几点targets是相对路径时建议基于“配置文件所在目录”来解析而不是基于“当前工作目录”来解析。否则在不同目录执行脚本结果可能不一致。rescue_dir是隔离目录不能放在targets内部否则后续扫描时可能会清理掉回收区本身。规则之间是“或”的关系只要命中任意一条规则文件就会进入候选列表。如果你需要更复杂的条件组合可以继续扩展匹配器。4. 核心代码实现为了让文章更接近实战代码不会只贴几个小函数而是给出一个可直接运行的cleaner.py。代码会拆成几个部分讲解方便你理解每一块的职责。4.1 配置数据模型先定义规则数据类。Rule的作用是描述一条清理规则包括文件名称匹配规则文件后缀匹配规则文件年龄阈值文件大小阈值。# cleaner.py 第一部分数据模型 import datetime as dt import fnmatch import json import logging import os import shutil import sys from dataclasses import dataclass, field from pathlib import Path from typing import List, Optional import yaml LOG logging.getLogger(cleaner) class ConfigError(Exception): 配置加载异常 dataclass class Rule: name: str patterns: List[str] field(default_factorylist) extensions: List[str] field(default_factorylist) older_than_days: Optional[float] None larger_than_mb: Optional[float] None classmethod def from_dict(cls, raw: dict) - Rule: name raw.get(name) or unnamed patterns raw.get(glob) or [] extensions raw.get(extensions) or [] if not patterns and not extensions: raise ConfigError(f规则 {name} 缺少 glob 或 extensions 配置) return cls( namename, patternspatterns, extensions[e.lower().lstrip(.) for e in extensions], older_than_daysraw.get(older_than_days), larger_than_mbraw.get(larger_than_mb), ) def match_by_name(self, file_path: Path) - bool: 先判断文件名和后缀是否命中规则。 name_ok not self.patterns or any( fnmatch.fnmatch(file_path.name, pattern) for pattern in self.patterns ) ext_ok not self.extensions or ( file_path.suffix.lower().lstrip(.) in self.extensions ) return name_ok and ext_okmatch_by_name只做静态匹配不做文件状态访问。这样做的好处是在扫描阶段可以先快速排除大部分不相关文件减少无谓的stat调用。4.2 条件判断match_by_stat负责根据文件元信息判断是否超过年龄阈值或者是否超过大小阈值。def match_by_stat(self, file_path: Path, file_stat: os.stat_result) - bool: if self.older_than_days is not None: threshold dt.datetime.now() - dt.timedelta(daysself.older_than_days) mtime dt.datetime.fromtimestamp(file_stat.st_mtime) if mtime threshold: return False if self.larger_than_mb is not None: limit_bytes self.larger_than_mb * 1024 * 1024 if file_stat.st_size limit_bytes: return False return True def is_match(self, file_path: Path, file_stat: os.stat_result) - bool: return self.match_by_name(file_path) and self.match_by_stat(file_path, file_stat)这里要提醒一个细节older_than_days使用的是文件修改时间也就是st_mtime。它比创建时间更接近业务人员的预期。比如临时文件生成了很久但一直没被写入修改时间仍然能表达“这个文件已经很久没有被使用了”。4.3 配置加载接下来把 YAML 配置转换成 Python 对象。为了让相对路径不依赖当前执行目录我们以配置文件所在目录为基准计算目标路径。class CleanConfig: def __init__(self, targets: List[Path], rescue_dir: Path, rules: List[Rule]): self.targets targets self.rescue_dir rescue_dir self.rules rules def load_config(config_path: Path) - CleanConfig: if not config_path.exists(): raise ConfigError(fconfig file not found: {config_path}) with open(config_path, r, encodingutf-8) as f: data yaml.safe_load(f) or {} base_dir config_path.resolve().parent raw_targets data.get(targets) or [] if isinstance(raw_targets, str): raw_targets [raw_targets] targets [] for item in raw_targets: p Path(item).expanduser() if not p.is_absolute(): p base_dir / p p p.resolve() if not p.exists(): raise ConfigError(ftarget directory not found: {p}) if not p.is_dir(): raise ConfigError(ftarget is not directory: {p}) targets.append(p) rescue_path Path(data.get(rescue_dir, ./.rescue)).expanduser() if not rescue_path.is_absolute(): rescue_path base_dir / rescue_path rescue_path rescue_path.resolve() rescue_path.mkdir(parentsTrue, exist_okTrue) raw_rules data.get(rules) or [] if not raw_rules: raise ConfigError(no cleaning rule configured) rules [Rule.from_dict(item) for item in raw_rules] return CleanConfig(targetstargets, rescue_dirrescue_path, rulesrules)路径解析逻辑里有个安全点先对目标目录做resolve()可以获取完整绝对路径方便后续在审计信息中输出。同时也避免 target 使用.或..时产生歧义。4.4 文件扫描扫描目标目录时需要对所有rglob(*)文件做一次遍历并判断哪个规则命中。为了保留文件的完整信息我们可以定义一个FileRecord数据结构。dataclass class FileRecord: file_path: Path file_size: int modified_time: str rule_name: str action: str preview target_path: str def is_under(path: Path, root: Path) - bool: try: path.relative_to(root) return True except ValueError: return False def scan_files(config: CleanConfig) - List[FileRecord]: records [] now dt.datetime.now() for target in config.targets: for file_path in sorted(target.rglob(*)): try: # 跳过符号链接避免错误处理对象 if file_path.is_symlink(): continue if not file_path.is_file(): continue abs_path file_path.resolve() # 隔离区不能再次被扫描 if is_under(abs_path, config.rescue_dir): continue st file_path.stat() for rule in config.rules: if rule.is_match(file_path, st): mtime dt.datetime.fromtimestamp(st.st_mtime) records.append( FileRecord( file_pathabs_path, file_sizest.st_size, modified_timemtime.strftime(%Y-%m-%d %H:%M:%S), rule_namerule.name, ) ) break except PermissionError as exc: LOG.warning(skip path due to permission: %s, reason: %s, file_path, exc) except FileNotFoundError: continue return recordssorted保证扫描顺序稳定。如果后续要复现问题或查看日志看到的结果会比较一致。多个规则冲突时我们只记录第一条命中的规则避免同一个文件重复出现在审计结果里。4.5 动作执行动作执行分为三种模式preview预览模式什么都不动trash移动到隔离目录delete真实删除文件。为了保证安全删除模式在执行前要求手动输入yes确认。生产环境可以结合定时任务的审批流程来做进一步控制。def move_to_rescue(file_record: FileRecord, rescue_root: Path) - Path: src file_record.file_path # 保留原始目标目录名称避免多目录文件重名 parent_dir_name src.parent.name rescue_target rescue_root / parent_dir_name / src.name # 处理同名文件冲突 counter 1 while rescue_target.exists(): rescue_target rescue_root / parent_dir_name / f{src.stem}_{counter}{src.suffix} counter 1 rescue_target.parent.mkdir(parentsTrue, exist_okTrue) shutil.move(str(src), str(rescue_target)) return rescue_target def execute_action(mode: str, records: List[FileRecord], rescue_root: Path) - List[dict]: audit_records [] preview_count 0 moved_count 0 deleted_count 0 failed_count 0 for record in records: audit_item { file_path: str(record.file_path), file_size: record.file_size, modified_time: record.modified_time, rule_name: record.rule_name, action: mode, } if mode preview: preview_count 1 audit_item[target_path] audit_records.append(audit_item) continue try: if mode trash: dest move_to_rescue(record, rescue_root) record.action trash record.target_path str(dest) audit_item[action] trash audit_item[target_path] str(dest) moved_count 1 elif mode delete: record.file_path.unlink(missing_okTrue) record.action delete audit_item[action] delete audit_item[target_path] deleted_count 1 else: raise ValueError(funsupported mode: {mode}) except Exception as exc: failed_count 1 audit_item[action] failed audit_item[error] str(exc) LOG.error(clean file failed: %s, reason: %s, record.file_path, exc) audit_records.append(audit_item) summary { preview_count: preview_count, moved_count: moved_count, deleted_count: deleted_count, failed_count: failed_count, } return audit_records, summary这段代码把“生成记录”和“执行动作”分开。生成记录时没有副作用执行动作时才真正修改文件系统这样便于主程序控制节奏。4.6 日志与审计日志输出到独立logs目录审计记录写入logs/cleaner-audit.jsonl。这里不把审计文件放在目标目录中可以避免清理任务误扫审计日志。def setup_logger(log_dir: Path) - logging.Logger: log_dir.mkdir(parentsTrue, exist_okTrue) log_file log_dir / cleaner.log formatter logging.Formatter( %(asctime)s [%(levelname)s] %(message)s, datefmt%Y-%m-%d %H:%M:%S, ) file_handler logging.FileHandler(log_file, encodingutf-8) file_handler.setFormatter(formatter) console_handler logging.StreamHandler(sys.stdout) console_handler.setFormatter(formatter) LOG.setLevel(logging.INFO) LOG.addHandler(file_handler) LOG.addHandler(console_handler) return LOG def write_audit(records: List[dict], log_dir: Path) - None: audit_file log_dir / cleaner-audit.jsonl with open(audit_file, a, encodingutf-8) as f: for item in records: f.write(json.dumps(item, ensure_asciiFalse)) f.write(\n)这里比较关键的是ensure_asciiFalse。如果清理路径中包含中文或其他非 ASCII 字符JSON 中仍然可读不会变成一串转义字符。4.7 主函数入口最后是主函数。默认模式是preview这样即使用户没有传参也不会误删文件。def parse_args(): parser argparse.ArgumentParser(descriptionFile cleaning tool) parser.add_argument( --config, typestr, defaultconfig.yaml, helpconfig yaml file path, ) parser.add_argument( --mode, typestr, choices[preview, trash, delete], defaultpreview, helpexecution mode, default preview, ) return parser.parse_args() def main(): args parse_args() try: config load_config(Path(args.config)) log_dir config.rescue_dir.parent / logs logger setup_logger(log_dir) logger.info(load config success, targets%s, [str(t) for t in config.targets]) if args.mode delete: answer input( you are about to permanently delete files, type yes to continue: ).strip() if answer ! yes: logger.warning(delete operation canceled by user) return records scan_files(config) logger.info(scan finished, candidate file count%s, len(records)) audit_records, summary execute_action(args.mode, records, config.rescue_dir) write_audit(audit_records, log_dir) logger.info(cleanup summary: %s, summary) except ConfigError as exc: LOG.error(config error: %s, exc) sys.exit(2) except Exception as exc: LOG.exception(unexpected error: %s, exc) sys.exit(1)为了方便阅读代码中日志使用英文占位符实际项目可以根据团队习惯改成中文或统一使用英文。注意input只适合人类登录终端时使用如果是无人值守的定时任务删除前还应经过其他审批或二次校验流程。4.8 补全 imports 和入口保护真正能运行的cleaner.py顶部需要导入argparse文件末尾也要加上入口保护。修正后的顶部区域应包含import argparse import datetime as dt import fnmatch import json import logging import os import shutil import sys from dataclasses import dataclass, field from pathlib import Path from typing import List, Optional import yaml文件末尾需要添加if __name__ __main__: main()如果你复制第 4 节的所有代码可以把它按顺序放在同一个cleaner.py中。也可以拆成config.py、scanner.py、actions.py、audit.py核心思路不变。5. 运行与验证5.1 构造测试文件在项目根目录下执行如下命令准备测试文件mkdir -p workspace/clean-demo echo test workspace/clean-demo/keep.txt echo old tmp file workspace/clean-demo/old.tmp echo old log file workspace/clean-demo/old.log sleep 1 echo new tmp file workspace/clean-demo/new.tmp这里故意保留一个keep.txt目的是验证清理工具不会误删普通文本文件。5.2 执行预览模式先执行预览模式python cleaner.py --config config.yaml --mode preview预期输出类似下面这样2025-01-01 10:00:00 [INFO] load config success, targets[/file-cleaner-demo/workspace/clean-demo] 2025-01-01 10:00:00 [INFO] scan finished, candidate file count3 2025-01-01 10:00:00 [INFO] cleanup summary: {preview_count: 3, moved_count: 0, deleted_count: 0, failed_count: 0}打开workspace/clean-demo你会发现文件都还在。预览模式只打印结果不执行移动或删除操作。5.3 执行隔离模式确认预览没有问题后再执行隔离模式python cleaner.py --config config.yaml --mode trash此时符合规则的文件会被移动到隔离目录。你可以去workspace/.rescue下查看移动结果旧文件依然存在只是不在原目录了。5.4 执行删除模式隔离模式验证完毕如果确认不需要保留再执行删除模式python cleaner.py --config config.yaml --mode delete执行后控制台会要求输入yes。确认后文件才真正从磁盘删除。在真实清理任务中建议先跑一次预览再跑一次隔离观察隔离区文件清单确实无误后再进入删除流程。6. 常见问题与排查思路Python 清理工具在实际使用中经常会遇到下面几类问题这里整理成表格供快速排查。问题现象常见原因解决思路启动时报config file not found当前目录没找到配置文件用绝对路径指定--config或在固定脚本目录执行命令扫描时报PermissionError当前用户对目录没有读权限脚本应跳过无权限文件并记录警告不要中断整批任务清理 Windows 文件时报文件被占用文件正被进程打开系统不允许移动或删除跳过该文件留到业务低峰期重试清理结果比预期多glob 规则写得太宽先开启 preview再看审计记录缩小范围到明确后缀或路径隔离目录被反复扫描rescue_dir 位于 targets 内部把隔离目录放到 targets 之外代码中也要二次跳过清不掉超大日志文件日志进程持续写入文件句柄未释放与业务方协调日志轮转机制而不是直接清理JSON 审计日志中中文变成\u写入 JSON 时ensure_ascii未关闭设置ensure_asciiFalse同一文件被多个规则匹配多条规则间缺少优先级逻辑命中第一条规则后 break避免重复记录面对任何异常首要原则是不要硬编码try-except跳过所有错误。日志里至少应该保留失败路径和失败原因方便后续人工审计。7. 最佳实践与工程建议7.1 默认安全动作清理工具最好把“预览”和“隔离”作为默认动作。永久删除必须显式声明甚至可以在代码层面设置二次确认。这样即使配置文件被写错也不会马上损害业务数据。7.2 禁止扫描边界不明确不要把整个磁盘根目录纳入扫描范围。目标目录越精确清理工具风险越低。如果业务有很多服务器目录建议在配置中逐一列出而不是开放一个总目录后靠排除列表来兜底。7.3 跟踪符号链接语义扫描时跳过符号链接是一个常见安全策略。符号链接可能指向系统文件、其他业务目录或网络目录一旦被当作普通文件遍历可能导致不可预期的清理结果。如果确实需要处理链接目标应该单独设计白名单而不是全局放开。7.4 保留完整审计链条清理任务执行后至少要能回答如下问题谁执行的清理清理了什么目录命中哪些规则文件移动到了哪里是否有失败文件文件删除后是否还能恢复如果这些问题无法回答说明审计日志还不够完整。建议把执行人、执行命令、配置文件 hash、任务编号都能记录到日志形成完整审计链条。7.5 恢复优先于删除在隔离模式下移动文件本质上是“先恢复后清理”。一旦业务方反馈数据丢失运维可以快速查看.rescue目录。实际工程中还可以为隔离目录增加保留天数策略超过保留天数后再自动清理隔离区避免磁盘空间被垃圾回收文件占满。7.6 生产环境变更流程清理工具在正式环境执行前需要经过一套变更流程而不是直接在终端复制粘贴命令。建议步骤是在测试目录准备一批模拟文件执行 preview 模式核对扫描清单执行 trash 模式检查隔离目录确认无误后再进入 delete 模式保留审计日志至少 30 天以上涉及重要系统目录时先做完整备份并验证备份可恢复。7.7 做好定时任务监控如果清理工具通过crontab或运维平台定时执行还需要考虑监控。例如文件失败数超过阈值时发送告警清理后释放空间超过预期时也需要告警因为可能发生了误删。没有监控的清理任务就像没有仪表盘的驾驶执行过程很难被信任。总结与下一步文中这套清理工具核心逻辑不复杂配置化定义清理范围预览模式验证命中文件隔离模式作为可恢复兜底删除模式后才算真正释放磁盘。你可以把它迁移成 Shell 脚本、Go 或 Java 工具但安全设计思路是通用的能预览就不直接删除能隔离就不永久删除能记录审计就不默默清理。接下来你可以继续完善的方向包括支持按磁盘目录大小分组统计、使用目录哈希避免重复移动、将审计记录对接统一日志平台、为删除动作增加操作人审批记录等。清理工具是一个很容易被忽视的小工程但越是容易被忽视的地方越值得保持敬畏。如果你也在维护类似的脚本工具建议先从“默认只预览”做起。 Python安全清理工具开发：从预览到审计的工程实践
如果项目的清理工具模块叫“赵光义”第一眼看到这个名字很多人都会有点出戏。不过它后面跟的另一句话很有价值清理工具“总是分外用心”。这句话听起来不像技术指标却是线上清理工具最容易踩坑的关键点。清理工具做得好不是看它删除速度有多快而是看它能不能做到少误删、可恢复、有记录。本文不写任何复杂的底层框架只讲一类朴素但实用的场景用 Python 实现一个面向日志文件、临时文件的清理工具。工具会包含配置化扫描、预览模式、隔离回收、审计日志这几个能力。代码可以直接复制到本地项目验证也可以按团队实际情况改造成运维小工具。无论你是在做脚本工具还是给业务系统写维护模块这套思路都适用。1. 背景与核心概念1.1 清理工具到底在清理什么清理工具的目标很直接把不再需要的临时文件、过期日志、构建产物、缓存文件从磁盘上释放出来。注意这里说的是“不再需要”而不是“我看着不顺眼”。如果一个工具只维护一份文件后缀列表然后前缀式地删除它就不是清理工具而是“高风险事故制造器”。我在实际项目里见过不少类似需求定时清理应用日志目录里超过 N 天的.log文件清理测试环境中的.tmp、.bak、.old文件限制单个日志文件超过多少 MB 后归档或移除报表服务生成的多余临时文件在任务结束后统一回收。这些需求看似简单实际实现时却会涉及几个共同问题要清理哪些目录、按什么规则判断、哪些文件需要保留、删除前是否需要审计、误删后能不能找回。如果只靠一条rm -rf或os.remove生产环境很容易出问题。1.2 清理工具最容易踩的风险清理工具首先面对的是“误删风险”。文件一旦被物理删除很多场景下是找不回来的尤其是没有版本管理、没有备份、没有审计日志的服务器目录。一个配置项写错、一个通配符写宽就可能导致整批业务文件被清掉。第二个风险是“规则误判”。比如有的项目会在日志目录里放模板文件只是文件名中带着log字样有的缓存文件虽然没有近期修改时间但业务启动时会重新读取。如果清理规则只看文件后缀就会把正在使用的文件移走。第三个风险是“配置影响面过大”。本地开发时目录是相对路径一旦上线后把 target 指向根目录或网络挂载目录扫描范围会成倍放大。实际处理时必须给扫描动作加上边界限制而不是盲目递归。第四个风险是“永久删除”。线上清理解析日志、归档文件后即便文件被删除也应该有一个兜底机制。比较好的做法是「先隔离后清理」先把文件移动到回收区确认无误之后再过期删除。整个过程还要有日志否则出现问题后连排查线索都没有。1.3 本文能帮你掌握什么本文将以一个名为cleaner.py的 Python 清理工具为例带大家实现如下能力通过 YAML 配置维护目标目录和清理规则使用dry-run先预览不实际改动文件使用隔离目录代替直接删除文件自动记录清理前后文件信息与审计日志限制扫描深度、跳过符号链接、避免误删关键目录通过命令行参数控制执行模式。代码重点在思路和可扩展性不依赖数据库、消息队列等重型组件适合中小项目和运维场景快速落地。2. 安全清理工具设计思路写清理工具不能一上来先写unlink方法要先想清楚清理流程里的几个核心概念。2.1 “清理”应该分成几个动作很多清理工具把“删除文件”当作唯一动作这是设计上的简化。更安全的设计应该把清理拆成三个层级第一层是“预览”。扫描文件后只输出哪些文件会命中规则文件大小、修改时间、所在目录是什么但不对文件做任何变更。这个动作适合在深夜任务执行前或者人工运维时先验证配置。第二层是“隔离”。把命中的文件移动到回收目录例如.rescue/20250101。这样文件虽然从原始目录消失了但仍然保留在磁盘上。如果业务发现文件还需要运维可以快速恢复。第三层是“过期清理”。隔离区中的文件在下一次巡检时如果超过保留期限再进行物理删除。这一步应该默认不开启由人工或定时任务显式触发。安全的清理工具默认动作应该是预览或隔离而不是永久删除。本文示例代码会把preview作为默认模式只有手动传入--mode delete时才会执行真实删除。2.2 配置与规则要分离清理规则的维护者不一定是开发清理工具的人。运维人员可能希望只修改 YAML而不是去改 Python 源码。所以一个规范的清理工具至少应该包含两部分内容一是“清理范围”也就是targets。它描述工具允许扫描哪些目录不能扫描哪些目录。二是“清理规则”也就是rules。它描述文件需要满足什么条件例如后缀必须匹配哪些类型、修改时间是否超过 N 天、文件大小是否超过 N MB。把配置和代码分离后新增一条规则只需要修改配置文件不用改动程序逻辑也更容易走配置评审流程。2.3 审计日志必须独立保存清理工具一旦在真实环境执行审计日志和清理动作本身同等重要。审计日志最好包含以下信息任务执行时间匹配到的清理规则名称文件绝对路径文件大小文件修改时间执行动作是预览、移动还是删除移动后的隔离路径任务结果。注意审计日志文件不能放在待清理目录中。如果日志目录本身命中规则可能被清理脚本删掉导致没有日志可查。通常建议把审计日志输出到独立的./audit或服务器集中日志目录并使用 UTF-8 编码保存。3. 环境准备与项目结构3.1 运行环境本文示例代码使用 Python 编写依赖很少Python 3.10 及以上版本PyYAML用于读取 YAML 配置文件操作系统Windows、Linux、macOS 均可不需要额外数据库不需要消息队列。Python 版本可以根据你的实际情况调整。如果你不想安装 PyYAML也可以把配置改成 JSON使用标准库json就行。但 YAML 注释能力更好所以下文示例使用 YAML 配置。安装依赖的命令如下pip install pyyaml如果使用的是 Python 3.11也可以通过虚拟环境管理依赖避免污染全局环境python -m venv .venv source .venv/bin/activate # Windows 系统使用 .venv\Scripts\activate pip install pyyaml3.2 项目目录约定为了便于理解我们创建一个file-cleaner-demo目录结构如下file-cleaner-demo/ ├── cleaner.py # 清理工具入口 ├── config.yaml # 清理规则配置 ├── workspace/ │ ├── clean-demo/ │ │ ├── app.log │ │ ├── old.tmp │ │ └── keep.txt其中workspace/clean-demo是测试目录用来模拟真实业务目录。这样设计的好处是我们可以在本地随意验证清理逻辑不需要直接操作系统目录。3.3 配置示例config.yaml内容如下# 清理工具配置文件 targets: - ./workspace/clean-demo rescue_dir: ./workspace/.rescue rules: - name: clear stale tmp files descriptions: 清理超过1小时的临时文件 glob: - *.tmp older_than_days: 0.1 - name: clear old log files descriptions: 清理超过2天的日志文件 glob: - *.log older_than_days: 2 - name: clear oversized log files descriptions: 清理超过5MB的日志文件 glob: - *.log larger_than_mb: 5这里需要注意几点targets是相对路径时建议基于“配置文件所在目录”来解析而不是基于“当前工作目录”来解析。否则在不同目录执行脚本结果可能不一致。rescue_dir是隔离目录不能放在targets内部否则后续扫描时可能会清理掉回收区本身。规则之间是“或”的关系只要命中任意一条规则文件就会进入候选列表。如果你需要更复杂的条件组合可以继续扩展匹配器。4. 核心代码实现为了让文章更接近实战代码不会只贴几个小函数而是给出一个可直接运行的cleaner.py。代码会拆成几个部分讲解方便你理解每一块的职责。4.1 配置数据模型先定义规则数据类。Rule的作用是描述一条清理规则包括文件名称匹配规则文件后缀匹配规则文件年龄阈值文件大小阈值。# cleaner.py 第一部分数据模型 import datetime as dt import fnmatch import json import logging import os import shutil import sys from dataclasses import dataclass, field from pathlib import Path from typing import List, Optional import yaml LOG logging.getLogger(cleaner) class ConfigError(Exception): 配置加载异常 dataclass class Rule: name: str patterns: List[str] field(default_factorylist) extensions: List[str] field(default_factorylist) older_than_days: Optional[float] None larger_than_mb: Optional[float] None classmethod def from_dict(cls, raw: dict) - Rule: name raw.get(name) or unnamed patterns raw.get(glob) or [] extensions raw.get(extensions) or [] if not patterns and not extensions: raise ConfigError(f规则 {name} 缺少 glob 或 extensions 配置) return cls( namename, patternspatterns, extensions[e.lower().lstrip(.) for e in extensions], older_than_daysraw.get(older_than_days), larger_than_mbraw.get(larger_than_mb), ) def match_by_name(self, file_path: Path) - bool: 先判断文件名和后缀是否命中规则。 name_ok not self.patterns or any( fnmatch.fnmatch(file_path.name, pattern) for pattern in self.patterns ) ext_ok not self.extensions or ( file_path.suffix.lower().lstrip(.) in self.extensions ) return name_ok and ext_okmatch_by_name只做静态匹配不做文件状态访问。这样做的好处是在扫描阶段可以先快速排除大部分不相关文件减少无谓的stat调用。4.2 条件判断match_by_stat负责根据文件元信息判断是否超过年龄阈值或者是否超过大小阈值。def match_by_stat(self, file_path: Path, file_stat: os.stat_result) - bool: if self.older_than_days is not None: threshold dt.datetime.now() - dt.timedelta(daysself.older_than_days) mtime dt.datetime.fromtimestamp(file_stat.st_mtime) if mtime threshold: return False if self.larger_than_mb is not None: limit_bytes self.larger_than_mb * 1024 * 1024 if file_stat.st_size limit_bytes: return False return True def is_match(self, file_path: Path, file_stat: os.stat_result) - bool: return self.match_by_name(file_path) and self.match_by_stat(file_path, file_stat)这里要提醒一个细节older_than_days使用的是文件修改时间也就是st_mtime。它比创建时间更接近业务人员的预期。比如临时文件生成了很久但一直没被写入修改时间仍然能表达“这个文件已经很久没有被使用了”。4.3 配置加载接下来把 YAML 配置转换成 Python 对象。为了让相对路径不依赖当前执行目录我们以配置文件所在目录为基准计算目标路径。class CleanConfig: def __init__(self, targets: List[Path], rescue_dir: Path, rules: List[Rule]): self.targets targets self.rescue_dir rescue_dir self.rules rules def load_config(config_path: Path) - CleanConfig: if not config_path.exists(): raise ConfigError(fconfig file not found: {config_path}) with open(config_path, r, encodingutf-8) as f: data yaml.safe_load(f) or {} base_dir config_path.resolve().parent raw_targets data.get(targets) or [] if isinstance(raw_targets, str): raw_targets [raw_targets] targets [] for item in raw_targets: p Path(item).expanduser() if not p.is_absolute(): p base_dir / p p p.resolve() if not p.exists(): raise ConfigError(ftarget directory not found: {p}) if not p.is_dir(): raise ConfigError(ftarget is not directory: {p}) targets.append(p) rescue_path Path(data.get(rescue_dir, ./.rescue)).expanduser() if not rescue_path.is_absolute(): rescue_path base_dir / rescue_path rescue_path rescue_path.resolve() rescue_path.mkdir(parentsTrue, exist_okTrue) raw_rules data.get(rules) or [] if not raw_rules: raise ConfigError(no cleaning rule configured) rules [Rule.from_dict(item) for item in raw_rules] return CleanConfig(targetstargets, rescue_dirrescue_path, rulesrules)路径解析逻辑里有个安全点先对目标目录做resolve()可以获取完整绝对路径方便后续在审计信息中输出。同时也避免 target 使用.或..时产生歧义。4.4 文件扫描扫描目标目录时需要对所有rglob(*)文件做一次遍历并判断哪个规则命中。为了保留文件的完整信息我们可以定义一个FileRecord数据结构。dataclass class FileRecord: file_path: Path file_size: int modified_time: str rule_name: str action: str preview target_path: str def is_under(path: Path, root: Path) - bool: try: path.relative_to(root) return True except ValueError: return False def scan_files(config: CleanConfig) - List[FileRecord]: records [] now dt.datetime.now() for target in config.targets: for file_path in sorted(target.rglob(*)): try: # 跳过符号链接避免错误处理对象 if file_path.is_symlink(): continue if not file_path.is_file(): continue abs_path file_path.resolve() # 隔离区不能再次被扫描 if is_under(abs_path, config.rescue_dir): continue st file_path.stat() for rule in config.rules: if rule.is_match(file_path, st): mtime dt.datetime.fromtimestamp(st.st_mtime) records.append( FileRecord( file_pathabs_path, file_sizest.st_size, modified_timemtime.strftime(%Y-%m-%d %H:%M:%S), rule_namerule.name, ) ) break except PermissionError as exc: LOG.warning(skip path due to permission: %s, reason: %s, file_path, exc) except FileNotFoundError: continue return recordssorted保证扫描顺序稳定。如果后续要复现问题或查看日志看到的结果会比较一致。多个规则冲突时我们只记录第一条命中的规则避免同一个文件重复出现在审计结果里。4.5 动作执行动作执行分为三种模式preview预览模式什么都不动trash移动到隔离目录delete真实删除文件。为了保证安全删除模式在执行前要求手动输入yes确认。生产环境可以结合定时任务的审批流程来做进一步控制。def move_to_rescue(file_record: FileRecord, rescue_root: Path) - Path: src file_record.file_path # 保留原始目标目录名称避免多目录文件重名 parent_dir_name src.parent.name rescue_target rescue_root / parent_dir_name / src.name # 处理同名文件冲突 counter 1 while rescue_target.exists(): rescue_target rescue_root / parent_dir_name / f{src.stem}_{counter}{src.suffix} counter 1 rescue_target.parent.mkdir(parentsTrue, exist_okTrue) shutil.move(str(src), str(rescue_target)) return rescue_target def execute_action(mode: str, records: List[FileRecord], rescue_root: Path) - List[dict]: audit_records [] preview_count 0 moved_count 0 deleted_count 0 failed_count 0 for record in records: audit_item { file_path: str(record.file_path), file_size: record.file_size, modified_time: record.modified_time, rule_name: record.rule_name, action: mode, } if mode preview: preview_count 1 audit_item[target_path] audit_records.append(audit_item) continue try: if mode trash: dest move_to_rescue(record, rescue_root) record.action trash record.target_path str(dest) audit_item[action] trash audit_item[target_path] str(dest) moved_count 1 elif mode delete: record.file_path.unlink(missing_okTrue) record.action delete audit_item[action] delete audit_item[target_path] deleted_count 1 else: raise ValueError(funsupported mode: {mode}) except Exception as exc: failed_count 1 audit_item[action] failed audit_item[error] str(exc) LOG.error(clean file failed: %s, reason: %s, record.file_path, exc) audit_records.append(audit_item) summary { preview_count: preview_count, moved_count: moved_count, deleted_count: deleted_count, failed_count: failed_count, } return audit_records, summary这段代码把“生成记录”和“执行动作”分开。生成记录时没有副作用执行动作时才真正修改文件系统这样便于主程序控制节奏。4.6 日志与审计日志输出到独立logs目录审计记录写入logs/cleaner-audit.jsonl。这里不把审计文件放在目标目录中可以避免清理任务误扫审计日志。def setup_logger(log_dir: Path) - logging.Logger: log_dir.mkdir(parentsTrue, exist_okTrue) log_file log_dir / cleaner.log formatter logging.Formatter( %(asctime)s [%(levelname)s] %(message)s, datefmt%Y-%m-%d %H:%M:%S, ) file_handler logging.FileHandler(log_file, encodingutf-8) file_handler.setFormatter(formatter) console_handler logging.StreamHandler(sys.stdout) console_handler.setFormatter(formatter) LOG.setLevel(logging.INFO) LOG.addHandler(file_handler) LOG.addHandler(console_handler) return LOG def write_audit(records: List[dict], log_dir: Path) - None: audit_file log_dir / cleaner-audit.jsonl with open(audit_file, a, encodingutf-8) as f: for item in records: f.write(json.dumps(item, ensure_asciiFalse)) f.write(\n)这里比较关键的是ensure_asciiFalse。如果清理路径中包含中文或其他非 ASCII 字符JSON 中仍然可读不会变成一串转义字符。4.7 主函数入口最后是主函数。默认模式是preview这样即使用户没有传参也不会误删文件。def parse_args(): parser argparse.ArgumentParser(descriptionFile cleaning tool) parser.add_argument( --config, typestr, defaultconfig.yaml, helpconfig yaml file path, ) parser.add_argument( --mode, typestr, choices[preview, trash, delete], defaultpreview, helpexecution mode, default preview, ) return parser.parse_args() def main(): args parse_args() try: config load_config(Path(args.config)) log_dir config.rescue_dir.parent / logs logger setup_logger(log_dir) logger.info(load config success, targets%s, [str(t) for t in config.targets]) if args.mode delete: answer input( you are about to permanently delete files, type yes to continue: ).strip() if answer ! yes: logger.warning(delete operation canceled by user) return records scan_files(config) logger.info(scan finished, candidate file count%s, len(records)) audit_records, summary execute_action(args.mode, records, config.rescue_dir) write_audit(audit_records, log_dir) logger.info(cleanup summary: %s, summary) except ConfigError as exc: LOG.error(config error: %s, exc) sys.exit(2) except Exception as exc: LOG.exception(unexpected error: %s, exc) sys.exit(1)为了方便阅读代码中日志使用英文占位符实际项目可以根据团队习惯改成中文或统一使用英文。注意input只适合人类登录终端时使用如果是无人值守的定时任务删除前还应经过其他审批或二次校验流程。4.8 补全 imports 和入口保护真正能运行的cleaner.py顶部需要导入argparse文件末尾也要加上入口保护。修正后的顶部区域应包含import argparse import datetime as dt import fnmatch import json import logging import os import shutil import sys from dataclasses import dataclass, field from pathlib import Path from typing import List, Optional import yaml文件末尾需要添加if __name__ __main__: main()如果你复制第 4 节的所有代码可以把它按顺序放在同一个cleaner.py中。也可以拆成config.py、scanner.py、actions.py、audit.py核心思路不变。5. 运行与验证5.1 构造测试文件在项目根目录下执行如下命令准备测试文件mkdir -p workspace/clean-demo echo test workspace/clean-demo/keep.txt echo old tmp file workspace/clean-demo/old.tmp echo old log file workspace/clean-demo/old.log sleep 1 echo new tmp file workspace/clean-demo/new.tmp这里故意保留一个keep.txt目的是验证清理工具不会误删普通文本文件。5.2 执行预览模式先执行预览模式python cleaner.py --config config.yaml --mode preview预期输出类似下面这样2025-01-01 10:00:00 [INFO] load config success, targets[/file-cleaner-demo/workspace/clean-demo] 2025-01-01 10:00:00 [INFO] scan finished, candidate file count3 2025-01-01 10:00:00 [INFO] cleanup summary: {preview_count: 3, moved_count: 0, deleted_count: 0, failed_count: 0}打开workspace/clean-demo你会发现文件都还在。预览模式只打印结果不执行移动或删除操作。5.3 执行隔离模式确认预览没有问题后再执行隔离模式python cleaner.py --config config.yaml --mode trash此时符合规则的文件会被移动到隔离目录。你可以去workspace/.rescue下查看移动结果旧文件依然存在只是不在原目录了。5.4 执行删除模式隔离模式验证完毕如果确认不需要保留再执行删除模式python cleaner.py --config config.yaml --mode delete执行后控制台会要求输入yes。确认后文件才真正从磁盘删除。在真实清理任务中建议先跑一次预览再跑一次隔离观察隔离区文件清单确实无误后再进入删除流程。6. 常见问题与排查思路Python 清理工具在实际使用中经常会遇到下面几类问题这里整理成表格供快速排查。问题现象常见原因解决思路启动时报config file not found当前目录没找到配置文件用绝对路径指定--config或在固定脚本目录执行命令扫描时报PermissionError当前用户对目录没有读权限脚本应跳过无权限文件并记录警告不要中断整批任务清理 Windows 文件时报文件被占用文件正被进程打开系统不允许移动或删除跳过该文件留到业务低峰期重试清理结果比预期多glob 规则写得太宽先开启 preview再看审计记录缩小范围到明确后缀或路径隔离目录被反复扫描rescue_dir 位于 targets 内部把隔离目录放到 targets 之外代码中也要二次跳过清不掉超大日志文件日志进程持续写入文件句柄未释放与业务方协调日志轮转机制而不是直接清理JSON 审计日志中中文变成\u写入 JSON 时ensure_ascii未关闭设置ensure_asciiFalse同一文件被多个规则匹配多条规则间缺少优先级逻辑命中第一条规则后 break避免重复记录面对任何异常首要原则是不要硬编码try-except跳过所有错误。日志里至少应该保留失败路径和失败原因方便后续人工审计。7. 最佳实践与工程建议7.1 默认安全动作清理工具最好把“预览”和“隔离”作为默认动作。永久删除必须显式声明甚至可以在代码层面设置二次确认。这样即使配置文件被写错也不会马上损害业务数据。7.2 禁止扫描边界不明确不要把整个磁盘根目录纳入扫描范围。目标目录越精确清理工具风险越低。如果业务有很多服务器目录建议在配置中逐一列出而不是开放一个总目录后靠排除列表来兜底。7.3 跟踪符号链接语义扫描时跳过符号链接是一个常见安全策略。符号链接可能指向系统文件、其他业务目录或网络目录一旦被当作普通文件遍历可能导致不可预期的清理结果。如果确实需要处理链接目标应该单独设计白名单而不是全局放开。7.4 保留完整审计链条清理任务执行后至少要能回答如下问题谁执行的清理清理了什么目录命中哪些规则文件移动到了哪里是否有失败文件文件删除后是否还能恢复如果这些问题无法回答说明审计日志还不够完整。建议把执行人、执行命令、配置文件 hash、任务编号都能记录到日志形成完整审计链条。7.5 恢复优先于删除在隔离模式下移动文件本质上是“先恢复后清理”。一旦业务方反馈数据丢失运维可以快速查看.rescue目录。实际工程中还可以为隔离目录增加保留天数策略超过保留天数后再自动清理隔离区避免磁盘空间被垃圾回收文件占满。7.6 生产环境变更流程清理工具在正式环境执行前需要经过一套变更流程而不是直接在终端复制粘贴命令。建议步骤是在测试目录准备一批模拟文件执行 preview 模式核对扫描清单执行 trash 模式检查隔离目录确认无误后再进入 delete 模式保留审计日志至少 30 天以上涉及重要系统目录时先做完整备份并验证备份可恢复。7.7 做好定时任务监控如果清理工具通过crontab或运维平台定时执行还需要考虑监控。例如文件失败数超过阈值时发送告警清理后释放空间超过预期时也需要告警因为可能发生了误删。没有监控的清理任务就像没有仪表盘的驾驶执行过程很难被信任。总结与下一步文中这套清理工具核心逻辑不复杂配置化定义清理范围预览模式验证命中文件隔离模式作为可恢复兜底删除模式后才算真正释放磁盘。你可以把它迁移成 Shell 脚本、Go 或 Java 工具但安全设计思路是通用的能预览就不直接删除能隔离就不永久删除能记录审计就不默默清理。接下来你可以继续完善的方向包括支持按磁盘目录大小分组统计、使用目录哈希避免重复移动、将审计记录对接统一日志平台、为删除动作增加操作人审批记录等。清理工具是一个很容易被忽视的小工程但越是容易被忽视的地方越值得保持敬畏。如果你也在维护类似的脚本工具建议先从“默认只预览”做起。