WebDemo/utils/data/data_processor.py

import pandas as pd
import json
import yaml
import csv
from pathlib import Path
from core.config.config_manager import ConfigManager
from core.utils.logger import Logger
from core.singleton import Singleton


class DataProcessor(metaclass=Singleton):
    def __init__(self):
        self.config = ConfigManager()
        self.logger = Logger.get_logger()
        self.data_dir = Path(__file__).parent.parent.parent / 'resources' / 'test_data'

    def read_json(self, file_path):
        """读取JSON文件"""
        try:
            with open(file_path, 'r', encoding='utf-8') as f:
                return json.load(f)
        except Exception as e:
            self.logger.error(f"读取JSON文件失败: {file_path}, 错误: {str(e)}")
            raise

    def write_json(self, data, file_path):
        """写入JSON文件"""
        try:
            with open(file_path, 'w', encoding='utf-8') as f:
                json.dump(data, f, indent=4, ensure_ascii=False)
            self.logger.info(f"JSON文件已写入: {file_path}")
        except Exception as e:
            self.logger.error(f"写入JSON文件失败: {file_path}, 错误: {str(e)}")
            raise

    def read_yaml(self, file_path):
        """读取YAML文件"""
        try:
            with open(file_path, 'r', encoding='utf-8') as f:
                return yaml.safe_load(f)
        except Exception as e:
            self.logger.error(f"读取YAML文件失败: {file_path}, 错误: {str(e)}")
            raise

    def write_yaml(self, data, file_path):
        """写入YAML文件"""
        try:
            with open(file_path, 'w', encoding='utf-8') as f:
                yaml.dump(data, f, allow_unicode=True)
            self.logger.info(f"YAML文件已写入: {file_path}")
        except Exception as e:
            self.logger.error(f"写入YAML文件失败: {file_path}, 错误: {str(e)}")
            raise

    def read_csv(self, file_path, **kwargs):
        """读取CSV文件"""
        try:
            return pd.read_csv(file_path, **kwargs)
        except Exception as e:
            self.logger.error(f"读取CSV文件失败: {file_path}, 错误: {str(e)}")
            raise

    def write_csv(self, data, file_path, **kwargs):
        """写入CSV文件"""
        try:
            if isinstance(data, pd.DataFrame):
                data.to_csv(file_path, **kwargs)
            else:
                with open(file_path, 'w', newline='', encoding='utf-8') as f:
                    writer = csv.writer(f)
                    writer.writerows(data)
            self.logger.info(f"CSV文件已写入: {file_path}")
        except Exception as e:
            self.logger.error(f"写入CSV文件失败: {file_path}, 错误: {str(e)}")
            raise

    def read_excel(self, file_path, **kwargs):
        """读取Excel文件"""
        try:
            return pd.read_excel(file_path, **kwargs)
        except Exception as e:
            self.logger.error(f"读取Excel文件失败: {file_path}, 错误: {str(e)}")
            raise

    def write_excel(self, data, file_path, **kwargs):
        """写入Excel文件"""
        try:
            if isinstance(data, pd.DataFrame):
                data.to_excel(file_path, **kwargs)
            else:
                df = pd.DataFrame(data)
                df.to_excel(file_path, **kwargs)
            self.logger.info(f"Excel文件已写入: {file_path}")
        except Exception as e:
            self.logger.error(f"写入Excel文件失败: {file_path}, 错误: {str(e)}")
            raise

    def get_test_data(self, key, default=None):
        """获取测试数据"""
        try:
            # 从配置中获取测试数据
            test_data = self.config.get('test_data', {})

            # 按点号分隔的键路径查找
            keys = key.split('.')
            value = test_data

            for k in keys:
                if isinstance(value, dict) and k in value:
                    value = value[k]
                else:
                    return default

            return value
        except Exception as e:
            self.logger.error(f"获取测试数据失败: {key}, 错误: {str(e)}")
            return default

    def generate_test_data(self, template, count=1, **kwargs):
        """生成测试数据"""
        # 在实际应用中，可以实现根据模板生成测试数据的逻辑
        # 这里返回模拟数据
        if count == 1:
            return {
                'name': '测试用户',
                'email': 'test@example.com',
                'age': 30,
                'city': '测试城市'
            }
        else:
            return [
                {
                    'name': f'测试用户{i}',
                    'email': f'test{i}@example.com',
                    'age': 20 + i,
                    'city': f'测试城市{i}'
                }
                for i in range(1, count + 1)
            ]

    def validate_data(self, data, schema):
        """验证数据是否符合模式"""
        # 在实际应用中，可以实现数据验证逻辑
        # 这里返回模拟验证结果
        return {
            'valid': True,
            'errors': []
        }