python实现excel2个表的数据匹配追加到表后面

匹配逻辑： 无论 URL 是 /goto/ 还是 /file/，只要 hash 部分相同就能匹配，适用于不同格式的 URL 变体。

python merge_by_url_hash.py <文件路径> <源表> <源键列> <目标表> <目标键列> [输出路径]

# 示例
python merge_by_url_hash.py /Users/creat/Desktop/aliso.xlsx Sheet1 0 数据 4

#!/usr/bin/env python3
"""
根据 URL 哈希值跨表匹配数据
用法: python merge_by_url_hash.py <xlsx文件> <源表名> <源键列> <目标表名> <目标键列> <追加列数> [输出路径]
"""

import sys
import pandas as pd
from openpyxl import load_workbook
import re


def extract_hash(url: str) -> str:
    """从 URL 中提取末尾文件名（不含扩展名）作为哈希键"""
    if not isinstance(url, str):
        return ''
    # 取最后一个 / 后的部分，去掉 .html/.htm 等后缀
    key = url.rsplit('/', 1)[-1]
    key = re.sub(r'\.(html?|htm|php)$', '', key, flags=re.IGNORECASE)
    return key


def merge_by_url_hash(
    filepath: str,
    src_sheet: str,
    src_key_col: int,      # 0-based 列索引
    src_append_cols: list,  # 要追加的源列索引列表（0-based）
    dst_sheet: str,
    dst_key_col: int,      # 0-based 列索引
    dst_header_row: int = 1,   # 目标表是否有表头行（1=有，None=无）
    out_path: str = None,
    new_col_names: list = None  # 新列名列表
):
    """
    根据两个表的 URL 哈希值匹配，将源表指定列的数据追加到目标表。
    
    参数:
        filepath: Excel 文件路径
        src_sheet: 源表名称
        src_key_col: 源表键列索引（0-based）
        src_append_cols: 源表要追加的列索引列表（0-based）
        dst_sheet: 目标表名称
        dst_key_col: 目标表键列索引（0-based）
        dst_header_row: 目标表表头行号（1=第1行是表头，None=无表头）
        out_path: 输出路径，默认覆盖原文件
        new_col_names: 新增列的列名列表
    """
    # 读取数据
    df_src = pd.read_excel(filepath, sheet_name=src_sheet, header=None)
    df_dst = pd.read_excel(filepath, sheet_name=dst_sheet, header=None)

    # 构建哈希 -> (col1, col2, ...) 的查找字典
    lookup = {}
    for _, row in df_src.iterrows():
        key = extract_hash(row.iloc[src_key_col])
        if key:
            vals = tuple(row.iloc[c] for c in src_append_cols)
            lookup[key] = vals

    # 匹配并追加
    new_cols_data = []
    for _, row in df_dst.iterrows():
        key = extract_hash(row.iloc[dst_key_col])
        if key and key in lookup:
            new_cols_data.append(lookup[key])
        else:
            new_cols_data.append(tuple(None for _ in src_append_cols))

    # 转换为 DataFrame
    n_new = len(src_append_cols)
    new_df = pd.DataFrame(new_cols_data, columns=new_col_names or [f'新增列{i+1}' for i in range(n_new)])

    # 用 openpyxl 写入目标表（保留格式）
    wb = load_workbook(filepath)
    ws = wb[dst_sheet]

    start_col = df_dst.shape[1] + 1  # 追加到最右侧

    # 写表头（如果目标表有表头行）
    if dst_header_row:
        for j, name in enumerate(new_col_names or []):
            ws.cell(row=dst_header_row, column=start_col + j).value = name

    # 写数据（从dst_header_row+1行开始）
    data_start = (dst_header_row + 1) if dst_header_row else 1
    for r_idx, new_vals in enumerate(new_cols_data, start=data_start):
        for j, val in enumerate(new_vals):
            ws.cell(row=r_idx, column=start_col + j).value = val

    # 保存
    save_path = out_path or filepath
    wb.save(save_path)

    # 统计
    matched = sum(1 for row in new_cols_data if any(pd.notna(v) for v in row))
    print(f"✅ 完成")
    print(f"   文件: {save_path}")
    print(f"   表: {dst_sheet}")
    print(f"   匹配行数: {matched}/{len(new_cols_data)}")
    print(f"   新增列数: {n_new} 列")
    print(f"   追加范围: {ws.cell(1, start_col).coordinate}:{ws.cell(len(df_dst)+1, start_col+n_new-1).coordinate}")

    return matched, len(new_cols_data)


