数据合并

纵向追加、横向绑定与按键合并。

核心代码

def join_dataframes(
    *,
    current_df: pd.DataFrame,
    secondary_df: pd.DataFrame,
    left_keys: list[str] | str | None,
    right_keys: list[str] | str | None,
    how: str = 'left',
    suffix: str = '_right',
) -> tuple[pd.DataFrame, dict[str, Any], list[str]]:
    """按键执行 inner/left/right/outer 合并。"""
    left_keys = _normalize_keys(left_keys)
    right_keys = _normalize_keys(right_keys)
    if len(left_keys) != len(right_keys):
        raise DataProcessingValidationError(_('左右合并键数量必须一致'))

    require_columns(current_df, left_keys, label=_('左侧合并键'))
    require_columns(secondary_df, right_keys, label=_('右侧合并键'))
    if how not in {'inner', 'left', 'right', 'outer'}:
        raise DataProcessingValidationError(_('匹配合并方式仅支持 inner、left、right、outer'))

    secondary_df = secondary_df.copy()
    rename_map = {}
    for left_key, right_key in zip(left_keys, right_keys):
        if left_key != right_key:
            rename_map[right_key] = left_key
    if rename_map:
        secondary_df = secondary_df.rename(columns=rename_map)
    right_keys = [rename_map.get(key, key) for key in right_keys]

    duplicate_left_rows = int(current_df.duplicated(subset=left_keys, keep=False).sum())
    duplicate_right_rows = int(secondary_df.duplicated(subset=left_keys, keep=False).sum())

    secondary_for_merge = secondary_df.copy()
    non_key_columns = [column for column in secondary_for_merge.columns if column not in left_keys]
    renamed_secondary, renamed_columns = resolve_conflicting_columns(
        existing_columns=[str(column) for column in current_df.columns],
        incoming_df=secondary_for_merge[non_key_columns],
        suffix=suffix,
    )
    secondary_for_merge = pd.concat([secondary_for_merge[left_keys], renamed_secondary], axis=1)

    merged_df = current_df.merge(
        secondary_for_merge,
        on=left_keys,
        how=how,
        indicator=True,
    )
    indicator = merged_df['_merge']
    match_counts = indicator.value_counts().to_dict()
    matched_rows = int(match_counts.get('both', 0))
    unmatched_left_rows = int(match_counts.get('left_only', 0))
    unmatched_right_rows = int(match_counts.get('right_only', 0))
    result_df = merged_df.drop(columns=['_merge'])

    base_rows = len(current_df) if how in {'left', 'inner'} else len(result_df)
    match_rate = float(matched_rows / base_rows) if base_rows else 0.0
    details = {
        'merge_mode': 'join',
        'how': how,
        'left_keys': left_keys,
        'right_keys': right_keys,
        'matched_rows': matched_rows,
        'unmatched_left_rows': unmatched_left_rows,
        'unmatched_right_rows': unmatched_right_rows,
        'match_rate': round(match_rate, 4),
        'duplicate_left_rows': duplicate_left_rows,
        'duplicate_right_rows': duplicate_right_rows,
        'renamed_conflicts': renamed_columns,
    }
    warnings = []
    if duplicate_left_rows:
        warnings.append(_('左侧数据存在 %(duplicate_left_rows)s 行重复合并键，结果可能出现一对多扩展') % {'duplicate_left_rows': duplicate_left_rows})
    if duplicate_right_rows:
        warnings.append(_('右侧数据存在 %(duplicate_right_rows)s 行重复合并键，结果可能出现多对一或多对多扩展') % {'duplicate_right_rows': duplicate_right_rows})
    if unmatched_left_rows:
        warnings.append(_('有 %(unmatched_left_rows)s 行左侧记录未匹配成功') % {'unmatched_left_rows': unmatched_left_rows})
    if unmatched_right_rows and how in {'right', 'outer'}:
        warnings.append(_('有 %(unmatched_right_rows)s 行右侧记录未匹配成功') % {'unmatched_right_rows': unmatched_right_rows})
    if renamed_columns:
        warnings.append(_('有 %(count)s 个重名变量已自动追加后缀 %(suffix)s') % {'count': len(renamed_columns), 'suffix': suffix})
    return result_df, details, warnings