



原理:纯文本拼接,不解析csv结构
import os
import glob
from typing import *
try:
from xbot.app.logging import trace as print
except:
from xbot import print
def merge_csv_files(folder_path, output_file_path):
# 获取所有CSV文件
csv_pattern = os.path.join(folder_path, "*.csv")
csv_files = glob.glob(csv_pattern)
if not csv_files:
raise ValueError("文件夹中没有找到CSV文件")
print(f"找到 {len(csv_files)} 个CSV文件,开始合并")
merged_lines = []
total_files_processed = 0
for file_path in csv_files:
filename = os.path.basename(file_path)
if os.path.getsize(file_path) == 0:
continue
# 尝试不同编码读取文件
content = None
for encoding in ['utf-8', 'gbk', 'gb2312', 'utf-8-sig']:
try:
with open(file_path, 'r', encoding=encoding, errors='ignore') as f:
content = f.read()
if content.strip():
break
except:
continue
if not content:
continue
lines = content.splitlines()
processed_lines = 0
for line in lines:
if line.strip():
merged_lines.append(line)
processed_lines += 1
if processed_lines > 0:
total_files_processed += 1
print(f"处理 {filename}: {processed_lines} 行")
if not merged_lines:
raise ValueError("所有CSV文件都无法正常读取或为空")
# 确保输出目录存在
output_dir = os.path.dirname(output_file_path)
if output_dir and not os.path.exists(output_dir):
os.makedirs(output_dir)
# 保存合并后的文件
with open(output_file_path, 'w', encoding='utf-8-sig', newline='') as f:
f.write('\n'.join(merged_lines))
print(f"合并完成: {total_files_processed} 个文件,{len(merged_lines)} 行数据")
print(f"保存至: {output_file_path}")
return output_file_path