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

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

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")

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}")

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

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")

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)