Python中线上目录遍历的防坑指南

作者：袖梨 2026-06-06

前言

做服务器运维、日志归档、磁盘巡检、过期文件清理时，几乎人人都要写目录遍历脚本。很多新手随手os.listdir一层遍历，本地测试几个文件跑得飞快，一丢线上环境直接翻车：权限报错中断程序、软链接无限递归死循环、海量小文件撑爆内存、文件中途被删除触发异常崩溃。

线上目录远比本地测试环境复杂：数十万细碎业务文件、带空格/中文/特殊符号的文件名、无权限的缓存目录、指向父级的软链接、业务进程实时增删文件，一套能扛住生产环境的遍历代码，核心从来不是实现递归，而是兜底各类异常、规避环境陷阱。

一、新手通用踩坑写法：单层listdir，线上完全不堪用

最经典的入门遍历写法，仅能读取同级目录，既无法递归深层目录，也没有任何异常捕获，碰到权限目录脚本直接PermissionError终止，也就是开篇日志卡在.cache权限目录的根源。

import os
root = "/data/upload"
for name in os.listdir(root):
    path = os.path.join(root, name)
    print(path)

适用场景仅限本地临时查看目录，严禁用于生产文件统计、过期清理、磁盘巡检。一旦遇到子目录、权限隔离目录、瞬时删除的目录，程序直接异常退出，前面扫描成果全部作废。

二、生产优选：os.scandir替代listdir，精准区分文件与目录

os.scandir相比os.listdir优势巨大：返回DirEntry对象，自带文件类型属性，无需额外os.path.isdir/isfile重复系统IO，遍历效率更高；搭配follow_symlinks=False屏蔽软链接跟随，从根源杜绝软链接指向父目录造成的无限递归。

单层安全遍历模板，提前校验路径是否存在、是否为目录，非法路径直接日志跳过：

from pathlib import Path
import os

def scan_one_level(folder: str):
    base = Path(folder)
    if not base.exists():
        print(f"[bad_path] not exists: {base}")
        return
    if not base.is_dir():
        print(f"[bad_path] not dir: {base}")
        return
    # 上下文管理器自动释放资源
    with os.scandir(base) as items:
        for item in items:
            # 禁止跟随软链接，防止死循环
            if item.is_dir(follow_symlinks=False):
                print(f"[dir ] {item.path}")
            elif item.is_file(follow_symlinks=False):
                print(f"[file] {item.path}")
            else:
                print(f"[skip] {item.path}")

scan_one_level("/data/upload")

线上坑点总结：不加follow_symlinks=False，碰到环形软链接，脚本会无限递归，拉高磁盘IO占用，拖垮服务器。

三、自定义栈式递归遍历：可控性完胜原生os.walk

原生os.walk可以实现递归遍历，但自定义栈结构能灵活控制遍历深度、拦截异常、按需跳过目录，是生产环境首选方案。

核心设计思路：用栈保存(目录路径, 当前层级)，循环出栈遍历，通过max_depth限制递归深度，逐层捕获权限不足、目录已删除、读取失败等所有异常，只打印错误日志，不终止主程序。

完整可复用生成器代码：

from pathlib import Path
import os

def walk_files(root: str, max_depth: int = 20):
    start = Path(root).resolve()
    stack = [(start, 0)]
    while stack:
        folder, depth = stack.pop()
        # 超出设定深度直接跳过
        if depth > max_depth:
            print(f"[skip_depth] {folder}")
            continue
        try:
            with os.scandir(folder) as entries:
                for entry in entries:
                    path = Path(entry.path)
                    try:
                        # 目录入栈实现递归，不跟随软链接
                        if entry.is_dir(follow_symlinks=False):
                            stack.append((path, depth + 1))
                            continue
                        # 文件通过yield逐个返回，惰性加载不占内存
                        if entry.is_file(follow_symlinks=False):
                            yield path
                    except OSError as e:
                        print(f"[entry_error] path={path} err={e}")
        except PermissionError as e:
            # 无权限目录记录日志，继续遍历其他目录
            print(f"[no_permission] path={folder} err={e}")
        except FileNotFoundError:
            # 遍历瞬间目录被业务删除，常见于动态上传目录
            print(f"[gone] path={folder}")
        except OSError as e:
            print(f"[scan_error] path={folder} err={e}")