# -------------------- 命令行入口 --------------------
if __name__ == '__main__':
    if len(sys.argv) < 6:
        print(__doc__)
        sys.exit(1)

    _, filepath, src_sheet, src_key_col, dst_sheet, dst_key_col = sys.argv[:6]
    src_key_col = int(src_key_col)
    dst_key_col = int(dst_key_col)
    out_path = sys.argv[6] if len(sys.argv) > 6 else None

    # 默认追加第1列和第2列（B、C列）
    src_append_cols = [1, 2]
    new_col_names = ['B列数据', 'C列数据']

    # 示例：aliso.xlsx 中 /goto/<hash> 匹配 /file/<hash>
    # python merge_by_url_hash.py /Users/creat/Desktop/aliso.xlsx Sheet1 0 数据 4
    merge_by_url_hash(
        filepath=filepath,
        src_sheet=src_sheet,
        src_key_col=src_key_col,
        src_append_cols=src_append_cols,
        dst_sheet=dst_sheet,
        dst_key_col=dst_key_col,
        dst_header_row=None,   # 数据表无表头
        out_path=out_path,
        new_col_names=new_col_names
    )

带有返回数据的版本

#!/usr/bin/env python3
"""
根据 URL 哈希值跨表匹配数据
用法: python merge_by_url_hash.py <xlsx文件> <源表名> <源键列> <目标表名> <目标键列> <追加列数> [输出路径]
"""

import sys
import pandas as pd
from openpyxl import load_workbook
import re
from datetime import datetime


def extract_hash(url: str) -> str:
    """从 URL 中提取末尾文件名（不含扩展名）作为哈希键"""
    if not isinstance(url, str):
        return ''
    # 取最后一个 / 后的部分，去掉 .html/.htm 等后缀
    key = url.rsplit('/', 1)[-1]
    key = re.sub(r'\.(html?|htm|php)$', '', key, flags=re.IGNORECASE)
    return key


