如何导入数据清洗文件中的多个DataFrame并实现类封装用于仪表盘开发

完全可以通过类组织数据处理逻辑，这种封装方式非常适配你「后续切换数据源、前后端职责分离」的长期目标。

• 你当前的所有处理逻辑都是World Bank通用的数据处理流程（读入原始数据→合并元数据→清洗列→宽转长→分组统计），类可以把这些通用逻辑统一封装，避免零散的全局变量污染命名空间
• 后续从CSV切换到API时，只需要修改类内部的「数据加载」实现逻辑，main.py里的调用代码完全不需要改动，完美实现职责分离
• 可以把所有处理好的DataFrame统一通过类的公开方法返回，不用在main.py里零散导入几十名变量

import pandas as pd
import os

class WorldBankDataProcessor:
    # 统一管理数据源路径，后续切API直接改这里或传参即可
    def __init__(self, data_dir="Data/", debug=False):
        self.data_dir = data_dir
        self.debug = debug
        if self.debug:
            print(os.getcwd())

    # 私有方法：加载原始数据，后续切API直接重写这个方法
    def _load_raw_data(self):
        # 修正原代码笔误：pd.r → pd.read_csv，同时移除未使用的csv导入
        fdi_data = pd.read_csv(f"{self.data_dir}fdi_data.csv")
        fdi_meta = pd.read_csv(f"{self.data_dir}fdi_metadata.csv")

        debt_data = pd.read_csv(f"{self.data_dir}debt_data.csv")
        debt_meta = pd.read_csv(f"{self.data_dir}debt_metadata.csv")

        gdp_percap_data = pd.read_csv(f"{self.data_dir}gdp_percap_data.csv", header=2)
        gdp_percap_meta = pd.read_csv(f"{self.data_dir}gdp_percap_metadata.csv")

        gov_exp_data = pd.read_csv(f"{self.data_dir}gov_exp_data.csv", header=2)
        gov_exp_meta = pd.read_csv(f"{self.data_dir}gov_exp_metadata.csv")

        pop_data = pd.read_csv(f"{self.data_dir}pop_data.csv", header=2)
        pop_meta = pd.read_csv(f"{self.data_dir}pop_metadata.csv")

        return {
            "fdi": (fdi_data, fdi_meta),
            "debt": (debt_data, debt_meta),
            "gdp_percap": (gdp_percap_data, gdp_percap_meta),
            "gov_exp": (gov_exp_data, gov_exp_meta),
            "pop": (pop_data, pop_meta)
        }

    # 私有方法：合并原始数据与元数据，和原逻辑完全一致
    def _merge_with_meta(self, data, metadata):
        return pd.merge(
            data,
            metadata,
            on="Country Code",
            how="inner"
        )

    # 私有方法：删除不需要的列，和原逻辑完全一致
    def _drop_unused_cols(self, data):
        return data.drop(
            ["Country Code","Indicator Name","Indicator Code","TableName","SpecialNotes","Unnamed: 5"],
            axis=1
        )

    # 私有方法：宽表转长表，和原逻辑完全一致
    def _reshape_to_long(self, data, value_name):
        data = data.melt(["Country Name","Region","IncomeGroup"]).reset_index()
        data = data.rename(columns={"variable": "Year", "value": value_name})
        return data.drop("index", axis=1)

    # 私有方法：按维度分组计算均值，和原逻辑完全一致
    def _calc_avg(self, data, col_cal, group_type):
        group_map = {
            "country": "Country Name",
            "income_group": "IncomeGroup",
            "region": "Region"
        }
        return data.groupby(group_map[group_type])[col_cal].mean().reset_index()

    # 公开方法：供外部调用，返回所有处理好的数据集
    def get_all_processed_data(self):
        raw_data = self._load_raw_data()
        processed = {}

        # 遍历所有指标，统一执行处理流程，消除原代码重复编写的冗余
        metric_config = [
            ("fdi", "FDI"),
            ("debt", "Debt"),
            ("gdp_percap", "GDP per Cap"),
            ("gov_exp", "Gov Expend."),
            ("pop", "Population")
        ]

        for metric_name, value_col in metric_config:
            data, meta = raw_data[metric_name]
            # 三步基础处理
            merged = self._merge_with_meta(data, meta)
            dropped = self._drop_unused_cols(merged)
            long_df = self._reshape_to_long(dropped, value_col)
            # 存入长表原始数据
            processed[f"{metric_name}_long"] = long_df
            # 计算三个维度的均值，和原变量命名逻辑完全对齐
            processed[f"avg_cn_{metric_name}"] = self._calc_avg(long_df, value_col, "country")
            processed[f"avg_ig_{metric_name}"] = self._calc_avg(long_df, value_col, "income_group")
            processed[f"avg_rg_{metric_name}"] = self._calc_avg(long_df, value_col, "region")

        return processed

from clean_data import WorldBankDataProcessor

# 初始化处理器，后续切API只需要改这里的参数或类内部逻辑
data_processor = WorldBankDataProcessor(debug=False)
# 一次性获取所有处理好的数据集
all_datasets = data_processor.get_all_processed_data()

# 直接取用需要的DataFrame，比如：
fdi_avg_country = all_datasets["avg_cn_fdi"]
pop_long = all_datasets["pop_long"]
# 后续直接传给仪表盘组件即可

等你要从CSV切到API时，只需要重写WorldBankDataProcessor类里的_load_raw_data方法，把原来的pd.read_csv替换成API请求、JSON转DataFrame的逻辑，上面的其他处理方法、main.py里的调用代码完全不用改，彻底实现数据源和业务逻辑、Web组件的解耦。

内容的提问来源于stack exchange，提问作者nizarhamood