CSV多列数据校验及错误列生成需求与awk实现疑问

下面提供两种简便的实现方式，分别适合命令行快速处理和需要灵活调整规则的场景：

一、Awk命令行实现（无需额外依赖） #

这个方案通过两次遍历CSV文件完成校验：第一次收集每个FNAME对应的XTRUC值，第二次逐行应用所有校验规则并生成错误列。

编写Awk脚本 #

创建名为validate_csv.awk的文件，内容如下：

BEGIN {
    FS = ","; OFS = ","
    # 如果原CSV有自定义表头，取消注释并替换为你的表头行
    # print "COL1,COL2,FNAME,XTRUC,XIZE,XIZE2,ORG,ORG2,ERROR"
}

# 第一次遍历：收集每个FNAME对应的XTRUC信息
NR == FNR {
    # 跳过表头（如果有）
    if (NR == 1) next
    fname = $3
    xtruc = $4
    if (fname != "") {
        if (!(fname in xtruc_map)) {
            xtruc_map[fname] = xtruc
        } else if (xtruc_map[fname] != xtruc) {
            # 标记该FNAME存在多个不同的XTRUC
            xtruc_map[fname] = "MULTIPLE"
        }
    }
    next
}

# 第二次遍历：逐行校验并生成错误信息
{
    error = ""
    # 规则1：第4列包含UNKNOWN
    if ($4 ~ /UNKNOWN/) {
        error = "XTRUC is UNKNOWN"
    }
    # 规则2：同一FNAME对应不同XTRUC
    fname = $3
    if (fname != "" && xtruc_map[fname] == "MULTIPLE") {
        error = error (error != "" ? "\n" : "") "multiple XTRUC value for the same FNAME"
    }
    # 规则3：XIZE和XIZE2非数字或值不匹配
    xize = $5
    xize2 = $6
    is_xize_num = xize ~ /^[0-9]+(\.[0-9]+)?$/
    is_xize2_num = xize2 ~ /^[0-9]+(\.[0-9]+)?$/
    if (!is_xize_num || !is_xize2_num) {
        error = error (error != "" ? "\n" : "") "XIZE or XIZE2 is not a number"
    } else if (xize != xize2) {
        error = error (error != "" ? "\n" : "") "XIZE and XIZE2 don't match"
    }
    # 规则4：ORG和ORG2不匹配
    if ($7 != $8) {
        error = error (error != "" ? "\n" : "") "ORG and ORG2 don't match"
    }
    # 输出该行及错误列
    print $0, error
}

执行命令 #

如果原CSV有表头，直接运行：

awk -f validate_csv.awk file.csv file.csv > file_with_errors.csv

注：两次传入file.csv是为了完成两次遍历，第一次收集数据，第二次校验输出。

二、Python Pandas实现（可读性高，易扩展） #

如果不熟悉Awk，用Pandas可以更直观地实现规则，适合后续需要调整校验逻辑的场景。

代码示例 #

import pandas as pd

# 读取CSV文件（如果表头不是默认格式，可指定header参数）
df = pd.read_csv("file.csv")

# 初始化错误列
df["ERROR"] = ""

# 规则1：第4列包含UNKNOWN
mask_unknown = df.iloc[:, 3].str.contains("UNKNOWN", na=False)
df.loc[mask_unknown, "ERROR"] += "XTRUC is UNKNOWN\n"

# 规则2：同一FNAME对应不同XTRUC
# 统计每个FNAME的XTRUC唯一值数量
xtruc_unique_counts = df.groupby("FNAME")["XTRUC"].nunique()
# 筛选出存在多个XTRUC的FNAME
multiple_xtruc_fnames = xtruc_unique_counts[xtruc_unique_counts > 1].index
mask_multiple_xtruc = df["FNAME"].isin(multiple_xtruc_fnames)
df.loc[mask_multiple_xtruc, "ERROR"] += "multiple XTRUC value for the same FNAME\n"

# 规则3：XIZE和XIZE2非数字或值不匹配
def check_numeric(val):
    try:
        float(val)
        return True
    except (ValueError, TypeError):
        return False

# 检查是否为数字
mask_not_numeric = ~df["XIZE"].apply(check_numeric) | ~df["XIZE2"].apply(check_numeric)
df.loc[mask_not_numeric, "ERROR"] += "XIZE or XIZE2 is not a number\n"

# 检查值不匹配（排除非数字的行）
mask_xize_mismatch = (df["XIZE"] != df["XIZE2"]) & ~mask_not_numeric
df.loc[mask_xize_mismatch, "ERROR"] += "XIZE and XIZE2 don't match\n"

# 规则4：ORG和ORG2不匹配
mask_org_mismatch = df["ORG"] != df["ORG2"]
df.loc[mask_org_mismatch, "ERROR"] += "ORG and ORG2 don't match\n"

# 去除错误列末尾的换行符
df["ERROR"] = df["ERROR"].str.rstrip("\n")

# 保存结果到新CSV
df.to_csv("file_with_errors.csv", index=False)

执行方式 #

确保已安装Pandas：

pip install pandas

然后运行脚本即可生成带错误列的CSV文件。

内容的提问来源于stack exchange，提问作者TenEM