本文介绍使用 python 清理无引号 csv（或类 csv）文件中字段内部的意外 crlf 换行符，保留合法行尾换行，确保每条记录严格对应一行，适用于含多行文本字段但无字段引号保护的原始数据。
本文介绍使用 python 清理无引号 csv（或类 csv）文件中字段内部的意外 crlf 换行符，保留合法行尾换行，确保每条记录严格对应一行，适用于含多行文本字段但无字段引号保护的原始数据。
在处理来自外部系统的原始文本数据时，常见一类“伪 CSV”格式：字段以分隔符（如 ~ 或 ,）划分，但字段值未加引号，且某些字段（如 description）本身可能包含换行符（\r\n 或 \n）。这会导致标准 CSV 解析器误将单条记录拆分为多行，破坏数据完整性。
以题目中的示例为例，第 3 条记录因 description 字段含两个内部换行，被错误切分为三行：
3~Howard~Sally~Walks along the beach Likes bike rides Loves to cook~9/7/67
而理想结果应为单行：
3~Howard~Sally~Walks along the beach Likes bike rides Loves to cook~9/7/67
核心挑战在于：区分“字段内换行”与“记录结束换行”。由于无引号约束，无法依赖语法结构，需借助业务规则进行启发式识别——例如：每行记录以数字 ID 开头，以日期（如 M/D/Y 格式）结尾，中间字段数固定（由首行 header 决定）。
以下提供两种稳健方案，均基于该业务特征：
✅ 方案一：内存友好型（逐行缓冲 + 正则匹配）
适用于大文件，流式处理，内存占用低：
import re
# 编译正则：匹配以数字ID开头、以 M/D/Y 结尾的完整记录（支持跨行）
pattern = re.compile(r'^\d+~[^~]*?~[^~]*?~[^~]*?~\d{1,2}/\d{1,2}/\d{2}$', re.MULTILINE)
with open('sample.txt') as f_in, open('cleaned.csv', 'w') as f_out:
header = f_in.readline().rstrip('\r\n')
f_out.write(header.replace('~', ',') + '\n') # 统一转为逗号分隔（可选）
buffer = ''
for line in f_in:
buffer += line
# 尝试从缓冲区开头匹配完整记录（非贪婪，确保最短有效匹配）
match = pattern.match(buffer.strip())
if match:
# 清理内部换行：替换所有 \n 和 \r 为空格，保留末尾换行
cleaned_line = buffer.strip().replace('\n', ' ').replace('\r', ' ')
f_out.write(cleaned_line.replace('~', ',') + '\n')
buffer = ''
# 若缓冲区过大仍未匹配，可添加超时保护（如 len(buffer) > 10000）
? 关键点说明：
- 正则
^\d+~...~\d{1,2}/\d{1,2}/\d{2}$锚定行首行尾，确保只匹配完整记录；buffer.strip()消除首尾空白后匹配，避免空行干扰；replace('\n', ' ')将所有内部换行转为空格（亦可替换为''或'等）；
'- 分隔符
~在输出前统一替换为,，便于后续 CSV 解析。
✅ 方案二：简洁版（全量读取 + finditer）
适合中小文件（
import re
with open('sample.txt') as f_in:
header = f_in.readline().rstrip('\r\n')
content = f_in.read()
# 构建正则：匹配 ID + 3 个非~字段 + 日期，支持跨行（re.DOTALL）
record_pattern = r'^(\d+)~([^~]*)~([^~]*)~([^~]*)~(\d{1,2}/\d{1,2}/\d{2})$'
compiled = re.compile(record_pattern, re.MULTILINE | re.DOTALL)
with open('cleaned.csv', 'w') as f_out:
f_out.write(header.replace('~', ',') + '\n')
for match in compiled.finditer(content):
# 重组字段，合并 description 中的换行
id_, first, last, desc, date = match.groups()
# 清洗 description：移除内部换行，保留空格分隔
clean_desc = ' '.join(desc.splitlines()).strip()
f_out.write(f'{id_},{first},{last},{clean_desc},{date}\n')
⚠️ 注意事项与增强建议
-
日期格式鲁棒性：若出生日期格式不统一（如
YYYY-MM-DD），请扩展正则中的日期部分，例如(\d{4}-\d{2}-\d{2}|\d{1,2}/\d{1,2}/\d{2})； -
字段数量动态适配：可先解析 header 行统计列数（
len(header.split('~'))），再生成对应数量的[^~]*?分组，提升可维护性； -
编码与 BOM 处理：打开文件时显式指定编码，如
open(..., encoding='utf-8-sig')，自动去除 UTF-8 BOM； -
异常记录隔离：对无法匹配的
buffer，可写入日志文件单独分析，避免静默丢弃脏数据； -
性能优化：对超大文件，可结合
mmap或pandas.read_csv(..., engine='python', skiprows=1)配合自定义lineterminator预处理。
通过以上方法，你能在不依赖外部库（如 csv 模块）的前提下，精准识别并修复无引号 CSV 中的换行污染问题，为后续数据分析奠定干净、可靠的结构化基础。
