json/csv数据处理工具

llmops/agents/data_manager.py

@@ -2,6 +2,8 @@ import pandas as pd
 import json
 from typing import List, Dict, Any, Tuple
 import os
+import csv
+import io
 
 
 class DataManager:
@@ -130,4 +132,161 @@ class DataManager:
             direction_counts = df['txDirection'].value_counts().to_dict()
             summary.append(f"收支分布: {direction_counts}")
 
-        return " | ".join(summary)
+        return " | ".join(summary)
+
+    @staticmethod
+    def load_data_from_csv_file(file_path: str) -> List[Dict[str, Any]]:
+        """
+        从CSV文件中加载数据
+        :param file_path:  json文件绝对路径
+        :return:
+        """
+
+        if not os.path.exists(file_path):
+            raise FileNotFoundError(f"数据文件不存在: {file_path}")
+
+        with open(file_path, 'r', encoding='utf-8') as f:
+            reader = csv.DictReader(f)
+            json_list = [row for row in reader]
+
+        if not isinstance(json_list, list):
+            raise ValueError("JSON文件内容必须是数组格式")
+
+        if not json_list:
+            raise ValueError("数据文件为空")
+
+        return json_list
+
+    def write_json_to_csv(json_data, csv_file_path, field_order=None) -> bool:
+        """
+        将符合 [{}] 结构的 JSON 对象写入 CSV 文件，并允许指定字段顺序
+
+        :param json_data: 符合 [{}] 结构的 JSON 对象，例如 [{
+            "txId": "TX202301050001",
+            "txDate": "2023-01-05",
+            "txTime": "09:15",
+            "txAmount": 3200,
+            "txBalance": 3200,
+            "txDirection": "收入",
+            "txSummary": "水稻销售收入 (优质粳稻)",
+            "txCounterparty": "金穗粮食贸易公司",
+            "createdAt": "2025-11-30 05:57"
+            }]
+        :param csv_file_path: CSV 文件的路径
+        :param field_order: 字段顺序列表，例如 ["txId", "txDate", "txTime"...]。如果未指定，则按字典键的顺序写入
+        :return 是否写入成功 True:成功 False:失败
+        """
+        succ = True
+        try:
+            # 将 JSON 数据转换为 Python 的列表
+            data = json.loads(json.dumps(json_data))
+
+            # 检查数据是否为空
+            if not data:
+                print("JSON 数据为空，无法写入 CSV 文件")
+                return False
+
+            # 如果未指定字段顺序，则使用第一个字典的键作为字段顺序
+            if field_order is None:
+                field_order = list(data[0].keys())
+
+            # 打开 CSV 文件并写入数据
+            with open(csv_file_path, mode='w', newline='', encoding='utf-8') as csv_file:
+                writer = csv.DictWriter(csv_file, fieldnames=field_order)
+
+                # 写入列名
+                writer.writeheader()
+
+                # 写入数据
+                writer.writerows(data)
+
+            print(f"数据已成功写入 {csv_file_path}")
+        except Exception as e:
+            print(f"写入 CSV 文件时发生错误：{e}")
+            succ = False
+
+        return succ
+
+    def write_json_to_csv(json_data, csv_file_path, field_order=None) -> bool:
+        """
+        将符合 [{}] 结构的 JSON 对象写入 CSV 文件，并允许指定字段顺序
+
+        :param json_data: 符合 [{}] 结构的 JSON 对象，例如 [{
+            "txId": "TX202301050001",
+            "txDate": "2023-01-05",
+            "txTime": "09:15",
+            "txAmount": 3200,
+            "txBalance": 3200,
+            "txDirection": "收入",
+            "txSummary": "水稻销售收入 (优质粳稻)",
+            "txCounterparty": "金穗粮食贸易公司",
+            "createdAt": "2025-11-30 05:57"
+            }]
+        :param csv_file_path: CSV 文件的路径
+        :param field_order: 字段顺序列表，例如 ["txId", "txDate", "txTime"...]。如果未指定，则按字典键的顺序写入
+        :return 是否写入成功 True:成功 False:失败
+        """
+        succ = True
+        try:
+            # 将 JSON 数据转换为 Python 的列表
+            data = json.loads(json.dumps(json_data))
+
+            # 检查数据是否为空
+            if not data:
+                print("JSON 数据为空，无法写入 CSV 文件")
+                return False
+
+            # 如果未指定字段顺序，则使用第一个字典的键作为字段顺序
+            if field_order is None:
+                field_order = list(data[0].keys())
+
+            # 打开 CSV 文件并写入数据
+            with open(csv_file_path, mode='w', newline='', encoding='utf-8') as csv_file:
+                writer = csv.DictWriter(csv_file, fieldnames=field_order)
+
+                # 写入列名
+                writer.writeheader()
+
+                # 写入数据
+                writer.writerows(data)
+
+            print(f"数据已成功写入 {csv_file_path}")
+        except Exception as e:
+            print(f"写入 CSV 文件时发生错误：{e}")
+            succ = False
+
+        return succ
+
+    @staticmethod
+    def json_to_csv_string(json_data: List[Dict[str, Any]], fieldnames: List[str]):
+        """
+        将 JSON 数据（格式为 [{}]）转换为 CSV 格式的字符串，并指定字段顺序。
+
+        :param json_data: JSON 数据，格式为 [{}]
+        :param fieldnames: 字段顺序列表
+        :return: CSV 格式的字符串
+        """
+        # 检查输入数据是否为空
+        if not json_data:
+            raise ValueError("JSON 数据为空")
+
+        # 检查字段顺序是否为空
+        if not fieldnames:
+            raise ValueError("字段顺序不能为空")
+
+        # 使用 StringIO 来生成 CSV 字符串
+        output = io.StringIO()
+        writer = csv.DictWriter(output, fieldnames=fieldnames)
+
+        # 写入表头
+        writer.writeheader()
+
+        # 写入每行数据
+        for item in json_data:
+            writer.writerow(item)
+
+        # 获取生成的 CSV 字符串
+        csv_string = output.getvalue()
+        output.close()
+
+        return csv_string