Python实现跨Excel工作表员工数据自动化比对

1. 项目概述：Python实现跨工作表员工数据比对

在人力资源管理和企业办公自动化场景中，经常需要处理来自不同部门或时间节点的员工数据表。这些表格可能包含入职记录、考勤统计、绩效评估等不同维度的信息。当我们需要快速找出两份数据之间的差异（如新入职/离职人员、信息变更记录）时，手动比对不仅效率低下，而且容易出错。

Python的pandas库配合openpyxl或xlrd等工具包，可以构建一个轻量级的自动化比对解决方案。这个方案能处理以下典型场景：

• 对比两个部门的在岗人员名单
• 核验月度考勤表的变更情况
• 找出培训前后人员技能评估的变化项
• 同步不同系统的员工基础信息

2. 核心工具链选型与配置

2.1 基础环境搭建

推荐使用Python 3.8+版本，通过以下命令安装必需库：

pip install pandas openpyxl xlrd==2.0.1 # 注意xlrd新版已不支持xlsx

2.2 库功能解析

• pandas：提供DataFrame数据结构，支持高效的表合并、差异检测
• openpyxl：处理xlsx格式的读写操作
• xlrd：旧版用于读取xls格式（需锁定2.0.1版本）

注意：若需处理xlsm等宏文件，需额外安装pywin32库

3. 数据加载与预处理

3.1 文件读取最佳实践

import pandas as pd def load_sheet(file_path, sheet_name): # 自动检测文件格式 if file_path.endswith('.xlsx'): return pd.read_excel(file_path, sheet_name=sheet_name, engine='openpyxl') else: return pd.read_excel(file_path, sheet_name=sheet_name) df1 = load_sheet('hr_q1.xlsx', '在职员工') df2 = load_sheet('hr_q2.xlsx', '人员名单')

3.2 数据清洗关键步骤

1. 统一标识字段格式（如工号去空格、大小写转换）

df1['工号'] = df1['工号'].astype(str).str.strip().str.upper() df2['工号'] = df2['工号'].astype(str).str.strip().str.upper()

2. 处理缺失值

df1.fillna({'部门': '未分配'}, inplace=True)

3. 日期字段标准化

df1['入职日期'] = pd.to_datetime(df1['入职日期'], errors='coerce')

4. 核心比对算法实现

4.1 基于集合的快速比对

# 获取工号集合 set1 = set(df1['工号']) set2 = set(df2['工号']) new_employees = list(set2 - set1) # 新增人员 left_employees = list(set1 - set2) # 离职人员

4.2 详细记录比对（基于merge）

merged = pd.merge(df1, df2, on='工号', how='outer', indicator=True) changes = merged[merged['_merge'] == 'both'].copy() # 检测变更字段 for col in ['部门', '职级']: changes[f'{col}_changed'] = changes[f'{col}_x'] != changes[f'{col}_y']

4.3 高性能大数据量处理

当记录超过10万条时：

# 使用dask加速 import dask.dataframe as dd ddf1 = dd.from_pandas(df1, npartitions=4) ddf2 = dd.from_pandas(df2, npartitions=4)

5. 可视化结果输出

5.1 差异报告生成

with pd.ExcelWriter('comparison_result.xlsx') as writer: # 新增人员表 df2[df2['工号'].isin(new_employees)].to_excel( writer, sheet_name='新增人员', index=False) # 变更明细表 changes[changes.filter(like='_changed').any(axis=1)].to_excel( writer, sheet_name='信息变更', index=False)

5.2 自动高亮设置

from openpyxl.styles import PatternFill red_fill = PatternFill(start_color='FFEE1111', end_color='FFEE1111', fill_type='solid') # 获取工作表对象 ws = writer.sheets['信息变更'] for row in ws.iter_rows(min_row=2): for cell in row: if '_changed' in cell.value: cell.fill = red_fill

6. 性能优化技巧

6.1 内存管理

• 分批读取大文件：

chunksize = 10**4 for chunk in pd.read_excel('large_file.xlsx', chunksize=chunksize): process(chunk)

6.2 数据类型优化

dtype_map = { '工号': 'string', '年龄': 'uint8', '薪资': 'float32' } df = pd.read_excel(..., dtype=dtype_map)

6.3 多进程加速

from multiprocessing import Pool def compare_chunk(args): chunk1, chunk2 = args return pd.merge(chunk1, chunk2, on='工号') with Pool(4) as p: results = p.map(compare_chunk, zip(df1_chunks, df2_chunks))

7. 典型问题排查指南

8. 扩展应用场景

8.1 多表联合比对

from functools import reduce dfs = [df1, df2, df3] common_cols = reduce(lambda x,y: x.intersection(y), [set(df.columns) for df in dfs]) result = pd.concat([df[common_cols] for df in dfs], keys=['Q1','Q2','Q3'])

8.2 与数据库联动

import sqlalchemy engine = sqlalchemy.create_engine('postgresql://user:pass@localhost/db') # 将比对结果写入数据库 df_diff.to_sql('employee_changes', engine, if_exists='append')

8.3 自动化邮件报告

import smtplib from email.mime.multipart import MIMEMultipart from email.mime.base import MIMEBase msg = MIMEMultipart() msg['Subject'] = '员工变动周报' with open('comparison_result.xlsx', 'rb') as f: part = MIMEBase('application', 'octet-stream') part.set_payload(f.read()) encoders.encode_base64(part) part.add_header('Content-Disposition', 'attachment', filename='result.xlsx') msg.attach(part) smtp = smtplib.SMTP('smtp.example.com') smtp.sendmail('hr@company.com', 'manager@company.com', msg.as_string())

9. 工程化建议

1. 日志记录标准化

import logging logging.basicConfig( filename='employee_compare.log', level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s' )

2. 配置参数外部化 创建config.ini：

[Files] source1 = data/hr_q1.xlsx source2 = data/hr_q2.xlsx key_column = 工号

3. 异常处理框架

class SheetCompareError(Exception): pass try: df1 = load_sheet(config['source1']) except FileNotFoundError as e: logging.error(f"文件不存在: {e}") raise SheetCompareError("源文件加载失败")

10. 版本迭代记录

v1.1 (2023-08-20)

• 新增对xlsm格式的支持
• 优化大数据处理性能
• 增加自动邮件通知功能

v1.2 (2023-09-05)

• 修复中文编码问题
• 添加多进程处理模式
• 完善日志记录系统

实际部署中发现，当比对字段超过20个时，merge操作会显著变慢。这时可以采用先hash再比对的方法：

df1['hash'] = pd.util.hash_pandas_object(df1[compare_cols]) df2['hash'] = pd.util.hash_pandas_object(df2[compare_cols]) changes = df1.merge(df2, on='工号')[df1['hash'] != df2['hash']]