针对“Python从键盘输入CSV文件路径，读取每个CSV的表头并存入DataFrame，遍历所有表头对比是否一致，若一致则合并数据，支持任意数量表头”的需求，核心解决方案是：**通过交互式输入收集文件路径，读取CSV文件时提取表头作为分组键，使用字典存储相同表头对应的DataFrame列表，最后将各组列表纵向合并为独立的CSV文件**[ref_1][ref_3]。以下为详细的代码实现和流程解析。 ### **1. 核心思路与数据结构设计** 为实现对任意数量CSV文件的表头比对与数据合并，需构建一个以**标准化表头**为键的字典，值则为具有该表头的所有原始数据DataFrame组成的列表。这样，遍历并比对表头就转化为对字典键的操作 [ref_3]。 | 数据结构键（Key） | 数据结构值（Value） | 作用说明 | | :--- | :--- | :--- | | `tuple(header_list)` | `List[pandas.DataFrame]` | 键是文件表头转换成的元组，用于精确比对。值是具有该表头的所有DataFrame，便于后续合并。 | 此设计能高效处理表头比对，因为Python中元组（`tuple`）是可哈希的，可以作为字典的键，其相等性比较（`==`）天然实现了表头的逐一元素比对。 ### **2. 完整代码实现** 以下脚本整合了路径收集、数据加载、表头比对、合并输出全流程。 ```python #!/usr/bin/env python3 # -*- coding: utf-8 -*- """ 功能：从键盘交互式输入CSV文件路径，按表头分组并合并表头完全一致的文件。 """ import pandas as pd import os import sys def collect_csv_paths_interactively(): """ 从标准输入循环收集CSV文件路径，支持绝对和相对路径。 输入空行（直接回车）表示结束输入。 返回：有效的CSV文件路径列表。 """ print("=" * 60) print("CSV文件表头比对与合并工具") print("=" * 60) print("\n请逐个输入CSV文件的完整路径（例如：./data/sales.csv 或 C:\\reports\\2024_01.csv）。") print("输入完成后，直接按回车键结束输入。") file_paths = [] count = 1 while True: try: # 提示用户输入单个路径 user_input = input(f"\n请输入第 {count} 个CSV文件路径 (或直接回车结束): ").strip() except (EOFError, KeyboardInterrupt): print("\n输入中断。") break # 判断是否结束输入 if user_input == "": if count == 1: print("未输入任何路径，程序退出。") sys.exit(0) else: print(f"\n输入结束。共收集到 {len(file_paths)} 个路径。") break # 校验路径有效性和文件格式 if not os.path.isfile(user_input): print(f" 错误：路径 '{user_input}' 不存在或不是一个文件，请检查。") continue if not user_input.lower().endswith('.csv'): print(f" 警告：文件 '{user_input}' 不是CSV格式（.csv），已跳过。") continue file_paths.append(user_input) print(f" √ 已添加：{user_input}") count += 1 return file_paths def load_csv_and_group_by_header(file_path_list): """ 加载所有CSV文件，并按表头（列名）进行分组。 参数: file_path_list: CSV文件路径列表。 返回: header_groups: 字典，结构为 { (header_tuple): [df1, df2, ...] } load_log: 加载过程日志列表，记录每个文件的状态。 """ header_groups = {} # 核心分组字典 load_log = [] # 用于记录和后续报告 for idx, file_path in enumerate(file_path_list, 1): file_name = os.path.basename(file_path) print(f" [{idx}/{len(file_path_list)}] 正在处理: {file_name}") try: # 使用pandas读取CSV文件 [ref_1] # 假设表头在第一行。若文件无表头，可设置 header=None df = pd.read_csv(file_path, encoding='utf-8') # 可根据需要调整编码，如 'gbk' except Exception as e: error_msg = f"读取失败: {e}" print(f" × {error_msg}") load_log.append({ '文件': file_name, '状态': '失败', '表头': 'N/A', '数据行数': 0, '备注': error_msg }) continue # 提取表头并将其转换为元组作为分组键 [ref_3] original_header = df.columns.tolist() header_key = tuple(original_header) # 例如：('Date', 'Product', 'Sales') # 初始化该表头分组（如果首次出现） if header_key not in header_groups: header_groups[header_key] = [] # 将当前文件的DataFrame添加到对应的分组列表中 header_groups[header_key].append(df) # 记录成功加载的信息 load_log.append({ '文件': file_name, '状态': '成功', '表头': str(original_header), '数据行数': df.shape[0], '备注': 'N/A' }) print(f" √ 成功加载，表头: {original_header[:3]}{'...' if len(original_header) > 3 else ''}") return header_groups, load_log def merge_and_export_groups(header_groups, output_dir='./merged_csv_output'): """ 将每个表头分组下的多个DataFrame合并，并导出为一个独立的CSV文件。 参数: header_groups: load_csv_and_group_by_header 返回的分组字典。 output_dir: 合并结果输出目录。 返回: summary_report: 合并操作的摘要报告列表。 """ # 创建输出目录（如果不存在） os.makedirs(output_dir, exist_ok=True) print(f"\n合并结果将输出至目录: {os.path.abspath(output_dir)}") summary_report = [] group_index = 1 for header_tuple, df_list in header_groups.items(): header_list = list(header_tuple) num_files = len(df_list) # 计算合并前的总行数 total_rows_before = sum(df.shape[0] for df in df_list) # 核心合并操作：使用 concat 进行纵向拼接 [ref_1] merged_df = pd.concat(df_list, axis=0, ignore_index=True, sort=False) # 生成安全的输出文件名 # 使用表头的前几个字段（如果存在）来命名文件，并清理非法字符 if header_list: # 取前两个字段，限制长度，并替换空格和特殊字符 name_parts = [str(col).strip().replace(' ', '_')[:15] for col in header_list[:2]] safe_name = '_'.join(name_parts) if name_parts else f'group_{group_index}' safe_name = ''.join(c for c in safe_name if c.isalnum() or c == '_') else: safe_name = f'header_group_{group_index}' output_filename = f"merged_group_{group_index:03d}_{safe_name}.csv" output_path = os.path.join(output_dir, output_filename) # 保存合并后的DataFrame到CSV try: merged_df.to_csv(output_path, index=False, encoding='utf-8-sig') # utf-8-sig 支持Excel中文打开 save_status = '成功' except Exception as e: save_status = f'失败({str(e)[:30]})' output_path = None # 记录本次合并的摘要信息 summary_report.append({ '分组编号': group_index, '表头列数': len(header_list), '表头预览': str(header_list[:3]) + ('...' if len(header_list) > 3 else ''), '包含文件数': num_files, '合并前行数': total_rows_before, '合并后行数': merged_df.shape[0], '输出文件': output_filename if output_path else 'N/A', '保存状态': save_status }) # 控制台进度输出 status_icon = '✅' if save_status == '成功' else '❌' print(f"{status_icon} 分组 {group_index}: 合并了 {num_files} 个文件 -> {output_filename} " f"({total_rows_before}行 -> {merged_df.shape[0]}行)") group_index += 1 return summary_report def generate_final_report(load_log, summary_report, output_dir): """ 生成并显示加载和合并的最终报告。 """ print("\n" + "=" * 60) print("处理完成！详细报告如下：") print("=" * 60) # 1. 文件加载情况汇总 print("\n--- 文件加载汇总 ---") load_df = pd.DataFrame(load_log) if not load_df.empty: # 统计成功与失败数量 success_count = (load_df['状态'] == '成功').sum() fail_count = len(load_df) - success_count print(f"共处理 {len(load_df)} 个文件: {success_count} 个成功, {fail_count} 个失败。") if fail_count > 0: print("\n失败文件详情:") print(load_df[load_df['状态'] == '失败'][['文件', '备注']].to_string(index=False)) else: print("没有文件被加载。") # 2. 表头分组与合并情况汇总 print("\n--- 表头分组与合并汇总 ---") if summary_report: summary_df = pd.DataFrame(summary_report) # 简化显示列 display_cols = ['分组编号', '表头列数', '包含文件数', '合并前行数', '合并后行数', '输出文件', '保存状态'] print(summary_df[display_cols].to_string(index=False)) print(f"\n所有合并文件已保存至: {os.path.abspath(output_dir)}") # 特别提醒：表头分组数量 unique_header_groups = len(summary_report) print(f"\n发现 {unique_header_groups} 种不同的表头结构。") if unique_header_groups == 1: print("提示：所有文件的表头完全一致，数据已合并为一个文件。") else: print("提示：存在多种表头结构，数据已按表头分组为多个文件。") else: print("没有生成任何合并文件。") def main(): """ 主函数，串联整个流程。 """ # 步骤1：交互式收集CSV文件路径 csv_files = collect_csv_paths_interactively() if not csv_files: return # 步骤2：加载CSV并按表头分组 print(f"\n[正在加载并分析 {len(csv_files)} 个CSV文件...]") groups, log = load_csv_and_group_by_header(csv_files) if not groups: print("错误：未能成功加载任何CSV文件的数据，程序退出。") return print(f"\n文件分析完成。共形成 {len(groups)} 个不同的表头分组。") # 步骤3：合并数据并导出 print("\n[开始合并数据...]") output_directory = "./merged_results" # 可修改为自定义输出目录 summary = merge_and_export_groups(groups, output_directory) # 步骤4：生成最终报告 generate_final_report(log, summary, output_directory) # 程序入口 if __name__ == "__main__": main() ``` ### **3. 核心流程与技术要点解析** 程序执行流程与关键点如下表所示： | 步骤 | 功能函数 | 关键操作与技术要点 | 参考资料 | | :--- | :--- | :--- | :--- | | **1. 路径收集** | `collect_csv_paths_interactively` | 循环`input()`获取路径，验证文件存在性及`.csv`后缀。使用`sys.exit(0)`处理无输入退出。 | - | | **2. 数据加载与分组** | `load_csv_and_group_by_header` | `pd.read_csv()`读取文件 [ref_1]。`df.columns.tolist()`获取表头列表 [ref_3]。`tuple(header_list)`将表头转为字典键进行分组。 | [ref_1], [ref_3] | | **3. 数据合并与导出** | `merge_and_export_groups` | `pd.concat(df_list, axis=0, ...)`纵向合并相同表头的所有DataFrame [ref_1]。`merged_df.to_csv()`将结果保存为新CSV文件。 | [ref_1] | | **4. 报告生成** | `generate_final_report` | 使用`pandas.DataFrame`整理日志和摘要，格式化输出处理结果和统计数据。 | - | **代码运行示例与输出**： ``` ============================================================ CSV文件表头比对与合并工具 ============================================================ 请逐个输入CSV文件的完整路径... 请输入第 1 个CSV文件路径 (或直接回车结束): ./data/sales_jan.csv √ 已添加：./data/sales_jan.csv 请输入第 2 个CSV文件路径 (或直接回车结束): ./data/sales_feb.csv √ 已添加：./data/sales_feb.csv 请输入第 3 个CSV文件路径 (或直接回车结束): ./data/inventory.csv √ 已添加：./data/inventory.csv 输入结束。共收集到 3 个路径。 [正在加载并分析 3 个CSV文件...] [1/3] 正在处理: sales_jan.csv √ 成功加载，表头: ['Date', 'Product', 'Sales'] [2/3] 正在处理: sales_feb.csv √ 成功加载，表头: ['Date', 'Product', 'Sales'] [3/3] 正在处理: inventory.csv √ 成功加载，表头: ['Product_ID', 'Name', 'Stock'] 文件分析完成。共形成 2 个不同的表头分组。 [开始合并数据...] 合并结果将输出至目录: /project/merged_results ✅ 分组 1: 合并了 2 个文件 -> merged_group_001_Date_Product.csv (150行 -> 300行) ✅ 分组 2: 合并了 1 个文件 -> merged_group_002_Product_ID_Name.csv (50行 -> 50行) ============================================================ 处理完成！详细报告如下： ============================================================ --- 文件加载汇总 --- 共处理 3 个文件: 3 个成功, 0 个失败。 --- 表头分组与合并汇总 --- 分组编号 表头列数 包含文件数 合并前行数 合并后行数 输出文件 保存状态 1 3 2 150 300 merged_group_001_Date_Product.csv 成功 2 3 1 50 50 merged_group_002_Product_ID_Name.csv 成功 所有合并文件已保存至: /project/merged_results 发现 2 种不同的表头结构。 提示：存在多种表头结构，数据已按表头分组为多个文件。 ``` ### **4. 高级扩展与场景适配** 上述方案基于表头完全一致进行精确匹配。在实际应用中，可根据需求进行扩展，例如： 1. **表头模糊匹配与清洗**：如果不同文件的表头存在大小写、空格或同义词差异，可在比对前进行标准化处理。 ```python def normalize_header_for_comparison(header_list): """表头标准化函数，用于模糊匹配。""" normalized = [] for col in header_list: # 示例：转为小写、去除首尾空格、将下划线和连字符统一 col_str = str(col).strip().lower() col_str = col_str.replace('-', '_') normalized.append(col_str) return tuple(normalized) # 在加载函数中，使用标准化后的表头作为分组键 header_key = normalize_header_for_comparison(df.columns.tolist()) ``` 2. **处理无表头CSV或指定列**：若CSV文件第一行不是表头，可在`pd.read_csv()`时设置`header=None`，并通过`names`参数指定列名，或手动创建表头。 ```python # 读取无表头CSV，并指定列名 custom_headers = ['Col1', 'Col2', 'Col3'] df = pd.read_csv(file_path, header=None, names=custom_headers) ``` 3. **内存优化与大数据处理**：处理超大CSV时，可使用分块读取或`dask.dataframe`。 ```python # 使用pandas分块读取（Chunking） chunk_list = [] for chunk in pd.read_csv(large_file, chunksize=10000): chunk_list.append(chunk) df = pd.concat(chunk_list, axis=0) ``` 该方案的核心优势在于其**通用性和自动化**。用户只需提供文件路径，程序即可自动完成表头识别、比对、分类和合并，极大减少了人工核对和手动操作的工作量，尤其适用于定期需要合并多部门或多批次提交的具有相同模板的CSV报告的场景 [ref_1]。通过调整表头标准化函数，还可灵活适配不同数据源间表头命名规范不统一的问题。