如何在Python2.7的AWS Lambda中处理S3 CSV表头的\r字符并验证

你遇到的问题其实有两个核心点：一是CSV每行残留的\r导致表头不匹配，二是环境变量里的表头字符串需要正确转换成列表才能做比较。我来一步步帮你搞定：

为什么会出现\r？ #

你的CSV文件应该是在Windows系统下生成的，Windows的换行符是\r\n，而Lambda运行在Linux环境中，Linux只用\n当换行符。当你用split('\n')分割内容时，每行末尾的\r会被保留下来，所以表头最后一项就带上了\r。

解决方案1：手动去除\r #

在分割每行之前，先用rstrip('\r')去掉末尾的回车符，这样分割后的表头就干净了：

def validateCSV(event,context):
    import os
    import boto3
    # 把环境变量的字符串转换成列表，否则直接比较会失败
    import ast
    EXPECTED_HEADERS = ast.literal_eval(os.environ['RM_EXPECTED_HEADERS'])
    
    s3 = boto3.client("s3")
    file_obj = event["Records"][0]
    bucketname = str(file_obj['s3']['bucket']['name'])
    filename = str(file_obj['s3']['object']['key'])
    fileObj = s3.get_object(Bucket=bucketname, Key=filename)
    rows = fileObj["Body"].read().split('\n')
    
    fList=[]
    for line in rows:
        # 先去掉末尾的\r，再分割
        cleaned_line = line.rstrip('\r')
        if cleaned_line:  # 跳过空行，避免最后一行空数据干扰
            fList.append(cleaned_line.split(','))
    
    print("fList Headers matched: ", fList[0] == EXPECTED_HEADERS)

解决方案2：用Python内置csv模块读取（更推荐） #

手动处理CSV容易踩各种格式坑（比如带引号的字段、不同系统的换行符），用Python标准库的csv模块可以自动处理这些问题，包括自动去除多余的\r：

def validateCSV(event,context):
    import os
    import boto3
    import csv
    from io import StringIO
    import ast
    
    EXPECTED_HEADERS = ast.literal_eval(os.environ['RM_EXPECTED_HEADERS'])
    s3 = boto3.client("s3")
    file_obj = event["Records"][0]
    bucketname = str(file_obj['s3']['bucket']['name'])
    filename = str(file_obj['s3']['object']['key'])
    
    fileObj = s3.get_object(Bucket=bucketname, Key=filename)
    # 读取内容并解码为字符串
    csv_content = fileObj["Body"].read().decode('utf-8')
    # 用csv.reader自动解析CSV格式，处理换行符和分隔符
    reader = csv.reader(StringIO(csv_content))
    rows = list(reader)
    
    # 跳过开头可能的空行
    while rows and not rows[0]:
        rows.pop(0)
    
    print("Headers matched: ", rows[0] == EXPECTED_HEADERS)

关于环境变量的关键注意点 #

你在环境变量里设置的RM_EXPECTED_HEADERS值是['Name','Age','PinCode']，但os.environ读出来的是字符串，不是列表！直接用列表fList[0]和字符串比较肯定会返回False，所以必须用ast.literal_eval()把字符串转换成真实的列表。

如果不想用ast模块，也可以手动解析：

header_str = os.environ['RM_EXPECTED_HEADERS'].strip('[]').replace("'", "").split(',')
EXPECTED_HEADERS = [h.strip() for h in header_str]

这样处理后，表头验证就能正常通过了！

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