4. 环保数据表结构识别与批量合并

环保监测数据在汇总阶段最常见的困境，不是缺少数据，而是“数据长得不一样”。同一个空气质量自动站，运维单位A导出的CSV第一行是标题“XX市2024年1月日报”，第二行才是字段名；运维单位B的Excel前两行是合并单元格的说明，第三行才是表头；而省平台导出的文件干脆没有表头，列顺序固定但列名缺失。再加上“PM2.5”“pm25”“细颗粒物(μg/m³)”这类同义不同名的列，以及mg/m³与μg/m³混用，直接纵向拼接必然错列、错单位。本节要解决的就是：让脚本先“看懂”每张表的结构，再按统一规则合并成一张宽表。

一、表头行识别：先定位真正的字段行

表头行识别的本质是“在一张表的前若干行中，找出最像字段名的那一行”。工程上不必上复杂模型，用三条启发式规则组合即可覆盖绝大多数环保数据文件：

• 非空率：字段行通常整行非空，而标题行往往只有第一个单元格有值，其余为空。
• 唯一性：字段名在同一行内不应重复，重复往往意味着这是数据行或合并单元格残留。
• 类型倾向：字段行中文本型单元格占比高，而数据行中数值型占比高。可对前10行逐行统计“能转成数字的单元格比例”，取该比例首次明显下降的行作为表头候选。

下面是一段可直接复用的表头探测函数，返回表头所在行号（0基）与列名列表：

import pandas as pd, re

def detect_header(path, max_scan=10):
    raw = pd.read_csv(path, header=None, nrows=max_scan, dtype=str)
    best_row, best_score = 0, -1
    for i in range(len(raw)):
        row = raw.iloc[i].dropna().astype(str).str.strip()
        row = row[row != ""]
        if len(row) == 0:
            continue
        non_empty = len(row) / raw.shape[1]          # 非空率
        unique = row.nunique() / len(row)            # 唯一性
        numeric_ratio = row.str.match(r"^-?\d+(\.\d+)?$").mean()
        score = non_empty * 0.5 + unique * 0.3 + (1 - numeric_ratio) * 0.2
        if score > best_score:
            best_row, best_score = i, score
    return best_row

# 示例：某文件探测结果为第2行（0基），即第3行是表头
# 读取时跳过前2行，并把该行作为列名
df = pd.read_csv("station_a_202401.csv", header=detect_header("station_a_202401.csv"))

对Excel文件，把 read_csv 换成 read_excel(..., header=None) 即可，逻辑完全一致。需要注意：合并单元格在pandas中会表现为“首格有值、其余为NaN”，因此非空率规则对这类文件尤其有效。若探测结果与人工判断不符，应把该文件加入“人工复核清单”，而不是强行套用。

二、列名映射：把同义列收敛到标准字段

识别出表头后，下一步是把各文件的列名映射到一套标准字段字典。环保数据建议先定义标准列，例如：站点编号、监测时间、PM2.5、PM10、SO2、NO2、CO、O3、温度、湿度，并明确每个字段的单位。映射采用“别名表 + 规范化”两级策略：先做字符串规范化（去空格、去括号内容、转小写、全角转半角），再查别名表。

import unicodedata

ALIAS = {
    "pm2.5": "PM2.5", "pm25": "PM2.5", "细颗粒物": "PM2.5",
    "pm10": "PM10", "可吸入颗粒物": "PM10",
    "so2": "SO2", "二氧化硫": "SO2",
    "no2": "NO2", "二氧化氮": "NO2",
    "co": "CO", "一氧化碳": "CO",
    "o3": "O3", "臭氧": "O3",
    "站点编号": "站点编号", "站点": "站点编号", "site_id": "站点编号",
    "监测时间": "监测时间", "时间": "监测时间", "datetime": "监测时间",
}

def norm(name):
    s = unicodedata.normalize("NFKC", str(name)).strip().lower()
    s = re.sub(r"[（(].*?[)）]", "", s)   # 去掉单位括号
    s = re.sub(r"\s+", "", s)
    return s

def map_columns(cols):
    out, unmapped = {}, []
    for c in cols:
        key = norm(c)
        if key in ALIAS:
            out[c] = ALIAS[key]
        else:
            unmapped.append(c)
    return out, unmapped

映射完成后务必打印 unmapped 列表。未映射列通常有两类：一类是确实无关的辅助列（如“备注”“录入人”），可直接丢弃；另一类是别名表没覆盖到的新写法，应补充进 ALIAS 后重跑。这个“未映射清单”就是映射规则持续演进的入口。

三、单位统一：在数值层面做换算

单位不统一是环保数据合并中最隐蔽的错误来源。常见情况是气体浓度有的用μg/m³、有的用mg/m³，CO甚至有用ppm的。处理原则是：先确定每个源列的单位，再统一换算到标准单位。单位信息可能藏在列名括号里（如“SO2(mg/m³)”），也可能藏在文件说明行里，还可能完全没有——此时必须依据来源约定人工确认，不能猜。

换算关系要写死在配置中，避免口算出错：1 mg/m³ = 1000 μg/m³；CO在标准状态下 1 ppm ≈ 1.145 mg/m³（20℃、101.325 kPa，仅作示例换算系数，实际应依据监测标准规定的参比状态取值）。换算时只对数值列操作，且要先 pd.to_numeric(errors="coerce")，把“--”“ND”“未检出”等替换为缺失值，再乘系数。

四、批量合并为统一宽表

把上述三步串成流水线：遍历目录下所有CSV/Excel → 探测表头 → 读取 → 列名映射 → 单位换算 → 补充来源标识列 → 纵向拼接 → 按“站点编号+监测时间”去重排序 → 输出宽表。下面的结构图展示了这一数据流与关键判断点：

拼接时用 pd.concat(dfs, ignore_index=True, sort=False)，sort=False 可避免因列顺序不同产生大量NaN列。合并后立即做三项校验：行数是否等于各文件有效行数之和（允许去重后减少）；关键字段缺失率是否在可接受范围；单位换算后的数值量级是否合理（例如PM2.5日均值应在个位数到数百μg/m³之间，若出现0.05这类值，多半是mg/m³未换算）。

五、一份可执行的合并配置示例

把易变的部分抽成配置，脚本只负责执行，便于非编程人员维护：

完成以上流程后，得到的宽表才具备跨来源可比性。后续的缺失值插补、异常值识别、统计汇总都应建立在这张结构统一、单位一致的表上，而不是在拼接阶段“先合了再说”。结构识别与单位统一看似是准备工作，实际上决定了整条清洗链路的上限。