Python并行处理XML生成CSV时行尾出现\x{0D}字符问题求助

\x{0D}是ASCII回车符（CR），UTF-8本身兼容该字符，问题根源是多进程并行写入CSV时的资源竞争或换行符处理不一致：

• 单进程写入时，Python的csv模块会按规范自动处理换行符（默认newline=''），不会出现冗余字符；
• 多进程场景下，多个进程同时操作同一文件句柄，会导致换行符被重复写入（比如各进程独立输出CRLF，最终叠加出多余的CR），或是进程间换行符处理逻辑冲突。

1. 先收集数据再统一写入（推荐） #

让每个进程单独处理XML并返回数据片段，最后由主进程一次性写入CSV，从根源避免多进程文件竞争：

from joblib import Parallel, delayed
import csv
import os
from xml.etree import ElementTree as ET

def process_xml(file_path):
    # 替换为你的XML提取逻辑，返回单条数据的字典/列表
    tree = ET.parse(file_path)
    root = tree.getroot()
    data = {
        'id': root.find('id').text,
        'content': root.find('content').text
    }
    return data

def main():
    xml_dir = './xml_files'
    xml_files = [os.path.join(xml_dir, f) for f in os.listdir(xml_dir) if f.endswith('.xml')]
    
    # 并行处理生成所有数据
    all_data = Parallel(n_jobs=-1)(delayed(process_xml)(f) for f in xml_files)
    
    # 主进程统一写入CSV
    with open('output.csv', 'w', encoding='utf-8', newline='') as csvfile:
        fieldnames = all_data[0].keys() if all_data else []
        writer = csv.DictWriter(csvfile, fieldnames=fieldnames)
        writer.writeheader()
        writer.writerows(all_data)

if __name__ == '__main__':
    main()

2. 多进程写入时加文件锁 #

如果必须让进程直接写入文件，需用文件锁保证同一时间只有一个进程操作文件，同时显式统一换行符：

from joblib import Parallel, delayed
import csv
import os
import fcntl
from xml.etree import ElementTree as ET

def process_xml(file_path):
    # 复用上述XML处理逻辑
    tree = ET.parse(file_path)
    root = tree.getroot()
    return {'id': root.find('id').text, 'content': root.find('content').text}

def process_and_write(file_path):
    data = process_xml(file_path)
    with open('output.csv', 'a', encoding='utf-8', newline='') as csvfile:
        # 加排他锁，防止多进程同时写入
        fcntl.flock(csvfile.fileno(), fcntl.LOCK_EX)
        writer = csv.DictWriter(csvfile, fieldnames=data.keys())
        # 仅文件为空时写入表头
        if os.path.getsize('output.csv') == 0:
            writer.writeheader()
        writer.writerow(data)
        # 释放锁
        fcntl.flock(csvfile.fileno(), fcntl.LOCK_UN)

def main():
    xml_dir = './xml_files'
    xml_files = [os.path.join(xml_dir, f) for f in os.listdir(xml_dir) if f.endswith('.xml')]
    Parallel(n_jobs=-1)(delayed(process_and_write)(f) for f in xml_files)

if __name__ == '__main__':
    main()

3. 事后清理已生成的错误CSV #

如果已经得到带多余\x0D的文件，用单进程脚本快速清理：

with open('bad_output.csv', 'r', encoding='utf-8') as infile, open('fixed_output.csv', 'w', encoding='utf-8', newline='') as outfile:
    for line in infile:
        # 移除多余的回车符
        cleaned_line = line.replace('\r', '')
        outfile.write(cleaned_line)

• 写入CSV时务必指定newline=''，这是Python csv模块的官方规范要求，避免自动转换换行符引发的问题；
• 多进程共享文件资源时，要么"先收集后写入"，要么加锁，绝对不要让多个进程直接裸写同一文件；
• 不要误判为UTF-8编码问题：\x0D是ASCII字符，UTF-8完全兼容，问题本质是换行符冗余。

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