构建选定特征字典：20K对象特征提取与代码灵活性需求

看来你是遇到了部分对象缺失目标特征导致代码报错/结果异常的问题对吧？结合你20K量级对象+动态变化特征列表的场景，我给你一套灵活又健壮的解决方案，既能适配特征列表的变动，又能优雅处理缺失特征的情况：

核心需求拆解 #

• 支持动态更新的特征提取列表，不用改核心代码就能调整提取目标
• 处理对象缺失特征的场景（避免报错、返回默认值或记录问题）
• 高效处理20K量级的对象，保证性能

基础版：兼容字典/类对象+默认值处理 #

先写一个最通用的提取函数，兼容字典和类实例两种常见对象格式，缺失特征时返回指定默认值：

def extract_features(obj, feature_list, default=None):
    """
    从对象中提取指定特征，缺失时返回默认值
    
    参数:
        obj: 待提取特征的对象（支持字典或带属性的类实例）
        feature_list: 需要提取的特征名称列表
        default: 特征缺失时的默认值，默认为None
    
    返回:
        提取后的特征字典
    """
    extracted = {}
    # 提前判断对象类型，避免循环内重复检查
    if isinstance(obj, dict):
        getter = lambda feat: obj.get(feat, default)
    else:
        getter = lambda feat: getattr(obj, feat, default)
    
    for feature in feature_list:
        extracted[feature] = getter(feature)
    return extracted

# 实际使用示例
# 假设你的批量对象存在objects列表中，目标特征列表是target_features
target_features = ['user_id', 'age', 'signup_date', 'preference']
processed_results = [extract_features(obj, target_features) for obj in objects]

进阶版：添加缺失特征日志记录 #

如果需要定位哪些对象缺失了哪些特征，可以给函数加上日志功能，方便后续排查问题：

import logging

# 配置日志输出
logging.basicConfig(level=logging.INFO, format="%(asctime)s - %(message)s")
logger = logging.getLogger(__name__)

def extract_features_with_log(obj, feature_list, obj_identifier=None, default=None):
    extracted = {}
    missing_features = []
    
    if isinstance(obj, dict):
        getter = lambda feat: obj.get(feat, default)
    else:
        getter = lambda feat: getattr(obj, feat, default)
    
    for feature in feature_list:
        value = getter(feature)
        extracted[feature] = value
        if value == default:
            missing_features.append(feature)
    
    # 记录缺失特征的对象信息
    if missing_features:
        obj_id = obj_identifier if obj_identifier else f"ID:{id(obj)}"
        logger.info(f"对象 {obj_id} 缺失特征: {', '.join(missing_features)}")
    
    return extracted

# 使用示例：给每个对象加索引方便定位
processed_results = []
for idx, obj in enumerate(objects):
    processed = extract_features_with_log(obj, target_features, obj_identifier=idx)
    processed_results.append(processed)

高级优化：动态配置+并行处理 #

1. 特征列表动态加载 #

把特征列表存在配置文件里（比如JSON），不用修改代码就能更新提取目标：

import json

# 从配置文件加载特征列表
with open('feature_config.json', 'r', encoding='utf-8') as f:
    target_features = json.load(f)

feature_config.json内容示例：

["user_id", "age", "signup_date", "preference", "last_login"]

2. 并行处理提升速度 #

如果20K对象的处理速度不够快，可以用多进程并行处理（注意对象要支持序列化）：

from multiprocessing import Pool

def process_object_batch(batch):
    """处理一个对象批次"""
    return [extract_features(obj, target_features) for obj in batch]

# 分批次处理，每批1000个对象
batch_size = 1000
object_batches = [objects[i:i+batch_size] for i in range(0, len(objects), batch_size)]

# 启动多进程池处理
with Pool() as pool:
    batch_results = pool.map(process_object_batch, object_batches)

# 合并所有批次的结果
processed_results = [item for sublist in batch_results for item in sublist]

自定义扩展：不同特征不同默认值 #

如果需要给不同特征设置不同的默认值，可以把特征列表改成{特征名: 默认值}的字典格式：

def extract_features_custom_defaults(obj, feature_default_map):
    extracted = {}
    if isinstance(obj, dict):
        getter = lambda feat, dft: obj.get(feat, dft)
    else:
        getter = lambda feat, dft: getattr(obj, feat, dft)
    
    for feature, default in feature_default_map.items():
        extracted[feature] = getter(feature, default)
    return extracted

# 使用示例
feature_defaults = {
    'user_id': 'unknown',
    'age': 0,
    'signup_date': '1970-01-01',
    'preference': []
}
processed_results = [extract_features_custom_defaults(obj, feature_defaults) for obj in objects]

这个方案的核心优势：

• 完全适配动态特征列表的变动需求，核心提取逻辑不用修改
• 兼容字典和类实例两种对象格式，覆盖大多数场景
• 优雅处理缺失特征的情况，避免程序中断
• 可根据需求扩展日志、并行处理、自定义默认值等功能
• 针对20K量级做了性能优化，保证处理效率

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