【Python办公】使用pandas批量读取csv保存为Excel

专栏导读

• 🌸 欢迎来到Python办公自动化专栏—Python处理办公问题，解放您的双手

• 🏳️‍🌈 博客主页：请点击——> 一晌小贪欢的博客主页求关注

• 👍 该系列文章专栏：请点击——>Python办公自动化专栏求订阅

• 🕷 此外还有爬虫专栏：请点击——>Python爬虫基础专栏求订阅

• 📕 此外还有python基础专栏：请点击——>Python基础学习专栏求订阅

• 文章作者技术和水平有限，如果文中出现错误，希望大家能指正🙏

• ❤️ 欢迎各位佬关注！ ❤️

Python使用pandas批量读取CSV保存为Excel

概述

在数据处理工作中，经常需要将多个CSV文件转换为Excel格式，或者将多个CSV文件合并到一个Excel文件的不同工作表中。pandas库提供了强大的数据处理功能，可以轻松实现CSV到Excel的批量转换。本文将详细介绍如何使用pandas进行批量CSV文件处理，包括基础转换、高级功能和性能优化等内容。

环境准备

安装必要的库

pip install pandas openpyxl xlsxwriter

库说明：

• pandas：核心数据处理库
• openpyxl：用于读写Excel 2010 xlsx/xlsm/xltx/xltm文件
• xlsxwriter：用于创建Excel文件，支持更多格式设置

导入必要的模块

import pandas as pd
import os
import glob
from pathlib import Path
import warnings
warnings.filterwarnings('ignore')

基础用法

1. 单个CSV文件转Excel

def csv_to_excel_single(csv_file, excel_file):"""将单个CSV文件转换为Excel文件参数:csv_file: CSV文件路径excel_file: 输出Excel文件路径"""try:# 读取CSV文件df = pd.read_csv(csv_file, encoding='utf-8')# 保存为Excel文件df.to_excel(excel_file, index=False, engine='openpyxl')print(f"转换成功: {csv_file} -> {excel_file}")return Trueexcept Exception as e:print(f"转换失败: {csv_file}, 错误: {str(e)}")return False# 使用示例
csv_to_excel_single('data.csv', 'data.xlsx')

2. 批量转换CSV文件到独立Excel文件

def batch_csv_to_excel(input_folder, output_folder):"""批量将CSV文件转换为独立的Excel文件参数:input_folder: 包含CSV文件的输入文件夹output_folder: Excel文件输出文件夹"""# 确保输出文件夹存在os.makedirs(output_folder, exist_ok=True)# 获取所有CSV文件csv_files = glob.glob(os.path.join(input_folder, "*.csv"))success_count = 0fail_count = 0print(f"发现 {len(csv_files)} 个CSV文件")for csv_file in csv_files:# 生成对应的Excel文件名base_name = os.path.splitext(os.path.basename(csv_file))[0]excel_file = os.path.join(output_folder, f"{base_name}.xlsx")# 转换文件if csv_to_excel_single(csv_file, excel_file):success_count += 1else:fail_count += 1print(f"\n转换完成！成功: {success_count}, 失败: {fail_count}")# 使用示例
batch_csv_to_excel('csv_files/', 'excel_files/')

高级功能

3. 多个CSV文件合并到一个Excel文件的不同工作表

def merge_csv_to_excel_sheets(input_folder, output_file):"""将多个CSV文件合并到一个Excel文件的不同工作表中参数:input_folder: 包含CSV文件的输入文件夹output_file: 输出Excel文件路径"""csv_files = glob.glob(os.path.join(input_folder, "*.csv"))if not csv_files:print("未找到CSV文件")return# 创建Excel写入器with pd.ExcelWriter(output_file, engine='openpyxl') as writer:for csv_file in csv_files:try:# 读取CSV文件df = pd.read_csv(csv_file, encoding='utf-8')# 使用文件名作为工作表名（去掉扩展名）sheet_name = os.path.splitext(os.path.basename(csv_file))[0]# 确保工作表名不超过31个字符（Excel限制）if len(sheet_name) > 31:sheet_name = sheet_name[:31]# 写入工作表df.to_excel(writer, sheet_name=sheet_name, index=False)print(f"已添加工作表: {sheet_name}")except Exception as e:print(f"处理文件失败: {csv_file}, 错误: {str(e)}")print(f"\n合并完成！输出文件: {output_file}")# 使用示例
merge_csv_to_excel_sheets('csv_files/', 'merged_data.xlsx')

