python - 使用 Pandas 链接数据转换方法的设计模式

Question

我每月收到一个包含一些列的 csv 文件。不管我收到什么列，如果可能的话，我应该输出一个带有 C1、C2、C3、...C29、C30 列的 csv + 一个包含我所采取步骤的日志文件。
我知道，我的数据转换顺序应该是 t1、t2、t3、t4、t5。

t1 generates columns C8, C9, C12, C22 using C1, C2, C3, C4
t2 generates columns C10, C11, C17 using C3, C6, C7, C8
t3 generates columns C13, C14, C15, C16 using C5, C8, C10, C11, C22
t4 generates columns C18, C19, C20, C21, C23, C24, C25 using C13, C15
t5 generates columns C26, C27, C28, C29, C30 using C5, C19, C20, C21

我无法控制我在输入数据中得到的列。
如果我的输入数据有 C1、C2、C3、C4、C5、C6、C7 列，我可以生成所有 C1 ... C30 列。
如果我的输入数据有 C1、C2、C3、C4、C5、C6、C7、C8、C10、C11、C17 列，我可以生成所有 C1 ... C30 列，但我应该跳过 t2，因为它不是必需的
如果我的输入数据有 C1、C2、C3、C4、C6、C7，我只能做 t1、t2、t3、t4。我无法运行 t5，因此我应该创建仅包含 NaN 值的 C26、C27、C28、C29、C30 列，并且我应该在日志中添加“无法执行 t5 转换，因为缺少 C5。C26、C27、C28、C29、C30 是填充 NaN 值"
我的 t1、t2、t3、t4、t5 已经创建，但我不知道如何以优雅的方式组织代码，以使代码重复最少。
我必须在很短的时间内开发我的代码。因此，我所有的 t1、t2、t3、t4、t5 方法看起来像

def ti(df):
    output_cols = get_output_cols()
    if output_cols_already_exist(df, output_cols):
        return df, "{} skipped, the output cols {} already exist".format(inspect.stack()[0][3], output_cols)
    else:
        input_cols = get_required_input_cols()
        missing_cols = get_missing_cols(df, input_cols):
        if missing_cols == []:
            // do stuff
            log = "Performed {} transformation. Created {} columns".format(inspect.stack()[0][3], input_cols)
        else:
            for col in input_cols:
                df[col] = np.NaN
            log = "Cannot perform {} transformation because {} columns are missing. {} are filled with NaN values".format(inspect.stack()[0][3], missing_cols, output_cols)

另外，我按以下方式使用这些功能:

text = ""
df = pd.read_csv(input_path)
df, log_text = t1(df)
text = text + log_text + "\n"
df, log_text = t2(df)
text = text + log_text + "\n"
df, log_text = t3(df)
text = text + log_text + "\n"
df, log_text = t4(df)
text = text + log_text + "\n"
df, log_text = t5(df)
text = text + log_text + "\n"
df.to_csv("output_data.csv", index = False)
logging.info(text)

如您所见，我的代码既丑陋又重复。现在我有时间重构它，但我不知道最好的方法是什么。我还希望我的代码是可扩展的，因为我也在考虑添加 t6 转换。你能帮我提供一些我可以遵循的方向/设计模式吗？ (我也开放使用 Pandas 以外的其他 python 库)

Answer 1

因为，在 Python 中，函数是 first-class objects ，您可以重构您的代码以概括您的 t[i]通过提取似乎区分它们的功能(do stuff 部分)，使其成为辅助函数并将其视为参数。
您还可以通过在列表上迭代来避免在调用函数(t1、t2 等或此后重构的帮助程序版本)时重复。
最后，使用 f-strings帮助使您的代码更具可读性。
像这样的东西:

# t function takes a dataframe and a function as parameters
def t(df, do_stuff_func):
    output_cols = get_output_cols()
    if output_cols_already_exist(df, output_cols):
        return (
            df,
            (
                f"{inspect.stack()[0][3]} skipped, "
                f"the output cols {output_cols} already exist",
            ),
        )
    else:
        input_cols = get_required_input_cols()
        missing_cols = get_missing_cols(df, input_cols)
        if missing_cols == []:
            # Call the helper function
            do_stuff_func()
            log = (
                f"Performed {inspect.stack()[0][3]} transformation."
                f"Created {input_cols} columns"
            )
        else:
            for col in input_cols:
                df[col] = np.NaN
            log = (
                f"Cannot perform {inspect.stack()[0][3]} transformation"
                f"because {missing_cols} columns are missing. "
                f"{output_cols} are filled with NaN values"
            )

# Define the five new 'do_stuff' functions
def do_stuff1():
    pass
...
def do_stuff5():
    pass

# Store the functions
do_stuff_funcs = [do_stuff1, do_stuff2, do_stuff3, do_stuff4, do_stuff5]

# Call t function in combination with df and do_stuff_funcs helpers
for do_stuff_func in do_stuff_funcs:
    df, log_text = t(df, do_stuff_func)
    text = text + log_text + "\n"

# Save the results
df.to_csv("output_data.csv", index = False)
logging.info(text)