
本文介绍使用 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', ' ')将所有内部换行转为空格(亦可替换为''或'<br>'等);- 分隔符
~在输出前统一替换为,,便于后续 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 中的换行污染问题,为后续数据分析奠定干净、可靠的结构化基础。


















