📚 课程目录第4章文件系统遍历与批量输入发现

4. 文件系统遍历与批量输入发现

批处理脚本的第一道工序，往往不是计算，而是“找文件”。如果输入文件靠手工复制路径，脚本就失去了批量价值；如果扫描规则写得太宽，又会把日志、临时文件、备份文件一并卷进来，导致后续处理报错或结果污染。本节要解决的核心问题是：用 pathlib 和 glob 递归扫描数据目录，按明确规则筛选出真正待处理的文件，并生成一份可复核、可复用的任务清单。

先明确一个基本认识：目录遍历得到的是“候选集”，任务清单才是“待处理集”。两者之间必须经过筛选规则这一层。候选集可能包含子目录、隐藏文件、扩展名不符的文件、命名不合规的文件、正在写入的临时文件。筛选规则的作用是把这些噪声剔除，并让筛选条件以代码形式固定下来，而不是每次靠人眼判断。

用 pathlib 建立目录遍历骨架

pathlib 把路径抽象为 Path 对象，比字符串拼接更安全，也更容易表达“递归”和“筛选”。最常用的两个遍历入口是 Path.rglob() 和 Path.glob()。rglob("*.csv") 会从当前目录开始递归匹配所有层级的 CSV 文件；glob("*") 只匹配当前层。对于数据目录通常按日期或批次分层的场景，rglob 是更自然的选择。

from pathlib import Path

DATA_ROOT = Path("/data/experiments")

# 递归收集所有 CSV 候选文件
candidates = sorted(DATA_ROOT.rglob("*.csv"))
print(f"候选文件数: {len(candidates)}")
for p in candidates[:5]:
    print(p.relative_to(DATA_ROOT))

这里有两个细节值得注意。第一，rglob 返回的是生成器，直接 len() 会报错，需要先转成列表或排序；排序的目的是让任务清单顺序稳定，便于复现和比对。第二，打印时用 relative_to(DATA_ROOT) 输出相对路径，清单更简洁，也避免把机器相关的绝对路径写死到日志里。

另一个常见需求是同时匹配多种扩展名。可以多次调用 rglob 再合并，也可以用 glob 的字符类写法。例如 rglob("*.[Cc][Ss][Vv]") 能兼容大小写扩展名。但更推荐的做法是先用较宽的候选模式收集，再在筛选阶段统一判断，这样规则集中、易读、易改。

筛选规则：把“什么算待处理”写清楚

筛选规则通常由几类条件组合而成：扩展名白名单、文件名模式、目录排除、文件状态检查。下面是一份可直接参考的筛选函数，它把规则拆成独立判断，便于逐条调整。

import re
from pathlib import Path

DATA_ROOT = Path("/data/experiments")
EXCLUDE_DIRS = {"archive", "tmp", ".cache"}
NAME_PATTERN = re.compile(r"^run_(?P<batch>\d{4})_(?P<sample>[a-z0-9]+)\.csv$")

def is_task_file(p: Path) -> bool:
    # 1. 必须是普通文件，排除目录和符号链接指向的目录
    if not p.is_file():
        return False
    # 2. 路径中任一层命中排除目录则跳过
    if any(part in EXCLUDE_DIRS for part in p.parts):
        return False
    # 3. 文件名必须符合命名规范
    if not NAME_PATTERN.match(p.name):
        return False
    # 4. 跳过空文件和正在写入的临时文件
    if p.stat().st_size == 0:
        return False
    if p.suffix == ".tmp":
        return False
    return True

tasks = [p for p in sorted(DATA_ROOT.rglob("*.csv")) if is_task_file(p)]

这段代码体现了几个工程上的取舍。用 p.parts 判断排除目录，比字符串包含判断更可靠，不会因为文件名里恰好出现 “tmp” 而误伤。用正则提取批次号和样本号，既完成筛选，又为后续分组处理预留了结构化信息。检查 st_size == 0 是为了跳过尚未写完或写入失败的空文件，这是批处理中最常见的“假输入”之一。

如果数据目录可能被其他进程同时写入，仅靠扩展名和大小检查仍可能读到半截文件。更稳妥的做法是要求上游先写临时文件再原子重命名，或在扫描时跳过最近若干分钟内修改过的文件。规则要与数据生产方的约定保持一致。

从文件列表到任务清单

筛选出的 Path 列表还只是路径集合。真正的任务清单应包含执行所需的关键字段：任务编号、输入路径、批次、样本、文件大小、修改时间，以及一个初始状态。把这些字段组织成字典或数据类，再序列化为 CSV 或 JSON，后续调度、断点续跑、结果核对都依赖它。

import csv
from datetime import datetime

rows = []
for idx, p in enumerate(tasks, start=1):
    m = NAME_PATTERN.match(p.name)
    st = p.stat()
    rows.append({
        "task_id": f"T{idx:05d}",
        "input_path": str(p),
        "batch": m.group("batch"),
        "sample": m.group("sample"),
        "size_bytes": st.st_size,
        "mtime": datetime.fromtimestamp(st.st_mtime).isoformat(timespec="seconds"),
        "status": "pending",
    })

with open("task_manifest.csv", "w", newline="", encoding="utf-8") as f:
    writer = csv.DictWriter(f, fieldnames=rows[0].keys())
    writer.writeheader()
    writer.writerows(rows)

清单生成后不要急着开跑，先做一次快速核对：任务总数是否与预期量级一致，批次分布是否合理，是否存在同一输入被重复收录，路径是否都在数据根目录之下。这些检查可以用几行代码完成，却能拦住大部分“扫错目录”“规则写宽”的问题。

上图把本节的关键关系压缩在一张图里：左侧是真实目录结构，中间是四条筛选规则，右侧是最终清单。可以看到 archive/ 下的历史文件、tmp/ 下的临时文件、以及 notes.txt 都在筛选阶段被排除，只有符合命名规范且状态正常的 CSV 进入清单。这种“候选—规则—清单”的三段结构，比把所有判断塞进一个循环更利于维护和复查。

几个容易踩的坑

第一，rglob 会跟随目录层级，但不会自动跳过隐藏目录。如果数据目录里存在 .git 或 .cache，务必在排除规则中显式处理。第二，符号链接可能导致重复收录或循环遍历，必要时用 p.is_symlink() 判断并跳过。第三，路径大小写在 Windows 与 Linux 上行为不同，扩展名匹配尽量统一转小写再比较。第四，清单中的路径建议保留绝对路径或相对于固定根目录的路径，不要混用，否则后续任务在不同工作目录下执行会找不到文件。

最后，把扫描与筛选封装成可重复调用的函数，并让规则参数化，是批处理脚本走向工程化的第一步。目录变了、命名规范变了、排除项变了，只需调整参数，而不是重写遍历逻辑。任务清单一旦稳定生成，后续的并行调度、失败重试、结果汇总才有可靠的输入基础。