def merge_by_url_hash(
    filepath: str,
    src_sheet: str,
    src_key_col: int,      # 0-based 列索引
    src_append_cols: list,  # 要追加的源列索引列表（0-based）
    dst_sheet: str,
    dst_key_col: int,      # 0-based 列索引
    dst_header_row: int = 1,   # 目标表是否有表头行（1=有，None=无）
    out_path: str = None,
    new_col_names: list = None  # 新列名列表
):
    """
    根据两个表的 URL 哈希值匹配，将源表指定列的数据追加到目标表。
    
    返回:
        (总行数, 成功数, 失败数, 匹配率)
    """
    print(f"\n{'='*60}")
    print(f"开始执行数据匹配任务")
    print(f"执行时间: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
    print(f"{'='*60}")
    
    # 读取数据
    print(f"\n📖 读取数据...")
    print(f"   源表: {src_sheet}")
    print(f"   目标表: {dst_sheet}")
    
    df_src = pd.read_excel(filepath, sheet_name=src_sheet, header=None)
    df_dst = pd.read_excel(filepath, sheet_name=dst_sheet, header=None)
    
    print(f"   源表行数: {len(df_src)}")
    print(f"   目标表行数: {len(df_dst)}")

    # 构建哈希 -> (col1, col2, ...) 的查找字典
    print(f"\n🔍 构建哈希索引...")
    lookup = {}
    src_empty_count = 0
    for idx, row in df_src.iterrows():
        key = extract_hash(row.iloc[src_key_col])
        if key:
            vals = tuple(row.iloc[c] for c in src_append_cols)
            lookup[key] = vals
        else:
            src_empty_count += 1
    
    print(f"   源表有效哈希数: {len(lookup)}/{len(df_src)}")
    if src_empty_count > 0:
        print(f"   警告: {src_empty_count} 行源数据无有效URL哈希")

    # 匹配并追加
    print(f"\n🔄 执行匹配操作...")
    new_cols_data = []
    matched_count = 0
    unmatched_count = 0
    empty_key_count = 0
    
    for idx, row in df_dst.iterrows():
        key = extract_hash(row.iloc[dst_key_col])
        if key and key in lookup:
            new_cols_data.append(lookup[key])
            matched_count += 1
        elif key:
            new_cols_data.append(tuple(None for _ in src_append_cols))
            unmatched_count += 1
        else:
            new_cols_data.append(tuple(None for _ in src_append_cols))
            empty_key_count += 1

    # 转换为 DataFrame
    n_new = len(src_append_cols)
    new_df = pd.DataFrame(new_cols_data, columns=new_col_names or [f'新增列{i+1}' for i in range(n_new)])

    # 用 openpyxl 写入目标表（保留格式）
    print(f"\n💾 写入数据到文件...")
    wb = load_workbook(filepath)
    ws = wb[dst_sheet]

    start_col = df_dst.shape[1] + 1  # 追加到最右侧

    # 写表头（如果目标表有表头行）
    if dst_header_row:
        for j, name in enumerate(new_col_names or []):
            ws.cell(row=dst_header_row, column=start_col + j).value = name

    # 写数据（从dst_header_row+1行开始）
    data_start = (dst_header_row + 1) if dst_header_row else 1
    for r_idx, new_vals in enumerate(new_cols_data, start=data_start):
        for j, val in enumerate(new_vals):
            ws.cell(row=r_idx, column=start_col + j).value = val

    # 保存
    save_path = out_path or filepath
    wb.save(save_path)
    
    # 计算统计信息
    total_rows = len(new_cols_data)
    success_rate = (matched_count / total_rows * 100) if total_rows > 0 else 0
    
    # 输出详细统计
    print(f"\n{'='*60}")
    print(f"✅ 任务执行完成")
    print(f"{'='*60}")
    print(f"📊 执行统计:")
    print(f"   文件路径: {save_path}")
    print(f"   目标表名: {dst_sheet}")
    print(f"   ─────────────────────────")
    print(f"   总处理行数: {total_rows}")
    print(f"   ✓ 成功匹配: {matched_count} 行")
    print(f"   ✗ 匹配失败: {unmatched_count} 行")
    print(f"   ⚠ 空键跳过: {empty_key_count} 行")
    print(f"   ─────────────────────────")
    print(f"   匹配成功率: {success_rate:.2f}%")
    print(f"   新增列数: {n_new} 列")
    print(f"   新增列名: {', '.join(new_col_names or [f'新增列{i+1}' for i in range(n_new)])}")
    print(f"   ─────────────────────────")
    print(f"   写入范围: {ws.cell(1, start_col).coordinate}:{ws.cell(len(df_dst)+1, start_col+n_new-1).coordinate}")
    print(f"{'='*60}")
    
    # 如果匹配率过低，给出警告
    if success_rate < 50 and total_rows > 0:
        print(f"⚠️  警告: 匹配率低于50%，请检查:")
        print(f"   1. 源表和目标表的URL格式是否一致")
        print(f"   2. 键列索引是否正确 (源键列: {src_key_col}, 目标键列: {dst_key_col})")
        print(f"   3. URL哈希提取规则是否匹配")
    
    return total_rows, matched_count, unmatched_count, success_rate


def main():
    """主函数"""
    try:
        if len(sys.argv) < 6:
            print(__doc__)
            print("\n❌ 错误: 参数不足")
            print("示例: python merge_by_url_hash.py data.xlsx Sheet1 0 Sheet2 4")
            sys.exit(1)

        _, filepath, src_sheet, src_key_col, dst_sheet, dst_key_col = sys.argv[:6]
        src_key_col = int(src_key_col)
        dst_key_col = int(dst_key_col)
        out_path = sys.argv[6] if len(sys.argv) > 6 else None

        # 默认追加第1列和第2列（B、C列）- 0-based索引
        src_append_cols = [1, 2]
        new_col_names = ['B列数据', 'C列数据']
        
        # 如果提供了追加列数参数
        if len(sys.argv) > 7:
            try:
                num_cols = int(sys.argv[7])
                src_append_cols = list(range(1, num_cols + 1))
                new_col_names = [f'第{chr(65+i)}列数据' for i in src_append_cols]
            except:
                pass

        # 执行合并
        total, success, failed, rate = merge_by_url_hash(
            filepath=filepath,
            src_sheet=src_sheet,
            src_key_col=src_key_col,
            src_append_cols=src_append_cols,
            dst_sheet=dst_sheet,
            dst_key_col=dst_key_col,
            dst_header_row=None,   # 数据表无表头
            out_path=out_path,
            new_col_names=new_col_names
        )
        
        # 根据结果设置退出码
        if success == 0 and total > 0:
            print("\n❌ 任务失败: 没有任何数据被匹配")
            sys.exit(1)
        elif rate < 50:
            print("\n⚠️  任务部分完成，但匹配率较低")
            sys.exit(2)
        else:
            print("\n🎉 任务成功完成!")
            sys.exit(0)
            
    except FileNotFoundError as e:
        print(f"\n❌ 错误: 找不到文件 - {e}")
        sys.exit(1)
    except ValueError as e:
        print(f"\n❌ 错误: 参数格式错误 - {e}")
        sys.exit(1)
    except Exception as e:
        print(f"\n❌ 错误: 执行失败 - {e}")
        import traceback
        traceback.print_exc()
        sys.exit(1)


# -------------------- 命令行入口 --------------------
if __name__ == '__main__':
    main()