将Pandas与面向对象编程结合：复杂数据管理的教程指南

import pandas as pd
from pydantic import BaseModel, ValidationError, Field
from typing import List, Dict, Any, Optional

# 1. 定义一个Pydantic模型用于数据验证 (可选，但推荐)
class TreeNodeSchema(BaseModel):
    id: str
    parent_id: Optional[str] = None
    name: str
    value: float = Field(..., description="节点关联的数值数据")
    # 可以在这里添加更多字段和验证规则

# 2. 创建一个封装Pandas DataFrame的类
class TreeDataAnalyzer:
    """
    一个用于管理和分析树形数据的类，内部封装了Pandas DataFrame。
    """
    def __init__(self, data: pd.DataFrame):
        """
        初始化TreeDataAnalyzer实例。
        Args:
            data (pd.DataFrame): 包含树形结构数据的DataFrame。
                                 期望包含 'id', 'parent_id', 'name', 'value' 等列。
        """
        self._df = self._validate_and_process_data(data)
        print("TreeDataAnalyzer initialized with validated data.")

    def _validate_and_process_data(self, data: pd.DataFrame) -> pd.DataFrame:
        """
        内部方法：验证传入的DataFrame数据，并进行初步处理。
        """
        required_columns = ['id', 'name', 'value']
        if not all(col in data.columns for col in required_columns):
            raise ValueError(f"DataFrame缺少必要的列: {required_columns}")

        # 尝试使用Pydantic进行行级验证
        validated_records = []
        for _, row in data.iterrows():
            try:
                # 将DataFrame行转换为字典，并验证
                node = TreeNodeSchema(**row.to_dict())
                validated_records.append(node.model_dump()) # 使用model_dump()获取字典形式
            except ValidationError as e:
                print(f"数据验证失败，行: {row.to_dict()}，错误: {e}")
                raise # 或者选择跳过/记录错误

        return pd.DataFrame(validated_records)

    def get_df(self) -> pd.DataFrame:
        """
        获取内部的DataFrame副本，防止外部直接修改。
        """
        return self._df.copy()

    def add_node(self, node_id: str, parent_id: Optional[str], name: str, value: float):
        """
        向树结构中添加一个新节点。
        """
        new_data = {'id': node_id, 'parent_id': parent_id, 'name': name, 'value': value}
        try:
            # 验证新节点数据
            validated_node = TreeNodeSchema(**new_data)
            self._df = pd.concat([self._df, pd.DataFrame([validated_node.model_dump()])], ignore_index=True)
            print(f"Node '{name}' added.")
        except ValidationError as e:
            print(f"添加节点失败: {e}")

    def get_children(self, parent_id: str) -> pd.DataFrame:
        """
        获取指定父节点的所有直接子节点。
        """
        return self._df[self._df['parent_id'] == parent_id]

    def calculate_total_value(self) -> float:
        """
        计算所有节点的总值。
        """
        return self._df['value'].sum()

    def find_node_by_id(self, node_id: str) -> Optional[Dict[str, Any]]:
        """
        根据ID查找节点，并返回其数据字典。
        """
        node = self._df[self._df['id'] == node_id]
        if not node.empty:
            return node.iloc[0].to_dict()
        return None

    def update_node_value(self, node_id: str, new_value: float):
        """
        更新指定节点的数值。
        """
        if node_id not in self._df['id'].values:
            print(f"错误: 节点ID '{node_id}' 不存在。")
            return

        # 使用loc进行位置更新，确保性能和正确性
        self._df.loc[self._df['id'] == node_id, 'value'] = new_value
        print(f"Node '{node_id}' value updated to {new_value}.")

# 3. 实际使用
if __name__ == "__main__":
    # 模拟输入数据 (通常来自CSV, DB等)
    initial_data = pd.DataFrame([
        {'id': 'root', 'parent_id': None, 'name': 'Root Node', 'value': 100.0},
        {'id': 'a1', 'parent_id': 'root', 'name': 'Child A1', 'value': 50.0},
        {'id': 'a2', 'parent_id': 'root', 'name': 'Child A2', 'value': 75.0},
        {'id': 'b1', 'parent_id': 'a1', 'name': 'Grandchild B1', 'value': 20.0},
    ])

    try:
        # 实例化类
        analyzer = TreeDataAnalyzer(initial_data)

        # 执行操作
        print("\n所有数据:")
        print(analyzer.get_df())

        print(f"\n总值: {analyzer.calculate_total_value()}")

        print("\nRoot节点的子节点:")
        print(analyzer.get_children('root'))

        analyzer.add_node('c1', 'a2', 'Grandchild C1', 30.0)
        print("\n添加节点后数据:")
        print(analyzer.get_df())

        analyzer.update_node_value('a1', 60.0)
        print("\n更新节点值后数据:")
        print(analyzer.get_df())

        node_data = analyzer.find_node_by_id('b1')
        if node_data:
            print(f"\n找到节点 B1: {node_data}")

        # 尝试添加无效数据 (例如，缺少必要字段)
        invalid_data = pd.DataFrame([{'id': 'd1', 'name': 'Invalid Node'}])
        # analyzer_invalid = TreeDataAnalyzer(invalid_data) # 这会引发ValueError

    except ValueError as e:
        print(f"\n初始化失败: {e}")
    except ValidationError as e:
        print(f"\n数据验证失败: {e}")