将非常规格式CSV文件拆分至多个CSV文件的技术方案咨询

1. 先明确数据格式规则 #

取数据集的一小段样本，手动梳理以下关键规则：

• 记录边界：每条记录的开始/结束标识（比如特定字符串、空行、固定行数）
• 字段分隔：字段之间的分隔符（比如特殊符号|#|、固定字符长度、空格分隔）
• 特殊情况：是否存在跨行记录、转义字符、注释行等

2. 用Python基础文件操作读取处理（无需依赖csv模块） #

读取文件内容

先把整个文件读入内存（数据量过大的话可以逐行处理）：

with open("your_dataset.txt", "r", encoding="utf-8") as f:
    raw_lines = f.readlines()

提取单条记录

根据你梳理的边界规则，把raw_lines拆分成独立记录：

• 示例：如果每条记录以[START]开头，以[END]结尾

records = []
current_record = []
for line in raw_lines:
    stripped_line = line.strip()
    if stripped_line == "[START]":
        current_record = []
    elif stripped_line == "[END]":
        records.append(current_record)
    else:
        current_record.append(stripped_line)

解析记录字段

根据字段规则提取每个记录的字段：

• 示例1：字段用;;分隔

parsed_records = []
for record in records:
    # 假设每条记录是一行字符串，直接拆分
    fields = record[0].split(";;")
    parsed_records.append(fields)

• 示例2：固定长度字段（比如前10位是ID，接下来20位是名称）

parsed_records = []
for line in raw_lines:
    if line.startswith("[DATA]"):  # 跳过非数据行
        id_field = line[5:15].strip()  # 提取第5到15位的ID
        name_field = line[15:35].strip()  # 提取名称字段
        parsed_records.append([id_field, name_field])

3. 拆分数据集为多个文件 #

根据需求选择拆分逻辑（按记录数、按字段分组等）：

按记录数批量拆分

batch_size = 1000  # 每个文件存1000条记录
for idx in range(0, len(parsed_records), batch_size):
    batch_data = parsed_records[idx:idx+batch_size]
    # 写入拆分后的文件
    with open(f"dataset_part_{idx//batch_size +1}.txt", "w", encoding="utf-8") as f:
        for fields in batch_data:
            # 按需要的格式写入，比如用逗号分隔字段
            f.write(",".join(fields) + "\n")

按字段值分组拆分

示例：按第一个字段（类别）拆分到不同文件

from collections import defaultdict

# 先按类别分组
grouped_data = defaultdict(list)
for fields in parsed_records:
    category = fields[0]
    grouped_data[category].append(fields)

# 每个类别写入单独文件
for category, data in grouped_data.items():
    with open(f"dataset_{category}.txt", "w", encoding="utf-8") as f:
        for fields in data:
            f.write(",".join(fields) + "\n")

4. 验证与调试 #

• 先处理小样本数据：取前10条记录测试解析和拆分逻辑，打印结果确认正确性
• 检查异常行：如果遇到不符合规则的行，添加判断跳过或单独记录，避免程序崩溃

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