Mac上使用python 和pandas拆分csv数据文件的方法记录

在 mac/python 环境下，将当前目录 csv 按固定行数拆分为多个带表头的小文件，默认每 10 条一个。适用于 seo、后台导出等格式不规范的脏 csv，不依赖 pandas，零第三方库。

排错日志

初期使用 pandas chunksize 拆分，遇到字段数不一致报错；增加容错参数后，又因引号未闭合再次中断。最终放弃严格 csv 解析，改用物理行拆分：仅读取首行作表头，后续按换行符切分，每满指定行数写入新文件并补写表头。该方案绕过解析器，对脏数据容错性强，但无法处理字段内含换行的合规 csv。

#!/usr/bin/env python3
# -*- coding: utf-8 -NewVFX-JSB开发*-
"""
CSV 自动拆分工具
- 自动检测脚本所在目录下的所有 .csv 文件
- 自动检测并安装 pandas
- 按每 100 条拆分为独立文件（保留表头）
"""

import subprocess
import sys
import os
import glob


def ensure_pandas():
    """检测 pandas 是否已安装，未安装则自动 pip install"""
    try:
        import pandas as pd
        print(f"✅ pandas 已就绪 (版本: {pd.__version__})")
        return pd
    except ImportError:
        print("⚠️  未检测到 pandas，正在自动安装...")
        try:
            subprocess.check_call(
                [sys.executable, "-m", "pip", "install", "pandas", "-q"],
                stdout=subprocess.DEVNULL,
                stderr=subprocess.STDOUT,
            )
            import pandas as pd
            print(f"✅ pandas 安装成功 (版本: {pd.__version__})")
            return pd
        except Exception as e:
            print(f"❌ pandas 安装失败: {e}")
            print("   请手动执行: pip3 install pandas")
            sys.exit(1)


def find_csv_files(script_dir):
    """查找脚本所在目录下的所有 csv 文件（不递归子目录）"""
    pattern = os.path.join(script_dir, "*.csv")
    files = sorted(glob.glob(pattern))
    # 排除输出目录中可能存在的 csv
    output_dir = os.path.join(script_dir, "split_output")
    files = [f for f in files if not f.startswith(output_dir)]
    return files

def split_csv(pd, input_file, output_dir, chunk_size=100):
    """
    纯文本行级拆分 - 完全绕过 CSV 解析器
    适用于格式严重损坏、pandas/csv 模块均无法解析的文件
    """
    os.makedirs(output_dir, exist_ok=True)
    base_name = os.path.splitext(os.path.basename(input_file))[0]

    try:
        with open(input_file, "r", encoding="utf-8-sig") as f:
            header = f.readline()  # 读取第一行作为表头
            if not header.strip():
                print("   ❌ 文件为空或无表头\n")
                return 0

            batch = []
            file_count = 0
            total_rows = 0

            for line in f:
                if not line.strip():
                    continue  # 跳过空行
                batch.append(line)

                if len(batch) == chunk_size:
                    file_count += 1
                    out_path = os.path.join(
                        output_dir, f"{base_name}_part{file_count:04d}.csv"
                    )
                    with open(out_path, "w", encoding="utf-8-sig") as out:
                        out.write(header)
                        out.writelines(batch)
                    total_rows += len(batch)
                    print(f"   📄 {os.path.basename(out_path)} ({len(batch)} 条)")
                    batch = []

            # 处理剩余不足 chunk_size 的行
            if batch:
                file_count += 1
                out_path = os.path.join(
                    output_dir, f"{base_name}_part{file_count:04d}.csv"
                )
                with open(out_path, "w", encoding="utf-8-sig") as out:
                    out.write(header)
                    out.writelines(batch)
                total_rows += len(batch)
                print(f"   📄 {os.path.basename(out_path)} ({len(batch)} 条)")

        print(f"   ✅ 完成: 共写入 {total_rows} 条 → {file_count} 个文件\n")
        return file_count

    except Exception as e:
        print(f"   ❌ 拆分失败: {e}\n")
        return 0


def main():
    # 1. 确保 pandas 可用
    pd = ensure_pandas()

    # 2. 定位脚本所在目录
    script_dir = os.path.dirname(os.path.abspath(__file__))
    print(f"\n📂 工作目录: {script_dir}")

    # 3. 查找 CSV 文件
    csv_files = find_csv_files(script_dir)
    if not csv_files:
        print("❌ 当前目录下未找到任何 .csv 文件，请将脚本与 CSV 放在同一文件夹。")
        sys.exit(0)

    print(f"🔍 发现 {len(csv_files)} 个 CSV 文件:\n")
    for f in csv_files:
        print(f"   • {os.path.basename(f)}")
    print()

    # 4. 逐个拆分
    output_dir = os.path.join(script_dir, "split_output")
    total_files_generated = 0

    for csv_file in csv_files:
        print(f"⏳ 正在拆分: {os.path.basename(csv_file)}")
        try:
            count = split_csv(pd, csv_file, output_dir, chunk_size=100)
            total_files_generated += count
        except Exception as e:
            print(f"   ❌ 拆分失败: {e}\n")

    print(f"🎉 全部完成! 共生成 {total_files_generated} 个文件")
    print(f"📁 输出目录: {output_dir}")


if __name__ == "__main__":
    main()

使用方法

1. 将脚本保存为 splitcsv.py，放入 csv 所在目录；
2. 终端执行 python3 splitcsv.py，默认每 10 条拆分；
3. 自定义行数：python3 splitcsv.py 20；
4. 结果自动存入带时间戳的 splitoutput 目录，不覆盖原文件。

核心规则

仅扫描脚本同级目录 csv，不递归子目录；
每个子文件保留原表头，命名采用四位序号便于排序；
输出统一 utf-8-sig 编码，兼容 excel 中文显示；
自动跳过空行，尝试多种编码读取，失败则跳过并提示；
不修改、不修复源 csv，仅作物理拆分。

注意事项
若 csv 字段内含换行且被引号包裹，会被误拆为多行，此类规范 csv 建议用 pandas 处理；
工具仅保证按文本行拆分，不校验 csv 语义完整性，导入前需抽查验证；
mac 环境务必使用 python3 命令，避免调用系统旧版 python；
乱码时可调整脚本内编码尝试顺序，优先匹配源文件实际编码。

faq
无需安装 pandas，纯标准库实现；
支持批量处理同级目录所有 csv；
不会覆盖原文件，每次生成独立输出目录；
文件数少于预期时，检查空行、跨行记录或编码问题；
仅适合列表型 csv，复杂嵌套结构不适用。

规范 csv 用 pandas 分块读取；脏 csv 优先用本物理行拆分方案，稳定、轻量、开箱即用，适合快速复用。