4. 带格式设置的转换

def csv_to_excel_with_format(csv_file, excel_file):"""将CSV转换为带格式的Excel文件"""# 读取CSV文件df = pd.read_csv(csv_file, encoding='utf-8')# 创建Excel写入器with pd.ExcelWriter(excel_file, engine='xlsxwriter') as writer:# 写入数据df.to_excel(writer, sheet_name='数据', index=False)# 获取工作簿和工作表对象workbook = writer.bookworksheet = writer.sheets['数据']# 定义格式header_format = workbook.add_format({'bold': True,'text_wrap': True,'valign': 'top','fg_color': '#D7E4BC','border': 1})# 设置列宽for i, col in enumerate(df.columns):# 计算列宽（基于列名和数据的最大长度）max_len = max(df[col].astype(str).map(len).max(),  # 数据最大长度len(str(col))  # 列名长度)worksheet.set_column(i, i, min(max_len + 2, 50))  # 限制最大宽度为50# 设置表头格式for col_num, value in enumerate(df.columns.values):worksheet.write(0, col_num, value, header_format)# 冻结首行worksheet.freeze_panes(1, 0)print(f"带格式转换完成: {excel_file}")# 使用示例
csv_to_excel_with_format('data.csv', 'formatted_data.xlsx')

5. 数据清洗和预处理

def csv_to_excel_with_cleaning(csv_file, excel_file):"""转换CSV到Excel并进行数据清洗"""try:# 读取CSV文件df = pd.read_csv(csv_file, encoding='utf-8')print(f"原始数据形状: {df.shape}")# 数据清洗步骤# 1. 删除完全重复的行df = df.drop_duplicates()# 2. 删除全为空值的行df = df.dropna(how='all')# 3. 删除全为空值的列df = df.dropna(axis=1, how='all')# 4. 清理字符串列的前后空格string_columns = df.select_dtypes(include=['object']).columnsfor col in string_columns:df[col] = df[col].astype(str).str.strip()# 5. 替换空字符串为NaNdf = df.replace('', pd.NA)print(f"清洗后数据形状: {df.shape}")# 保存到Excelwith pd.ExcelWriter(excel_file, engine='openpyxl') as writer:# 原始数据df.to_excel(writer, sheet_name='清洗后数据', index=False)# 数据概览summary_df = pd.DataFrame({'列名': df.columns,'数据类型': [str(dtype) for dtype in df.dtypes],'非空值数量': [df[col].count() for col in df.columns],'空值数量': [df[col].isnull().sum() for col in df.columns],'唯一值数量': [df[col].nunique() for col in df.columns]})summary_df.to_excel(writer, sheet_name='数据概览', index=False)print(f"数据清洗和转换完成: {excel_file}")return Trueexcept Exception as e:print(f"处理失败: {str(e)}")return False# 使用示例
csv_to_excel_with_cleaning('messy_data.csv', 'cleaned_data.xlsx')

完整的批量处理解决方案

6. 多功能批量处理类

