tx_flow_analysis/data_files/csv2json.py

import csv
import json
import os


def csv_to_json(csv_file_path, json_file_path=None, encoding='utf-8'):
    """
    将CSV文件转换为JSON文件

    参数:
    csv_file_path: CSV文件路径
    json_file_path: JSON文件输出路径（可选，默认为CSV同目录同名.json）
    encoding: 文件编码格式
    """

    # 检查CSV文件是否存在
    if not os.path.exists(csv_file_path):
        print(f"错误：找不到CSV文件 '{csv_file_path}'")
        return False

    # 设置JSON输出路径
    if json_file_path is None:
        base_name = os.path.splitext(csv_file_path)[0]
        json_file_path = f"{base_name}.json"

    try:
        # 读取CSV文件
        with open(csv_file_path, 'r', encoding=encoding) as csv_file:
            # 使用csv.DictReader读取数据
            csv_reader = csv.DictReader(csv_file)

            # 将数据转换为列表
            data = list(csv_reader)

            # 数据类型转换（可选）
            for item in data:
                # 尝试将数值字段转换为浮点数
                for key in ['txAmount', 'txBalance']:
                    if key in item and item[key]:
                        try:
                            item[key] = float(item[key])
                        except ValueError:
                            pass  # 如果转换失败，保持原样

                # 创建完整的日期时间字段（可选）
                if 'txDate' in item and 'txTime' in item:
                    item['txDateTime'] = f"{item['txDate']} {item['txTime']}"

            print(f"成功读取 {len(data)} 条记录")

            # 写入JSON文件
            with open(json_file_path, 'w', encoding='utf-8') as json_file:
                json.dump(data, json_file, ensure_ascii=False, indent=2)

            print(f"JSON文件已保存到: {json_file_path}")
            print(f"文件大小: {os.path.getsize(json_file_path)} 字节")

            # 显示前几条记录作为示例
            print("\n前3条记录示例:")
            for i, record in enumerate(data[:3]):
                print(f"\n记录 {i + 1}:")
                for key, value in record.items():
                    print(f"  {key}: {value}")

            return True

    except FileNotFoundError:
        print(f"错误：找不到文件 '{csv_file_path}'")
        return False
    except UnicodeDecodeError:
        print(f"错误：编码问题，请尝试使用不同的编码格式（如'gbk'）")
        return False
    except Exception as e:
        print(f"错误：处理文件时发生错误 - {str(e)}")
        return False


# 使用示例
if __name__ == "__main__":
    # 请根据实际情况修改这些参数
    csv_file = "交易流水样例数据.csv"  # 您的CSV文件名
    json_file = "test_temp_agriculture_transaction_flow.json"  # 输出的JSON文件名
    file_encoding = "utf-8"  # 文件编码

    print("开始转换CSV到JSON...")
    print(f"CSV文件: {csv_file}")
    print(f"JSON文件: {json_file}")
    print(f"编码格式: {file_encoding}")
    print("-" * 50)

    # 执行转换
    success = csv_to_json(csv_file, json_file, file_encoding)

    if success:
        print("\n✓ 转换完成！")
    else:
        print("\n✗ 转换失败，请检查错误信息")