|
|
@@ -0,0 +1,97 @@
|
|
|
+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✗ 转换失败，请检查错误信息")
|