class CSVToExcelConverter:"""CSV到Excel的批量转换器"""def __init__(self, input_folder, output_folder=None):self.input_folder = Path(input_folder)self.output_folder = Path(output_folder) if output_folder else self.input_folder / 'excel_output'self.output_folder.mkdir(exist_ok=True)# 统计信息self.total_files = 0self.success_count = 0self.fail_count = 0self.error_log = []def get_csv_files(self):"""获取所有CSV文件"""return list(self.input_folder.glob('*.csv'))def convert_single_file(self, csv_file, clean_data=True, add_format=True):"""转换单个文件参数:csv_file: CSV文件路径clean_data: 是否进行数据清洗add_format: 是否添加格式"""try:# 读取CSVdf = pd.read_csv(csv_file, encoding='utf-8')# 数据清洗if clean_data:df = self._clean_data(df)# 生成输出文件名excel_file = self.output_folder / f"{csv_file.stem}.xlsx"# 保存Excelif add_format:self._save_with_format(df, excel_file)else:df.to_excel(excel_file, index=False, engine='openpyxl')self.success_count += 1print(f"✓ 转换成功: {csv_file.name}")return Trueexcept Exception as e:self.fail_count += 1error_msg = f"✗ 转换失败: {csv_file.name}, 错误: {str(e)}"print(error_msg)self.error_log.append(error_msg)return Falsedef _clean_data(self, df):"""数据清洗"""# 删除重复行df = df.drop_duplicates()# 删除全空行和列df = df.dropna(how='all').dropna(axis=1, how='all')# 清理字符串string_cols = df.select_dtypes(include=['object']).columnsfor col in string_cols:df[col] = df[col].astype(str).str.strip().replace('nan', '')return dfdef _save_with_format(self, df, excel_file):"""保存带格式的Excel文件"""with pd.ExcelWriter(excel_file, engine='xlsxwriter') as writer:df.to_excel(writer, sheet_name='数据', index=False)workbook = writer.bookworksheet = writer.sheets['数据']# 表头格式header_format = workbook.add_format({'bold': True,'text_wrap': True,'valign': 'top','fg_color': '#D7E4BC','border': 1})# 设置列宽和表头格式for col_num, value in enumerate(df.columns.values):# 计算列宽max_len = max(df.iloc[:, col_num].astype(str).map(len).max(),len(str(value)))worksheet.set_column(col_num, col_num, min(max_len + 2, 50))# 设置表头格式worksheet.write(0, col_num, value, header_format)# 冻结首行worksheet.freeze_panes(1, 0)def batch_convert(self, clean_data=True, add_format=True):"""批量转换"""csv_files = self.get_csv_files()self.total_files = len(csv_files)if self.total_files == 0:print("未找到CSV文件")returnprint(f"开始批量转换，共 {self.total_files} 个文件")print("=" * 50)for csv_file in csv_files:self.convert_single_file(csv_file, clean_data, add_format)self._print_summary()def merge_to_single_excel(self, output_file='merged_data.xlsx'):"""合并所有CSV到单个Excel文件"""csv_files = self.get_csv_files()if not csv_files:print("未找到CSV文件")returnoutput_path = self.output_folder / output_filewith pd.ExcelWriter(output_path, engine='openpyxl') as writer:for csv_file in csv_files:try:df = pd.read_csv(csv_file, encoding='utf-8')sheet_name = csv_file.stem[:31]  # Excel工作表名限制df.to_excel(writer, sheet_name=sheet_name, index=False)print(f"✓ 已添加工作表: {sheet_name}")except Exception as e:print(f"✗ 处理失败: {csv_file.name}, 错误: {str(e)}")print(f"\n合并完成！输出文件: {output_path}")def _print_summary(self):"""打印处理摘要"""print("=" * 50)print(f"批量转换完成！")print(f"总文件数: {self.total_files}")print(f"成功转换: {self.success_count}")print(f"转换失败: {self.fail_count}")print(f"输出目录: {self.output_folder}")if self.error_log:print("\n错误详情:")for error in self.error_log:print(f"  {error}")# 使用示例
if __name__ == "__main__":# 创建转换器实例converter = CSVToExcelConverter(input_folder='csv_files',output_folder='excel_output')# 方式1: 批量转换为独立Excel文件converter.batch_convert(clean_data=True, add_format=True)# 方式2: 合并为单个Excel文件# converter.merge_to_single_excel('all_data.xlsx')

性能优化

7. 多线程批量处理

