Python批量转换CSV为DBF并处理中文编码的实现方案

# 安装必要的依赖库
# 在线安装方式
pip install dbf
pip install dbfread

# 或者离线安装指定版本的wheel包
pip install dbf-0.99.11-py3-none-any.whl

import csv
import dbf
import time
import unicodedata
import sys
import os
import io

# 设置标准输出编码为UTF-8，确保控制台正常显示中文
sys.stdout = io.TextIOWrapper(sys.stdout.buffer, encoding='utf-8')

def contains_chinese(text):
    """
    检测字符串是否包含中文字符
    中文字符在Unicode中属于CJK统一表意文字区块
    """
    for char in str(text):
        try:
            if 'CJK' in unicodedata.name(char, ''):
                return True
        except ValueError:
            continue
    return False

def calculate_field_length(data_list):
    """
    计算字段的最大长度，中文按双倍长度计算
    """
    max_length = 0
    for data in data_list:
        if contains_chinese(str(data)):
            # 中文字符占用两个字节
            current_length = len(str(data)) * 2
        else:
            current_length = len(str(data))
        
        if current_length > max_length:
            max_length = current_length
    
    # 设置最小长度为8，避免字段过短
    return max(8, max_length)

def csv_to_dbf(csv_file_path):
    """
    将单个CSV文件转换为DBF格式
    """
    # 获取文件名（不含扩展名）
    file_name = os.path.splitext(os.path.basename(csv_file_path))[0]
    dbf_file_path = os.path.join(os.path.dirname(csv_file_path), f"{file_name}.dbf")
    
    print(f"开始处理文件: {csv_file_path}")
    
    try:
        # 读取CSV文件，使用GBK编码处理中文
        with open(csv_file_path, newline='', encoding='gbk') as csvfile:
            csv_reader = csv.reader(csvfile)
            headers = next(csv_reader)  # 读取表头
            csv_data = list(csv_reader)  # 读取所有数据行
        
        print(f"CSV文件列数: {len(headers)}, 数据行数: {len(csv_data)}")
        
        # 构建DBF字段规格
        field_specs = []
        data_for_dbf = []
        
        # 为每个字段计算合适的长度
        for i, column in enumerate(headers):
            # 获取该列的所有数据
            column_data = [row[i] for row in csv_data]
            # 计算字段长度
            field_length = calculate_field_length([column] + column_data)
            
            # 创建字段规格，使用字符类型(C)
            field_spec = f"{column} C({field_length})"
            field_specs.append(field_spec)
        
        # 将字段规格连接成DBF要求的格式
        field_specs_str = ';'.join(field_specs)
        
        # 准备数据
        for row in csv_data:
            data_for_dbf.append(tuple(row))
        
        # 创建内存中的DBF表格
        table = dbf.Table(
            'temp_table',
            field_specs=field_specs_str,
            on_disk=False,
            codepage='cp936'  # 使用中文编码
        )
        
        # 打开表格并写入数据
        table.open(mode=dbf.READ_WRITE)
        for datum in tuple(data_for_dbf):
            table.append(datum)
        
        # 创建最终的DBF文件
        custom = table.new(
            filename=dbf_file_path,
            default_data_types=dict(
                C=dbf.Char,
                D=dbf.Date,
                L=dbf.Logical
            ),
        )
        
        # 将数据写入最终文件
        with custom:
            for record in table:
                custom.append(record)
        
        table.close()
        print(f"DBF文件已生成：{dbf_file_path}")
        
    except Exception as e:
        print(f"处理文件 {csv_file_path} 时发生错误: {str(e)}")
        raise

def list_files(folder_path, file_extension=None):
    """
    递归列出文件夹下的所有指定类型文件
    
    参数:
    - folder_path: 文件夹路径
    - file_extension: 文件扩展名，如未指定则返回所有文件
    
    返回值:
    - 文件完整路径列表
    """
    file_list = []
    for root, dirs, files in os.walk(folder_path):
        for file in files:
            if file_extension is None or file.endswith(file_extension):
                full_path = os.path.join(root, file)
                file_list.append(full_path)
    
    print(f"找到 {len(file_list)} 个{file_extension}文件")
    return file_list

def batch_convert_csv_to_dbf(csv_folder_path):
    """
    批量转换指定文件夹中的所有CSV文件为DBF格式
    """
    start_time = time.time()
    print(f"开始批量转换: {time.strftime('%Y-%m-%d %H:%M:%S')}")
    
    # 获取所有CSV文件
    csv_files = list_files(csv_folder_path, '.csv')
    
    if not csv_files:
        print("未找到任何CSV文件")
        return
    
    success_count = 0
    failed_files = []
    
    # 逐个处理CSV文件
    for csv_file in csv_files:
        try:
            csv_to_dbf(csv_file)
            success_count += 1
        except Exception as e:
            print(f"文件 {csv_file} 转换失败: {str(e)}")
            failed_files.append((csv_file, str(e)))
    
    # 输出处理结果统计
    end_time = time.time()
    processing_time = end_time - start_time
    
    print(f"
转换完成: {time.strftime('%Y-%m-%d %H:%M:%S')}")
    print(f"处理耗时: {processing_time:.2f}秒")
    print(f"成功转换: {success_count} 个文件")
    print(f"失败文件: {len(failed_files)} 个")
    
    if failed_files:
        print("
失败文件列表:")
        for file, error in failed_files:
            print(f"  {file}: {error}")

if __name__ == '__main__':
    """
    命令行调用方式
    使用方法: python csv_to_dbf.py <csv_folder_path>
    """
    try:
        # 验证命令行参数
        if len(sys.argv) != 2:
            print("使用方法: python csv_to_dbf.py <csv_folder_path>")
            print("示例: python csv_to_dbf.py D:/data/csv_files")
            sys.exit(1)
        
        csv_folder_path = sys.argv[1]
        
        # 检查文件夹是否存在
        if not os.path.exists(csv_folder_path):
            print(f"错误: 文件夹路径不存在: {csv_folder_path}")
            sys.exit(1)
        
        # 执行批量转换
        batch_convert_csv_to_dbf(csv_folder_path)
        
    except Exception as e:
        print(f"程序执行错误: {str(e)}")
        sys.exit(-1)

# 关键编码设置点
sys.stdout = io.TextIOWrapper(sys.stdout.buffer, encoding='utf-8')  # 控制台输出编码
with open(csv_file_path, encoding='gbk') as csvfile:  # CSV文件读取编码
table = dbf.Table(..., codepage='cp936')  # DBF文件编码

def calculate_field_length(data_list):
    """
    智能计算字段长度，特别处理中文字符
    中文字符在DBF中需要双倍存储空间
    """
    max_length = 0
    for data in data_list:
        if contains_chinese(str(data)):
            current_length = len(str(data)) * 2  # 中文双字节
        else:
            current_length = len(str(data))      # 英文单字节
        max_length = max(max_length, current_length)
    return max(8, max_length)  # 确保最小长度

# 支持自定义字段类型的扩展版本
def advanced_csv_to_dbf(csv_file_path, field_type_mapping=None):
    """
    支持自定义字段类型映射的高级转换函数
    """
    if field_type_mapping is None:
        field_type_mapping = {}
    
    # 实现字段类型自动检测和映射
    # 可根据数据内容自动推断数值型、日期型字段