如何处理含内嵌换行的CSV文件：过滤、排序与去重

你的问题核心是普通文本工具（sed/sort）无法正确解析带引号包裹的多行JSON字段，导致过滤和列识别失效。下面给出两种靠谱的解决方案，适配你的日常批量处理需求：

方案一：用csvkit命令行工具（贴近原工作流） #

csvkit是专门处理CSV的工具集，能自动识别引号包裹的特殊字段（包括换行、逗号），完全适配你的场景。

步骤： #

1. 先安装csvkit：

pip install csvkit

2. 执行批量处理命令：

# 合并所有CSV（自动去重表头）→ 过滤含KEYWORD的行 → 按日期降序 → 按ID去重（保留最新记录）
csvstack -n *.csv | \
csvfilter -c 3,4 -r '(?i)KEYWORD' --invert | \
csvsort -c 5 -r | \
csvuniq -c 1 > processed.csv

命令解析： #

• csvstack -n *.csv：合并所有CSV文件，自动保留一份表头，忽略后续文件的表头
• csvfilter -c 3,4 -r '(?i)KEYWORD' --invert：检查第3列（NAME）和第4列（JSON），去掉包含KEYWORD的行（(?i)表示不区分大小写）
• csvsort -c 5 -r：按第5列（TIME，ISO日期可直接字符串排序）降序排列
• csvuniq -c 1：按第1列（ID）去重，保留排序后第一个出现的行（也就是最新时间的记录）

方案二：Python脚本（自定义性更强） #

如果需要更灵活的逻辑（比如自定义JSON解析规则、特殊去重逻辑），用Python脚本更靠谱，无需额外安装工具（只要有Python环境）。

示例脚本： #

import csv
import json
from pathlib import Path

# 可配置参数
KEYWORD = "KEYWORD"
INPUT_CSV_PATTERN = "*.csv"
OUTPUT_FILE = "processed.csv"

# 存储处理后的记录，key为ID，value为对应最新时间的行数据
processed_data = {}
header = None

# 遍历所有CSV文件
for csv_file in Path().glob(INPUT_CSV_PATTERN):
    with open(csv_file, 'r', newline='', encoding='utf-8') as f:
        reader = csv.reader(f)
        current_header = next(reader)
        # 只保留一份表头
        if not header:
            header = current_header
        
        for row in reader:
            if len(row) < 5:
                continue  # 跳过无效行
            
            row_id = row[0]
            name_col = row[2]
            json_col = row[3]
            time_col = row[4]
            
            # 过滤第3列含KEYWORD的行（不区分大小写）
            if KEYWORD.lower() in name_col.lower():
                continue
            
            # 过滤JSON字段中含KEYWORD的行
            try:
                # 处理CSV里的双引号转义（""转成"）
                json_content = json_col.replace('""', '"')
                json_data = json.loads(json_content)
                # 把JSON转成字符串后检查是否含KEYWORD
                if KEYWORD.lower() in json.dumps(json_data).lower():
                    continue
            except json.JSONDecodeError:
                # JSON解析失败时，默认保留该行（可根据需求修改）
                pass
            
            # 去重逻辑：保留同一ID下时间最新的记录
            if row_id in processed_data:
                existing_time = processed_data[row_id][4]
                if time_col > existing_time:
                    processed_data[row_id] = row
            else:
                processed_data[row_id] = row

# 按时间降序排序
sorted_rows = sorted(processed_data.values(), key=lambda x: x[4], reverse=True)

# 写入输出文件，保持原CSV格式
with open(OUTPUT_FILE, 'w', newline='', encoding='utf-8') as f:
    writer = csv.writer(f, quoting=csv.QUOTE_NONNUMERIC)
    writer.writerow(header)
    writer.writerows(sorted_rows)

脚本优势： #

• 完全正确解析带换行的JSON字段，不会把JSON内的换行/逗号当成CSV行分隔符
• 可灵活调整过滤规则（比如只检查JSON的某个字段）
• 去重逻辑可自定义（比如按整行去重，而非ID）

日常自动化处理 #

如果每天新增文件，可把命令或脚本加入定时任务（比如Linux的crontab）：

# 每天凌晨2点自动执行处理
0 2 * * * /usr/bin/python3 /path/to/your/script.py
# 或用csvkit命令
0 2 * * * /usr/local/bin/csvstack -n /path/to/csvs/*.csv | csvfilter -c 3,4 -r '(?i)KEYWORD' --invert | csvsort -c 5 -r | csvuniq -c 1 > /path/to/output/processed.csv

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