如何用Python读取CSV判断流程编码以跳过/触发清理流程？

以下是修复后的实现方案，同时梳理了常见的错误原因：

修复后的代码（csv模块版，轻量高效） #

import csv

def complete_stage_purge_process(process_cd):
    skip_processes = set()
    # 读取CSV并提取需跳过的流程编码
    try:
        with open('test.csv', 'r', encoding='utf-8') as csvfile:
            # 假设CSV表头为'process_cd'，若实际列名不同请修改
            reader = csv.DictReader(csvfile)
            for row in reader:
                # 提取并清理编码，过滤空值
                cd = row.get('process_cd', '').strip()
                if cd:
                    skip_processes.add(cd)
    except FileNotFoundError:
        print("错误：test.csv文件不存在")
        return

    # 核心判断逻辑
    if process_cd.strip() in skip_processes:
        print(f"流程编码{process_cd}在跳过列表中，跳过清理")
        return
    
    # 执行清理流程（替换为你的DELETE语句）
    print(f"启动流程{process_cd}的清理操作")
    # cursor.execute("DELETE FROM your_table WHERE process_cd = %s", (process_cd.strip(),))
    # conn.commit()

备选方案（pandas版，适合大数据量快速处理） #

import pandas as pd

def complete_stage_purge_process(process_cd):
    skip_processes = set()
    try:
        # 仅读取目标列，跳过空行
        df = pd.read_csv('test.csv', usecols=['process_cd'], skip_blank_lines=True)
        # 清理空值并转成集合
        skip_processes = set(df['process_cd'].dropna().astype(str).str.strip())
    except FileNotFoundError:
        print("错误：test.csv文件不存在")
        return

    if process_cd.strip() in skip_processes:
        print(f"流程编码{process_cd}在跳过列表中，跳过清理")
        return
    
    # 执行清理流程
    print(f"启动流程{process_cd}的清理操作")
    # cursor.execute("DELETE FROM your_table WHERE process_cd = %s", (process_cd.strip(),))
    # conn.commit()

常见失效原因排查 #

1. 列名不匹配：CSV中的流程编码列名不是process_cd（比如是中文“流程编码”），需修改代码中的列名或用索引（如row[0]）获取值。
2. 编码问题：CSV使用GBK等非UTF-8编码，读取时需指定encoding='gbk'，否则会出现乱码导致匹配失败。
3. 空格/换行符干扰：读取的编码或传入的process_cd前后有空格、换行符，必须用strip()清理后再判断。
4. 路径错误：脚本运行的当前目录与CSV所在目录不一致，需使用绝对路径（如'D:/projects/test.csv'）。
5. 空值/空行未过滤：CSV中存在空行或空的流程编码，未过滤会导致集合中混入空字符串，干扰判断逻辑。
6. 数据类型不匹配：CSV中的编码是数字格式，传入的process_cd是字符串，需统一转成字符串后再判断。

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