重复值处理

重复行检测与删除。

核心代码

def handle_duplicates(
    *,
    df: pd.DataFrame,
    params: dict[str, Any],
    column_descriptions: dict[str, str],
) -> tuple[pd.DataFrame, dict[str, Any], list[str], dict[str, str]]:
    """重复值检测与处理。"""
    subset = normalize_string_list(params.get('subset')) or list(df.columns)
    require_columns(df, subset, label=_('去重变量'))
    action = str(params.get('action') or 'remove').strip().lower()
    keep = str(params.get('keep') or 'first').strip().lower()
    if keep not in {'first', 'last', 'none'}:
        raise DataProcessingValidationError(_('重复值保留策略仅支持 first、last、none'))

    keep_value: str | bool = keep if keep in {'first', 'last'} else False
    duplicate_all_mask = df.duplicated(subset=subset, keep=False)
    duplicate_remove_mask = df.duplicated(subset=subset, keep=keep_value)
    duplicate_rows = int(duplicate_all_mask.sum())

    if action == 'remove':
        result_df = df.loc[~duplicate_remove_mask].reset_index(drop=True)
        details = {
            'action': action,
            'subset': subset,
            'duplicate_rows': duplicate_rows,
            'removed_rows': int(duplicate_remove_mask.sum()),
            'keep': keep,
        }
        warnings = []
        if keep == 'none' and duplicate_rows:
            warnings.append(_('已删除所有重复组中的全部记录'))
        return result_df, details, warnings, column_descriptions

    if action == 'indicator':
        result_df = df.copy()
        new_name = ensure_new_column_absent(result_df, 'duplicate_flag')
        result_df[new_name] = duplicate_all_mask.astype(int)
        details = {
            'action': action,
            'subset': subset,
            'duplicate_rows': duplicate_rows,
            'created_column': new_name,
        }
        return result_df, details, [], column_descriptions

    raise DataProcessingValidationError(_('重复值操作仅支持 remove 或 indicator'))