📚 课程目录第2章监测数据文件批量读取

2. 监测数据文件批量读取

污染源监测数据通常按“企业—排口—监测类型—日期”分散存放在多层文件夹中，一个地市级的季度数据包往往包含数百个 Excel、CSV 和 TXT 文件。如果逐个用 pandas 打开，既无法维护，也容易漏读。批量读取的核心不是“循环打开文件”，而是先建立一份可核查的文件清单，再按格式分派解析器，最后把结果汇总成统一的数据框。本节围绕这条主线展开。

一、先看清目录结构，再写遍历代码

假设示例数据包解压在 D:/pollution_data/ 下，目录组织如下（示例结构，实际项目按当地平台导出规则调整）：

pollution_data/
├─ 2024Q1/
│  ├─ 废气/
│  │  ├─ 企业A_烟囱1_202401.xlsx
│  │  ├─ 企业A_烟囱1_202402.xlsx
│  │  └─ 企业B_烟囱2_202403.csv
│  ├─ 废水/
│  │  ├─ 企业A_总排口_202401.csv
│  │  └─ 企业C_总排口_202402.txt
│  └─ 说明.txt
└─ 2024Q2/
   └─ 废气/
      └─ 企业A_烟囱1_202404.xlsx

这里有两个关键判断：第一，说明.txt 是元数据说明文件，不是监测记录，必须排除；第二，不同子目录下存在同名规则的文件，汇总时需要保留来源路径，否则无法追溯。因此遍历阶段就要把“文件路径”作为一列记录下来，而不是只取文件名。

二、pathlib 遍历与 glob 模式匹配

Python 标准库 pathlib 比 os.listdir 更适合这类任务，它把路径当作对象，支持递归遍历和通配符匹配。下面这段代码完成“遍历 + 按扩展名筛选 + 排除说明文件”三件事：

from pathlib import Path

root = Path("D:/pollution_data")
patterns = ("*.xlsx", "*.csv", "*.txt")
files = []

for pat in patterns:
    for p in root.rglob(pat):          # rglob 递归匹配所有子目录
        if p.name == "说明.txt":        # 排除元数据说明文件
            continue
        if p.is_file():
            files.append({
                "path": str(p),
                "name": p.name,
                "suffix": p.suffix.lower(),
                "parent": p.parent.name,   # 记录所属类别目录，如“废气”
            })

print(len(files))   # 示例数据包中应得到 6 条记录

几个容易出错的点值得强调。rglob 会递归进入所有子目录，如果数据包中混有 ~$ 开头的 Excel 临时文件，需要额外过滤 p.name.startswith("~$")。p.suffix.lower() 统一小写，避免 .XLSX 与 .xlsx 被当成两种格式。另外，rglob("*.txt") 会把说明文件也匹配进来，所以排除逻辑必须写在循环内部，而不是事后靠人工检查。

三、按格式分派解析器

三类文件的读取方式不同：Excel 用 pandas.read_excel，CSV 用 read_csv 并注意编码，TXT 需要先判断是分隔符文本还是定长文本。下面给出一个可复用的分派函数，示例中假设 TXT 为制表符分隔：

import pandas as pd

def read_one(path: str, suffix: str) -> pd.DataFrame:
    if suffix == ".xlsx":
        df = pd.read_excel(path, sheet_name=0)
    elif suffix == ".csv":
        df = pd.read_csv(path, encoding="utf-8-sig")  # 兼容带 BOM 的导出文件
    elif suffix == ".txt":
        df = pd.read_csv(path, sep="\t", encoding="gbk")  # 示例：制表符分隔、GBK 编码
    else:
        raise ValueError(f"不支持的格式: {suffix}")
    df["来源文件"] = path          # 保留溯源信息
    return df

编码是 CSV/TXT 读取中最常见的失败原因。监测平台导出的 CSV 常带 UTF-8 BOM，用 utf-8-sig 可避免首列列名出现 \ufeff；部分老系统导出 GBK 编码的 TXT，直接按 UTF-8 读会抛 UnicodeDecodeError。稳妥做法是先尝试 utf-8-sig，捕获异常后回退到 gbk，并把实际使用的编码记录到日志中，便于后续排查。

四、批量汇总与结果核查

把清单和解析器串起来，就得到完整的批量读取流程。下图展示了从目录遍历到汇总核查的数据流与判断分支：

汇总时用 pd.concat(dfs, ignore_index=True) 纵向拼接。由于各文件列名可能不完全一致（例如有的文件用“排口编号”，有的用“排口代码”），拼接前应先做列名标准化，否则会出现大量 NaN 列。一个实用的核查方式是汇总后按“来源文件”分组统计行数，与清单逐条比对：

import pandas as pd

dfs = [read_one(f["path"], f["suffix"]) for f in files]
all_df = pd.concat(dfs, ignore_index=True)

check = all_df.groupby("来源文件").size().reset_index(name="行数")
print(check)
print("清单文件数:", len(files), " 成功读取:", check.shape[0])

如果两者数量不一致，说明某个文件读取失败或返回空表，需要回到清单定位。示例数据包中 6 个文件应全部出现在 check 中；若某个 CSV 因编码问题抛异常，应在 read_one 外层用 try/except 捕获并记录失败路径，而不是让整个批处理中断。

建议把“文件清单”先导出为一份 CSV（含 path、suffix、parent、读取状态、行数、编码），作为本次批处理的审计记录。后续清洗阶段发现异常值时，可以凭“来源文件”列快速回溯到原始文件，这对污染源数据的可追溯性要求尤为重要。

五、常见问题与判断依据

批量读取阶段最典型的失败有三类：一是路径中含中文或空格导致某些旧库读取失败，pathlib 与 pandas 组合一般可正常处理，但需确保路径字符串未被错误转义；二是 Excel 文件存在多个工作表，sheet_name=0 只读第一个，若监测数据在第二个工作表，需要改为按表名读取或遍历全部工作表；三是空文件或只有表头的文件，读取后行数为 0，应在核查表中标记为“空表”而非“失败”，二者处理方式不同。判断标准很直接：清单文件数、成功读取数、总行数三者能对上，且每个来源文件都有明确状态，才算批量读取环节完成。