基于pandas自动化的csv信息提取保存的脚本

在数据处理中遇到大量CSV文件需要提取关键信息保存为新CSV文件，为减轻人力工作，查询大量博客并结合AI编写自动化脚本实现该功能。
测试学习pandas模块能力

 

1. import pandas as pd
2. from pathlib import Path
3. import chardet
4. import os
6. input_file = '123/LURM.csv' #待处理文件
8. #检测文件编码
9. with open(input_file, 'rb') as f:
10.     raw_data = f.read(10000)  # 只读取前10000字节进行检测
11.     result = chardet.detect(raw_data)
12.     encoding = result['encoding']
13.     print(encoding)
15. #导入文件
16. df = pd.read_csv(input_file, encoding=encoding)
17. #print(df)
18. print(f"原始数据列名: {list(df.columns)}")
20. #提取指定列
21. columns_to_extract = ['vn', '报文类型', '采集时间', '上报时间', '车辆状态', '采集时间', '定位有效性', '纬度类型', '经度类型', '经度', '纬度']
22. extracted_df = df[columns_to_extract]
23. print(extracted_df)
25. #保存文件
26. output_file = 'extracted_result.csv'
27. #extracted_df.to_csv(output_file, index=False, encoding='utf-8-sig')
28. extracted_df.to_csv(output_file, index=False, encoding=encoding)

复制代码

上面这段代码实现单文件信息提取并保存。要实现CSV文件处理需要使用到pandas模块，该模块是大数据文件处理较为热门模块。有编程基础的可以看注释，基本可以理解，如果没有建议使用AI工具指导。
 
正式处理程序

 

