Python difflib实现列名相似度模糊匹配

from difflib import SequenceMatcher

def calc_column_similarity(col_a: str, col_b: str) -> float:
    """计算两个列名之间的相似度比率 (0.0 ~ 1.0)"""
    return SequenceMatcher(None, col_a.strip().lower(), col_b.strip().lower()).ratio()

# 示例测试
score_1 = calc_column_similarity("cust_name", "customer_name")
score_2 = calc_column_similarity("user_id", "userid")
score_3 = calc_column_similarity("order_date", "total_amount")

print(f"cust_name vs customer_name: {score_1:.4f}")   # 输出约 0.8182
print(f"user_id vs userid:           {score_2:.4f}")   # 输出约 0.9231
print(f"order_date vs total_amount:  {score_3:.4f}")   # 输出约 0.2727

from difflib import get_close_matches

standard_cols = ["user_id", "customer_name", "order_date", "total_amount", "phone_number"]

incoming_cols = ["cust_name", "userid", "order_dt", "tot_amount", "unknown_field"]

for col in incoming_cols:
    matches = get_close_matches(col, standard_cols, n=1, cutoff=0.6)
    matched_target = matches[0] if matches else None
    print(f"原始列: {col:<15} -> 匹配标准列: {matched_target}")

# 示例输出：
# 原始列: cust_name       -> 匹配标准列: customer_name
# 原始列: userid          -> 匹配标准列: user_id
# 原始列: order_dt        -> 匹配标准列: order_date
# 原始列: tot_amount      -> 匹配标准列: total_amount
# 原始列: unknown_field   -> 匹配标准列: None

import pandas as pd
from difflib import SequenceMatcher
from typing import Dict, List, Tuple

def build_column_mapping(
    actual_columns: List[str],
    target_columns: List[str],
    cutoff: float = 0.70
) -> Tuple[Dict[str, str], List[str]]:
    """
    基于 difflib 构建列名模糊匹配字典
    
    参数:
        actual_columns: 原始数据表的实际列名列表
        target_columns: 预期对齐的标准列名列表
        cutoff: 相似度截断阈值，默认 0.70
        
    返回:
        (mapping_dict, unmatched_columns)
        mapping_dict: 用于 df.rename 的映射字典 {原始列名: 标准列名}
        unmatched_columns: 未能成功匹配的原始列名列表（用于审计与报警）
    """
    mapping: Dict[str, str] = {}
    assigned_targets = set()
    unmatched: List[str] = []

    # 预先记录小写标准化列名
    clean_target_map = {col.strip().lower(): col for col in target_columns}
    clean_targets = list(clean_target_map.keys())

    # 按相似度高低进行贪心匹配
    for orig_col in actual_columns:
        clean_orig = orig_col.strip().lower().replace("-", "_")
        
        # 1. 优先完全一致匹配
        if clean_orig in clean_target_map:
            std_col = clean_target_map[clean_orig]
            mapping[orig_col] = std_col
            assigned_targets.add(std_col)
            continue

        # 2. 模糊匹配打分
        best_match = None
        highest_score = 0.0

        for tgt in clean_targets:
            score = SequenceMatcher(None, clean_orig, tgt).ratio()
            if score > highest_score:
                highest_score = score
                best_match = tgt

        # 3. 阈值判断与防重保护
        if highest_score >= cutoff and best_match is not None:
            std_col = clean_target_map[best_match]
            if std_col not in assigned_targets:
                mapping[orig_col] = std_col
                assigned_targets.add(std_col)
            else:
                # 冲突发生，暂记入未分配并输出警告
                print(f"[警告] 列名冲突: '{orig_col}' 匹配到已分配的标准列 '{std_col}' (分值: {highest_score:.3f})")
                unmatched.append(orig_col)
        else:
            unmatched.append(orig_col)

    return mapping, unmatched


# ==================== 场景模拟与运行验证 ====================

if __name__ == "__main__":
    # 统一的标准字段规范 Schema
    TARGET_COLUMNS = ["user_id", "customer_name", "order_date", "total_amount", "contact_phone"]

    # 模拟数据源 A（某分公司导出的异构表格）
    df_branch_a = pd.DataFrame([
        {"userid": "U1001", "cust_name": "张三", "order_dt": "2026-03-01", "tot_amount": 350.5, "phone": "13800000001"},
        {"userid": "U1002", "cust_name": "李四", "order_dt": "2026-03-02", "tot_amount": 120.0, "phone": "13800000002"},
    ])

    # 模拟数据源 B（第三方合作商提供的表格，存在中划线和大写）
    df_branch_b = pd.DataFrame([
        {"User-ID": "U2001", "Customer_Name": "王五", "Order_Date": "2026-03-03", "Total_Amount": 890.0, "ext_memo": "加急"},
    ])

    print("--- 原始数据源 A 列名 ---")
    print(list(df_branch_a.columns))

    # 执行模糊映射计算
    mapping_a, unmatched_a = build_column_mapping(list(df_branch_a.columns), TARGET_COLUMNS, cutoff=0.72)
    print("\n数据源 A 生成的重命名映射字典:")
    print(mapping_a)
    print(f"数据源 A 未识别字段: {unmatched_a}")

    # 执行重命名
    df_a_cleaned = df_branch_a.rename(columns=mapping_a)

    # 对数据源 B 执行相同处理
    mapping_b, unmatched_b = build_column_mapping(list(df_branch_b.columns), TARGET_COLUMNS, cutoff=0.72)
    print("\n数据源 B 生成的重命名映射字典:")
    print(mapping_b)
    print(f"数据源 B 未识别字段: {unmatched_b} (如额外业务字段保留或存入扩展字段)")

    df_b_cleaned = df_branch_b.rename(columns=mapping_b)

    # 最终合并
    unified_df = pd.concat([df_a_cleaned, df_b_cleaned], ignore_index=True)
    
    print("\n--- 最终自动化对齐合并后的统一表格 ---")
    print(unified_df[["user_id", "customer_name", "order_date", "total_amount"]])

--- 原始数据源 A 列名 ---
['userid', 'cust_name', 'order_dt', 'tot_amount', 'phone']

数据源 A 生成的重命名映射字典:
{'userid': 'user_id', 'cust_name': 'customer_name', 'order_dt': 'order_date', 'tot_amount': 'total_amount'}
数据源 A 未识别字段: ['phone']

数据源 B 生成的重命名映射字典:
{'User-ID': 'user_id', 'Customer_Name': 'customer_name', 'Order_Date': 'order_date', 'Total_Amount': 'total_amount'}
数据源 B 未识别字段: ['ext_memo'] (如额外业务字段保留或存入扩展字段)

--- 最终自动化对齐合并后的统一表格 ---
  user_id customer_name  order_date  total_amount
0   U1001            张三  2026-03-01         350.5
1   U1002            李四  2026-03-02         120.0
2   U2001            王五  2026-03-03         890.0

def sort_tokens(col_name: str) -> str:
    """消除词序颠倒影响：例如 'order_status' -> 'order_status'，'status_order' -> 'order_status'"""
    parts = col_name.strip().lower().replace("-", "_").split("_")
    return "_".join(sorted(filter(None, parts)))

# 测试
col_x = sort_tokens("order_status")  # 'order_status'
col_y = sort_tokens("status_order")  # 'order_status'
print(SequenceMatcher(None, col_x, col_y).ratio()) # 输出 1.0 完全匹配！