import concurrent.futures
from threading import Lockclass MultiThreadCSVConverter(CSVToExcelConverter):"""多线程CSV转换器"""def __init__(self, input_folder, output_folder=None, max_workers=4):super().__init__(input_folder, output_folder)self.max_workers = max_workersself.lock = Lock()  # 线程锁，用于安全地更新计数器def _thread_safe_update(self, success=True, error_msg=None):"""线程安全的计数器更新"""with self.lock:if success:self.success_count += 1else:self.fail_count += 1if error_msg:self.error_log.append(error_msg)def _convert_file_thread(self, csv_file, clean_data=True, add_format=True):"""线程中执行的转换函数"""try:# 读取CSVdf = pd.read_csv(csv_file, encoding='utf-8')# 数据清洗if clean_data:df = self._clean_data(df)# 生成输出文件名excel_file = self.output_folder / f"{csv_file.stem}.xlsx"# 保存Excelif add_format:self._save_with_format(df, excel_file)else:df.to_excel(excel_file, index=False, engine='openpyxl')self._thread_safe_update(success=True)print(f"✓ [线程] 转换成功: {csv_file.name}")return Trueexcept Exception as e:error_msg = f"✗ [线程] 转换失败: {csv_file.name}, 错误: {str(e)}"self._thread_safe_update(success=False, error_msg=error_msg)print(error_msg)return Falsedef batch_convert_multithread(self, clean_data=True, add_format=True):"""多线程批量转换"""csv_files = self.get_csv_files()self.total_files = len(csv_files)if self.total_files == 0:print("未找到CSV文件")returnprint(f"开始多线程批量转换，共 {self.total_files} 个文件")print(f"使用线程数: {self.max_workers}")print("=" * 50)# 使用线程池执行转换with concurrent.futures.ThreadPoolExecutor(max_workers=self.max_workers) as executor:# 提交所有任务futures = [executor.submit(self._convert_file_thread, csv_file, clean_data, add_format)for csv_file in csv_files]# 等待所有任务完成concurrent.futures.wait(futures)self._print_summary()# 使用示例
if __name__ == "__main__":# 多线程转换器mt_converter = MultiThreadCSVConverter(input_folder='csv_files',output_folder='excel_output',max_workers=8  # 根据CPU核心数调整)# 执行多线程批量转换mt_converter.batch_convert_multithread(clean_data=True, add_format=True)

常见问题解决方案

编码问题

def read_csv_with_encoding_detection(file_path):"""自动检测CSV文件编码并读取"""import chardet# 检测文件编码with open(file_path, 'rb') as f:raw_data = f.read(10000)  # 读取前10KB用于检测encoding = chardet.detect(raw_data)['encoding']print(f"检测到编码: {encoding}")# 尝试不同的编码encodings = [encoding, 'utf-8', 'gbk', 'gb2312', 'utf-8-sig']for enc in encodings:try:df = pd.read_csv(file_path, encoding=enc)print(f"成功使用编码: {enc}")return dfexcept UnicodeDecodeError:continueraise ValueError(f"无法读取文件 {file_path}，尝试了编码: {encodings}")

大文件处理

def process_large_csv(csv_file, excel_file, chunk_size=10000):"""分块处理大型CSV文件"""# 创建Excel写入器with pd.ExcelWriter(excel_file, engine='xlsxwriter') as writer:# 分块读取CSVchunk_iter = pd.read_csv(csv_file, chunksize=chunk_size)for i, chunk in enumerate(chunk_iter):# 处理每个块sheet_name = f'数据_{i+1}'chunk.to_excel(writer, sheet_name=sheet_name, index=False)print(f"已处理块 {i+1}，行数: {len(chunk)}")print(f"大文件处理完成: {excel_file}")

内存优化

def memory_efficient_conversion(csv_files, output_file):"""内存高效的批量转换"""with pd.ExcelWriter(output_file, engine='xlsxwriter') as writer:for csv_file in csv_files:# 逐个处理文件，避免同时加载所有数据df = pd.read_csv(csv_file)# 立即写入Excel并释放内存sheet_name = Path(csv_file).stem[:31]df.to_excel(writer, sheet_name=sheet_name, index=False)# 显式删除DataFrame释放内存del dfprint(f"已处理: {csv_file}")