1. """ CSV文件提取保存程序
2.     代码第一版时间20250911
3. """
5. import pandas as pd
6. from pathlib import Path
7. import chardet
8. import os
10. def detect_encoding(file_path):
11.     """检测文件编码"""
12.     try:
13.         with open(file_path, 'rb') as f:
14.             raw_data = f.read(10000)  # 只读取前10000字节进行检测
15.             result = chardet.detect(raw_data)
16.             return result['encoding']
17.     except Exception as e:
18.         print(f"编码检测失败: {e}")
19.         return None
21. def read_csv_with_auto_encoding(file_path):
22.     """自动检测编码并读取CSV文件"""
23.     # 常见的编码尝试顺序
24.     encodings_to_try = ['utf-8', 'gbk', 'gb2312', 'gb18030', 'latin1', 'cp1252']
25.    
26.     # 首先尝试自动检测编码
27.     detected_encoding = detect_encoding(file_path)
28.     if detected_encoding:
29.         encodings_to_try.insert(0, detected_encoding)
30.    
31.     for encoding in encodings_to_try:
32.         try:
33.             df = pd.read_csv(file_path, encoding=encoding)
34.             print(f"✓ 成功使用 {encoding} 编码读取文件")
35.             return df, encoding
36.         except UnicodeDecodeError:
37.             continue
38.         except Exception as e:
39.             # 如果不是编码问题，可能是其他错误
40.             if 'codec' not in str(e).lower():
41.                 raise e
42.    
43.     # 如果所有编码都失败，尝试使用错误处理
44.     try:
45.         df = pd.read_csv(file_path, encoding='utf-8', errors='ignore')
46.         print("✓ 使用utf-8编码并忽略错误字符读取文件")
47.         return df, 'utf-8-with-errors-ignored'
48.     except Exception as e:
49.         raise Exception(f"所有编码尝试都失败: {e}")
50. def save_dataframe_to_csv(extracted_df, file_name, encoding_used, output_folder="output"):
51.     """
52.     保存提取后的DataFrame到CSV文件
53.    
54.     参数:
55.         extracted_df (pd.DataFrame): 要保存的数据框
56.         file_name (str): 原始文件名（不含扩展名）
57.         encoding_used (str): 使用的编码格式
58.         output_folder (str): 输出文件夹名，默认为"output"
59.    
60.     返回:
61.         bool: 保存成功返回True，失败返回False
62.     """
63.     try:
64.         # 创建输出文件夹（如果不存在）
65.         output_path = Path(output_folder)
66.         output_path.mkdir(exist_ok=True)
67.         
68.         # 构建输出文件名：原文件名 + "o" + .csv
69.         output_file = output_path / f"{file_name}_o.csv"
70.         
71.         # 检查文件是否已存在
72.         if output_file.exists():
73.             raise FileExistsError(f"文件已存在: {output_file}")
74.         
75.         # 保存为CSV文件
76.         extracted_df.to_csv(output_file, index=False, encoding=encoding_used)
77.         
78.         print(f"✓ 数据已成功保存到: {output_file}")
79.         print(f"  使用编码: {encoding_used}")
80.         print(f"  数据形状: {extracted_df.shape[0]} 行, {extracted_df.shape[1]} 列")
81.         
82.         return True
83.         
84.     except FileExistsError as e:
85.         print(f"⚠️  文件已存在错误: {e}")
86.         return False
87.     except Exception as e:
88.         print(f"❌ 保存文件失败: {e}")
89.         return False
90.    
91. def read_all_csv_files(folder_path):
92.     """
93.     读取指定文件夹下的所有CSV文件（支持自动编码检测）
94.     """
95.     if not os.path.exists(folder_path):
96.         print(f"错误: 文件夹 '{folder_path}' 不存在")
97.         return {}
98.    
99.     csv_files = list(Path(folder_path).glob("*.csv"))
100.    
101.     if len(csv_files) == 0:
102.         print(f"警告: 在 '{folder_path}' 中没有找到CSV文件")
103.         return {}
104.    
105.     dataframes = {}
106.    
107.     print(f"找到 {len(csv_files)} 个CSV文件:")
108.     print("-" * 50)
109.    
110.     for csv_file in csv_files:
111.         try:
112.             print(f"\n正在处理: {csv_file.name}")
113.             df, encoding_used = read_csv_with_auto_encoding(csv_file)
115.             print(f"✓ 成功读取: {csv_file.name}")
116.             print(f"  编码: {encoding_used}")
117.             print(f"  数据形状: {df.shape[0]} 行, {df.shape[1]} 列")
119.             print("-" * 50)
120.             print(f"原始数据列名: {list(df.columns)}")
121.             columns_to_extract = ['vn', '报文类型', '采集时间', '上报时间', '车辆状态', '采集时间', '定位有效性', '纬度类型', '经度类型', '经度', '纬度']
122.             extracted_df = df[columns_to_extract]
123.             #print(extracted_df)
124.             
126.             # 使用文件名（不含路径和扩展名）作为键
127.             file_name = csv_file.stem
128.             dataframes[file_name] = df
130.             #保存提取后信息到文件
131.             save_success = save_dataframe_to_csv(extracted_df, file_name, encoding_used)
132.             if save_success:
133.                 print("✓ 提取的数据已成功保存")
134.             else:
135.                 print("⚠️  提取的数据保存失败")
136.             
137.         except Exception as e:
138.             print(f"✗ 无法读取 {csv_file.name}: {str(e)}")
139.    
140.     return dataframes
145. # 主程序
146. if __name__ == "__main__":
147.    
148.     # 设置你的文件夹路径
149.     folder_path = "C:/Users/Downloads/LURM"
150.    
151.     # 读取所有CSV文件
152.     all_data = read_all_csv_files(folder_path)
153.    
154.     print(f"\n{'='*50}")
155.     print(f"总结: 成功读取了 {len(all_data)} 个CSV文件")
156.    
157.     # 显示每个文件的基本信息
158.     # for name, df in all_data.items():
159.     #     print(f"\n{name}:")
160.     #     print(f"  形状: {df.shape}")
161.     #     if len(df.columns) > 0:
162.     #         print(f"  列名: {list(df.columns)[:5]}{'...' if len(df.columns) > 5 else ''}")
163.     #     if len(df) > 0:
164.     #         print(f"  数据预览:")
165.     #         print(df.head(2))

复制代码

正式程序的核心代码和测试程序一样，只不过添加大量错误检测和现实文件信息提取的代码（该段代码已注释）。
代码写的比较烂，勿喷，有问题请回复指出。
 

来源：程序园用户自行投稿发布，如果侵权，请联系站长删除
免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！