关键设计亮点

yield惰性迭代：逐个产出文件路径，不用一次性把全量文件存入列表，海量小文件场景避免内存溢出，脚本意外中断，已扫描数据已有输出；
全链路异常捕获：区分权限不足、目录消失、读取异常三类生产高频报错，异常仅日志记录，不中断全量扫描；
深度限制：max_depth防止意外遍历系统根目录无限下沉，规避扫描全磁盘带来的IO风暴。

调用示例：

for file_path in walk_files("/data/upload"):
    print(file_path)

四、基于通用遍历器，封装四大生产高频业务脚本

依托walk_files生成器，可快速封装超大文件排查、后缀筛选、磁盘占用统计、过期文件清理四类运维常用工具，所有脚本延续异常捕获、安全校验逻辑。

扫描超大文件：排查磁盘异常暴涨

自动筛选超过指定MB的文件，捕获文件无法获取大小的异常：

def find_big_files(root: str, limit_mb: int = 100):
    limit = limit_mb * 1024 * 1024
    for path in walk_files(root):
        try:
            size = path.stat().st_size
        except OSError as e:
            print(f"[stat_error] path={path} err={e}")
            continue
        if size >= limit:
            print(f"[big_file] size_mb={size / 1024 / 1024:.1f} path={path}")

find_big_files("/data/upload", 100)

按后缀过滤文件：日志、临时文件定向筛选

统一小写后缀匹配，规避大小写后缀漏匹配问题：

def iter_by_suffix(root: str, suffixes: set[str]):
    suffixes = {s.lower() for s in suffixes}
    for path in walk_files(root):
        if path.suffix.lower() in suffixes:
            yield path

# 只遍历.log .txt文件
for log_file in iter_by_suffix("/data/app", {".log", ".txt"}):
    print(f"[match] {log_file}")

目录磁盘占用汇总：快速定位耗空间文件类型

按文件后缀汇总数量与占用空间，从大到小排序，排查磁盘爆满神器：

from collections import defaultdict

def summary_folder(root: str):
    counter = defaultdict(int)
    size_map = defaultdict(int)
    for path in walk_files(root):
        suffix = path.suffix.lower() or "<no_ext>"
        try:
            size = path.stat().st_size
        except OSError as e:
            print(f"[stat_error] path={path} err={e}")
            continue
        counter[suffix] += 1
        size_map[suffix] += size
    # 按占用空间倒序输出
    rows = sorted(size_map.items(), key=lambda x: x[1], reverse=True)
    for suffix, total_size in rows:
        print(f"{suffix:10s} count={counter[suffix]:6d} size_mb={total_size / 1024 / 1024:10.2f}")

summary_folder("/data/upload")

过期临时文件清理：dry_run试运行杜绝误删

生产清理铁律：先试运行、后真实删除，通过dry_run开关控制，试运行仅打印待删文件，确认无误再关闭试运行执行删除，避免路径传参错误、逻辑bug导致批量误删业务文件。

import time
def delete_old_tmp(root: str, days: int = 30, dry_run: bool = True):
    now = time.time()
    expire_seconds = days * 24 * 3600
    for path in iter_by_suffix(root, {".tmp", ".bak"}):
        try:
            mtime = path.stat().st_mtime
        except OSError as e:
            print(f"[stat_error] path={path} err={e}")
            continue
        if now - mtime <= expire_seconds:
            continue
        if dry_run:
            print(f"[dry_run] delete {path}")
            continue
        try:
            path.unlink()
            print(f"[deleted] {path}")
        except OSError as e:
            print(f"[delete_error] path={path} err={e}")
# 先试运行，确认输出无误再改为False
delete_old_tmp("/data/upload/tmp", 30, dry_run=True)