如何为多数据源配置灵活的属性映射结构

绝对可以！你的思路非常棒——通过预设数据源的结构映射来统一字段提取逻辑，完美解决重复代码的问题。我来给你具体实现方案，用Python举例（从你的伪代码风格判断应该是用Python）：

基础实现：统一基础节点提取 #

首先我们可以定义一个数据源结构映射表，用lambda函数来封装每个数据源的基础节点获取逻辑，这样后续提取字段时就不用反复写if判断了：

# 定义每个数据源的基础节点获取逻辑
source_structures = {
    'source1': lambda row: row,
    'source2': lambda row: row['data'],
    'source3': lambda row: row[0]
}

# 统一的字段提取函数
def extract_user_info(source, row):
    # 根据数据源获取对应的基础节点
    base_node = source_structures[source](row)
    # 统一提取所有字段，无需重复if分支
    return {
        'first_name': base_node['first_name'],
        'last_name': base_node['last_name'],
        'age': base_node.get('age'),  # 可选字段用get避免KeyError
        # 其他需要提取的字段都在这里统一写
    }

调用的时候只需要传入数据源标识和原始行数据：

# 示例调用
row_source1 = {'first_name': 'Alice', 'last_name': 'Smith', 'age': 30}
row_source2 = {'data': {'first_name': 'Bob', 'last_name': 'Johnson'}}
row_source3 = [{'first_name': 'Charlie', 'last_name': 'Brown', 'age': 25}]

print(extract_user_info('source1', row_source1))
print(extract_user_info('source2', row_source2))
print(extract_user_info('source3', row_source3))

进阶优化：支持嵌套字段路径 #

如果不同数据源的字段嵌套路径也有差异（比如有的是row['contact']['email']，有的是row['email']），我们可以把配置做得更灵活，把字段路径也纳入映射：

# 更灵活的数据源配置：包含基础节点和字段路径
source_configs = {
    'source1': {
        'get_base': lambda row: row,
        'fields': {
            'first_name': 'first_name',
            'email': 'email'
        }
    },
    'source2': {
        'get_base': lambda row: row['data'],
        'fields': {
            'first_name': 'first_name',
            'email': 'contact.email'  # 用点分隔嵌套路径
        }
    },
    'source3': {
        'get_base': lambda row: row[0],
        'fields': {
            'first_name': 'first_name',
            'email': 'user_info.email'
        }
    }
}

# 辅助函数：根据路径提取嵌套字段
def get_nested_value(node, path):
    keys = path.split('.')
    for key in keys:
        node = node[key]
    return node

# 通用提取函数
def extract_user_info(source, row):
    config = source_configs[source]
    base_node = config['get_base'](row)
    user_info = {}
    for field, path in config['fields'].items():
        user_info[field] = get_nested_value(base_node, path)
    return user_info

为什么这种方式更好？ #

• 消除重复代码：不用再写大量结构相同的if分支，新增数据源只需要在配置里加一条记录
• 逻辑集中易维护：所有数据源的结构规则都集中在配置表中，字段提取逻辑统一管理
• 扩展性强：后续如果要加字段、改结构，只需要修改配置，不用动核心提取逻辑

内容的提问来源于stack exchange，提问作者A. Nurb