最佳实践

1. 错误处理和日志记录

import logging
from datetime import datetime# 配置日志
logging.basicConfig(level=logging.INFO,format='%(asctime)s - %(levelname)s - %(message)s',handlers=[logging.FileHandler(f'csv_conversion_{datetime.now().strftime("%Y%m%d_%H%M%S")}.log'),logging.StreamHandler()]
)def robust_csv_to_excel(csv_file, excel_file):"""带完整错误处理的转换函数"""try:logging.info(f"开始处理: {csv_file}")# 检查文件是否存在if not os.path.exists(csv_file):raise FileNotFoundError(f"CSV文件不存在: {csv_file}")# 检查文件大小file_size = os.path.getsize(csv_file)logging.info(f"文件大小: {file_size / 1024 / 1024:.2f} MB")# 读取CSVdf = pd.read_csv(csv_file, encoding='utf-8')logging.info(f"数据形状: {df.shape}")# 保存Exceldf.to_excel(excel_file, index=False, engine='openpyxl')logging.info(f"转换成功: {excel_file}")return Trueexcept Exception as e:logging.error(f"转换失败: {csv_file}, 错误: {str(e)}")return False

2. 配置文件管理

import json# config.json
config = {"input_folder": "csv_files","output_folder": "excel_output","clean_data": True,"add_format": True,"max_workers": 8,"chunk_size": 10000,"supported_encodings": ["utf-8", "gbk", "gb2312", "utf-8-sig"]
}def load_config(config_file='config.json'):"""加载配置文件"""try:with open(config_file, 'r', encoding='utf-8') as f:return json.load(f)except FileNotFoundError:# 创建默认配置文件with open(config_file, 'w', encoding='utf-8') as f:json.dump(config, f, indent=2, ensure_ascii=False)return config

3. 进度监控

from tqdm import tqdmdef batch_convert_with_progress(csv_files, output_folder):"""带进度条的批量转换"""success_count = 0# 使用tqdm显示进度with tqdm(total=len(csv_files), desc="转换进度") as pbar:for csv_file in csv_files:excel_file = os.path.join(output_folder, f"{Path(csv_file).stem}.xlsx")if csv_to_excel_single(csv_file, excel_file):success_count += 1# 更新进度条pbar.set_postfix({'成功': success_count,'当前': Path(csv_file).name})pbar.update(1)return success_count

总结

本文详细介绍了使用pandas进行CSV到Excel批量转换的各种方法，从基础的单文件转换到高级的多线程批量处理。主要要点包括：

1. 基础转换：使用pd.read_csv()和to_excel()实现基本转换
2. 批量处理：通过循环和文件操作实现批量转换
3. 数据清洗：在转换过程中进行数据预处理
4. 格式设置：使用xlsxwriter添加Excel格式
5. 性能优化：多线程处理和内存管理
6. 错误处理：完善的异常处理和日志记录

选择建议：

• 小批量文件（<100个）：使用基础批量转换
• 大批量文件（>100个）：使用多线程转换
• 需要合并数据：使用多工作表合并方案
• 需要格式化：使用xlsxwriter引擎
• 处理大文件：使用分块读取方案

通过合理选择和组合这些方法，可以高效地完成各种CSV到Excel的转换需求。

结尾

• 希望对初学者有帮助；致力于办公自动化的小小程序员一枚

• 希望能得到大家的【❤️一个免费关注❤️】感谢！

• 求个 🤞 关注 🤞 +❤️ 喜欢 ❤️ +👍 收藏 👍

• 此外还有办公自动化专栏，欢迎大家订阅：Python办公自动化专栏

• 此外还有爬虫专栏，欢迎大家订阅：Python爬虫基础专栏

• 此外还有Python基础专栏，欢迎大家订阅：Python基础学习专栏